Add message rate limit to initiator and instrument

acceptor.go

@@ -334,10 +334,10 @@ func (a *Acceptor) handleConnection(netConn net.Conn) {
 
 	go func() {
 		msgIn <- fixIn{msgBytes, parser.lastRead}
-		readLoop(parser, msgIn)
+		readLoop(parser, msgIn, session.sessionID.String())
 	}()
 
-	writeLoop(netConn, msgOut, a.globalLog)
+	writeLoop(netConn, msgOut, a.globalLog, 0, session.sessionID.String())
 }
 
 func (a *Acceptor) dynamicSessionsLoop() {

config/configuration.go

@@ -65,4 +65,5 @@ const (
 	RejectInvalidMessage         string = "RejectInvalidMessage"
 	DynamicSessions              string = "DynamicSessions"
 	DynamicQualifier             string = "DynamicQualifier"
+	MaxMessagesPerSecond         string = "MaxMessagesPerSecond"
 )

config/doc.go

@@ -230,6 +230,12 @@ Time between reconnection attempts in seconds. Only used for initiators.    Valu
 
 Defaults to 30
 
+MaxMessagesPerSecond
+
+Maximum number of messages to send per second. Required if the remote party enforces rate limiting for incoming messages. A value of 0 means no rate limiting is enforced. Only used for initiators at present. Value must be 0 or a positive integer.
+
+Defaults to 0.
+
 LogoutTimeout
 
 Session setting for logout timeout in seconds. Only used for initiators. Value must be positive integer.

connection.go

@@ -1,28 +1,85 @@
 package quickfix
 
-import "io"
+import (
+	"context"
+	"io"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"golang.org/x/time/rate"
+)
+
+var (
+	outgoingMessageCount = prometheus.NewCounterVec(prometheus.CounterOpts{
+		Namespace: "quickfix",
+		Name:      "outgoing_message_count",
+		Help:      "Count of messages sent inside quickfix writeLoop function",
+	}, []string{"sessionID"})
+
+	outgoingMessageBytes = prometheus.NewCounterVec(prometheus.CounterOpts{
+		Namespace: "quickfix",
+		Name:      "ougoing_message_bytes",
+		Help:      "Count of bytes sent inside quickfix writeLoop function",
+	}, []string{"sessionID"})
+
+	incomingMessageCount = prometheus.NewCounterVec(prometheus.CounterOpts{
+		Namespace: "quickfix",
+		Name:      "incoming_message_count",
+		Help:      "Count of messages received inside quickfix writeLoop function",
+	}, []string{"sessionID"})
+
+	incomingMessageBytes = prometheus.NewCounterVec(prometheus.CounterOpts{
+		Namespace: "quickfix",
+		Name:      "incoming_message_bytes",
+		Help:      "Count of bytes received inside quickfix readLoop function",
+	}, []string{"sessionID"})
+)
+
+func init() {
+	prometheus.MustRegister(outgoingMessageCount)
+	prometheus.MustRegister(outgoingMessageBytes)
+	prometheus.MustRegister(incomingMessageCount)
+	prometheus.MustRegister(incomingMessageBytes)
+}
+
+func writeLoop(connection io.Writer, messageOut chan []byte, log Log, maxMessagesPerSecond int, sessionID string) {
+	limiter := rate.NewLimiter(rate.Limit(maxMessagesPerSecond), 1)
 
-func writeLoop(connection io.Writer, messageOut chan []byte, log Log) {
 	for {
 		msg, ok := <-messageOut
 		if !ok {
 			return
 		}
 
+		if maxMessagesPerSecond > 0 {
+			err := limiter.Wait(context.Background())
+			if err != nil {
+				log.OnEvent(err.Error())
+				continue
+			}
+		}
+
+		outgoingMessageCount.WithLabelValues(sessionID).Inc()
+		outgoingMessageBytes.WithLabelValues(sessionID).Add(float64(len(msg)))
+
 		if _, err := connection.Write(msg); err != nil {
 			log.OnEvent(err.Error())
 		}
 	}
 }
 
-func readLoop(parser *parser, msgIn chan fixIn) {
+func readLoop(parser *parser, msgIn chan fixIn, sessionID string) {
 	defer close(msgIn)
 
 	for {
+
 		msg, err := parser.ReadMessage()
 		if err != nil {
 			return
 		}
+
+		incomingMessageCount.WithLabelValues(sessionID).Inc()
+		incomingMessageBytes.WithLabelValues(sessionID).Add(float64(msg.Len()))
+
 		msgIn <- fixIn{msg, parser.lastRead}
 	}
 }

connection_internal_test.go

@@ -16,7 +16,7 @@ func TestWriteLoop(t *testing.T) {
 		msgOut <- []byte("test msg 3")
 		close(msgOut)
 	}()
-	writeLoop(writer, msgOut, nullLog{})
+	writeLoop(writer, msgOut, nullLog{}, 0, "sessionID")
 
 	expected := "test msg 1 test msg 2 test msg 3"
 
@@ -30,7 +30,7 @@ func TestReadLoop(t *testing.T) {
 	stream := "hello8=FIX.4.0
9=5
blah
10=103
garbage8=FIX.4.0
9=4
foo
10=103
"
 
 	parser := newParser(strings.NewReader(stream))
-	go readLoop(parser, msgIn)
+	go readLoop(parser, msgIn, "sessionID")
 
 	var tests = []struct {
 		expectedMsg   string

go.mod

@@ -7,13 +7,36 @@ require (
 	github.com/globalsign/mgo v0.0.0-20181015135952-eeefdecb41b8
 	github.com/mattn/go-sqlite3 v1.14.14
 	github.com/pkg/errors v0.9.1
+	github.com/prometheus/client_golang v1.13.0
+	github.com/quickfixgo/enum v0.0.0-20210629025633-9afc8539baba
+	github.com/quickfixgo/field v0.0.0-20171007195410-74cea5ec78c7
+	github.com/quickfixgo/fix40 v0.0.0-20171007200002-cce875b2c2e7
+	github.com/quickfixgo/fix41 v0.0.0-20171007212429-b272ca343ed2
+	github.com/quickfixgo/fix42 v0.0.0-20171007212724-86a4567f1c77
+	github.com/quickfixgo/fix43 v0.0.0-20171007213001-a7ff4f2a2470
+	github.com/quickfixgo/fix44 v0.0.0-20171007213039-f090a1006218
+	github.com/quickfixgo/fixt11 v0.0.0-20171007213433-d9788ca97f5d
+	github.com/quickfixgo/tag v0.0.0-20171007194743-cbb465760521
 	github.com/shopspring/decimal v1.3.1
 	github.com/stretchr/testify v1.8.0
 	golang.org/x/net v0.0.0-20220708220712-1185a9018129
+	golang.org/x/time v0.0.0-20220722155302-e5dcc9cfc0b9
+)
+
+require (
+	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/cespare/xxhash/v2 v2.1.2 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/golang/protobuf v1.5.2 // indirect
 	github.com/kr/text v0.2.0 // indirect
+	github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/prometheus/client_model v0.2.0 // indirect
+	github.com/prometheus/common v0.37.0 // indirect
+	github.com/prometheus/procfs v0.8.0 // indirect
 	github.com/stretchr/objx v0.4.0 // indirect
+	golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a // indirect
+	google.golang.org/protobuf v1.28.1 // indirect
 	gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )