Make Initiator`s inbound channel buffered (#22)

We have observed slow consumption of FIX messages in the Initiator.

Since the outbound channel is buffered, the event loop is sending too
many messages at once but reads incoming messages one by one. This
creates an inconsistency between incoming consumption messages and
outgoing production messages.

In this PR we resolve this issue by making the inbound channel buffered.
So the Initiator can now read and write in a truly asynchronous way.

This change is feature flagged. Can be used by setting
`InitiatorInChanCapacity` with an integer representing the capacity of
the inbound channel to the session settings config.

Author: AlexandrosKyriakakis

acceptor.go

@@ -359,7 +359,7 @@ func (a *Acceptor) handleConnection(netConn net.Conn) {
 	}
 
 	a.sessionAddr.Store(sessID, netConn.RemoteAddr())
-	msgIn := make(chan fixIn)
+	msgIn := make(chan fixIn, session.InChanCapacity)
 	msgOut := make(chan []byte)
 
 	if err := session.connect(msgIn, msgOut); err != nil {

config/configuration.go

@@ -489,6 +489,16 @@ const (
 	// Valid Values:
 	//  - Any positive integer
 	MaxLatency string = "MaxLatency"
+
+	// InChanCapacity sets the maximum number of messages that can be buffered in the channel for incoming FIX messages.
+	//
+	// Required: No
+	//
+	// Default: 1
+	//
+	// Valid Values:
+	//  - A positive integer, or zero for an unbuffered channel
+	InChanCapacity string = "InChanCapacity"
 )
 
 const (

initiator.go

@@ -205,7 +205,7 @@ func (i *Initiator) handleConnection(session *session, tlsConfig *tls.Config, di
 			netConn = tlsConn
 		}
 
-		msgIn = make(chan fixIn)
+		msgIn = make(chan fixIn, session.InChanCapacity)
 		msgOut = make(chan []byte)
 		if err := session.connect(msgIn, msgOut); err != nil {
 			session.log.OnEventf("Failed to initiate: %v", err)

internal/session_settings.go

@@ -21,6 +21,7 @@ type SessionSettings struct {
 	TimeZone                     *time.Location
 	ResetSeqTime                 time.Time
 	EnableResetSeqTime           bool
+	InChanCapacity               int
 
 	// Required on logon for FIX.T.1 messages.
 	DefaultApplVerID string

session.go

@@ -757,6 +757,21 @@ func (s *session) checkBeginString(msg *Message) MessageRejectError {
 	return nil
 }
 
+func (s *session) drainMessageIn() {
+	s.log.OnEventf("Draining %d messages from inbound channel...", len(s.messageIn))
+	for {
+		select {
+		case fixInc, ok := <-s.messageIn:
+			if !ok {
+				return
+			}
+			s.Incoming(s, fixInc)
+		default:
+			return
+		}
+	}
+}
+
 func (s *session) doReject(msg *Message, rej MessageRejectError) error {
 	reply := msg.reverseRoute()
 
@@ -824,6 +839,9 @@ func (s *session) onDisconnect() {
 		s.messageOut = nil
 	}
 
+	// s.messageIn is buffered so we need to drain it before disconnection
+	s.drainMessageIn()
+
 	s.messageIn = nil
 }
 

session_factory.go

@@ -431,6 +431,18 @@ func (f sessionFactory) newSession(
 		s.DisableMessagePersist = !persistMessages
 	}
 
+	if settings.HasSetting(config.InChanCapacity) {
+		if s.InChanCapacity, err = settings.IntSetting(config.InChanCapacity); err != nil {
+			return
+		} else if s.InChanCapacity < 0 {
+			err = IncorrectFormatForSetting{Setting: config.InChanCapacity, Value: []byte(strconv.Itoa(s.InChanCapacity))}
+			return
+		}
+	} else {
+		// Default to 1 buffered message per channel
+		s.InChanCapacity = 1
+	}
+
 	if f.BuildInitiators {
 		if err = f.buildInitiatorSettings(s, settings); err != nil {
 			return