dedicated resend channel during resend state

Author: bhaan

connection.go

@@ -116,9 +116,7 @@ func writeLoop(connection io.Writer, messageOut chan []byte) {
 }
 
 func readLoop(parser *parser, msgIn chan fixIn) {
-	defer func() {
-		close(msgIn)
-	}()
+	defer close(msgIn)
 
 	for {
 		msg, err := parser.ReadMessage()

resend_state.go

@@ -22,10 +22,7 @@ func (state resendState) handleNextState(session *session, nextState sessionStat
 	targetSeqNum := session.store.NextTargetMsgSeqNum()
 	if msg, ok := session.messageStash[targetSeqNum]; ok {
 		delete(session.messageStash, targetSeqNum)
-
-		// FIXME add a "resend" channel to the session loop to differentiate between
-		// new incoming fix messages, and stashed messages from the resend state
-		go func() { session.messageIn <- fixIn{msg.rawMessage, msg.ReceiveTime} }()
+		session.resendIn <- msg
 	}
 
 	return resendState{}

session.go

@@ -19,6 +19,7 @@ type session struct {
 	messageOut chan []byte
 	messageIn  chan fixIn
 	toSend     chan sendRequest
+	resendIn   chan Message
 
 	sessionEvent chan event
 	messageEvent chan bool
@@ -495,6 +496,7 @@ func (s *session) run(msgIn chan fixIn, msgOut chan []byte, quit chan bool) {
 	s.messageIn = msgIn
 	s.messageOut = msgOut
 	s.toSend = make(chan sendRequest)
+	s.resendIn = make(chan Message, 1)
 
 	type fromCallback struct {
 		msg Message
@@ -505,7 +507,6 @@ func (s *session) run(msgIn chan fixIn, msgOut chan []byte, quit chan bool) {
 	defer func() {
 		close(s.messageOut)
 		close(s.toSend)
-		close(fromCallbackCh)
 		s.onDisconnect()
 	}()
 
@@ -532,6 +533,19 @@ func (s *session) run(msgIn chan fixIn, msgOut chan []byte, quit chan bool) {
 		s.send(logon)
 	}
 
+	fixMsgIn := func(msg Message) {
+		if rej := s.sessionState.VerifyMsgIn(s, msg); rej != nil {
+			s.sessionState = s.sessionState.FixMsgInRej(s, msg, rej)
+		} else {
+			// "turn off" incoming fix messages until the call
+			// to FromAdmin/App returns
+			msgIn = nil
+			go func() {
+				fromCallbackCh <- fromCallback{msg, s.fromCallback(msg)}
+			}()
+		}
+	}
+
 	for {
 
 		switch s.sessionState.(type) {
@@ -553,21 +567,14 @@ func (s *session) run(msgIn chan fixIn, msgOut chan []byte, quit chan bool) {
 					s.log.OnEventf("Msg Parse Error: %v, %q", err.Error(), fixIn.bytes)
 				} else {
 					msg.ReceiveTime = fixIn.receiveTime
-					if rej := s.sessionState.VerifyMsgIn(s, msg); rej != nil {
-						s.sessionState = s.sessionState.FixMsgInRej(s, msg, rej)
-					} else {
-						// "turn off" incoming fix messages until the call
-						// to FromAdmin/App returns
-						msgIn = nil
-						go func() {
-							fromCallbackCh <- fromCallback{msg, s.fromCallback(msg)}
-						}()
-					}
+					fixMsgIn(msg)
 				}
 			} else {
 				return
 			}
 			s.peerTimer.Reset(time.Duration(int64(1.2 * float64(s.heartBeatTimeout))))
+		case msg := <-s.resendIn:
+			fixMsgIn(msg)
 		case callback := <-fromCallbackCh:
 			// "turn on" incoming fix message now that
 			// FromAdmin/App has completed