block messages during resend request

Author: hlibman-connamara

in_session.go

@@ -230,6 +230,17 @@ func (state inSession) resendMessages(session *session, beginSeqNo, endSeqNo int
 		return state.generateSequenceReset(session, beginSeqNo, endSeqNo+1, inReplyTo)
 	}
 
+	session.resendMutex.Lock()
+	session.resendRequestActive = true
+	session.resendMutex.Unlock()
+
+	defer func() {
+		session.resendMutex.Lock()
+		session.resendRequestActive = false
+		session.resendMutex.Unlock()
+		session.resendCond.Broadcast()
+	}()
+
 	seqNum := beginSeqNo
 	nextSeqNum := seqNum
 	msg := NewMessage()

in_session_test.go

@@ -373,6 +373,42 @@ func (s *InSessionTestSuite) TestFIXMsgInResendRequestDoNotSendApp() {
 	s.State(inSession{})
 }
 
+func (s *InSessionTestSuite) TestSendBlockedWhenResendRequestActive() {
+	s.MockApp.On("ToApp").Return(nil)
+
+	s.session.resendMutex.Lock()
+	s.session.resendRequestActive = true
+	s.session.resendMutex.Unlock()
+
+	sendCompleted := make(chan struct{})
+	go func() {
+		err := s.session.send(s.NewOrderSingle())
+		s.Require().NoError(err)
+		close(sendCompleted)
+	}()
+
+	select {
+	case <-sendCompleted:
+		s.Fail("send should be blocked during active resend")
+	case <-time.After(50 * time.Millisecond):
+		s.MockApp.AssertNumberOfCalls(s.T(), "ToApp", 0)
+	}
+
+	s.session.resendMutex.Lock()
+	s.session.resendRequestActive = false
+	s.session.resendMutex.Unlock()
+	s.session.resendCond.Broadcast()
+
+	select {
+	case <-sendCompleted:
+		s.LastToAppMessageSent()
+		s.MessageType("D", s.MockApp.lastToApp)
+		s.MockApp.AssertNumberOfCalls(s.T(), "ToApp", 1)
+	case <-time.After(100 * time.Millisecond):
+		s.Fail("send did not proceed after resend was cleared")
+	}
+}
+
 func (s *InSessionTestSuite) TestFIXMsgInTargetTooLow() {
 	s.IncrNextTargetMsgSeqNum()
 

quickfix_test.go

@@ -16,6 +16,7 @@
 package quickfix
 
 import (
+	"sync"
 	"time"
 
 	"github.com/stretchr/testify/mock"
@@ -221,6 +222,7 @@ func (s *SessionSuiteRig) Init() {
 		messageOut:   s.Receiver.sendChannel,
 		sessionEvent: make(chan internal.Event),
 	}
+	s.session.resendCond = sync.NewCond(&s.resendMutex)
 	s.MaxLatency = 120 * time.Second
 }
 

session.go

@@ -42,6 +42,10 @@ type session struct {
 	// Mutex for access to toSend.
 	sendMutex sync.Mutex
 
+	resendRequestActive bool
+	resendMutex         sync.Mutex
+	resendCond          *sync.Cond
+
 	sessionEvent chan internal.Event
 	messageEvent chan bool
 	application  Application
@@ -302,6 +306,12 @@ func (s *session) sendInReplyTo(msg *Message, inReplyTo *Message) error {
 		return s.queueForSend(msg)
 	}
 
+	s.resendMutex.Lock()
+	for s.resendRequestActive {
+		s.resendCond.Wait()
+	}
+	s.resendMutex.Unlock()
+
 	s.sendMutex.Lock()
 	defer s.sendMutex.Unlock()
 

session_factory.go

@@ -89,6 +89,7 @@ func (f sessionFactory) newSession(
 		sessionID: sessionID,
 		stopOnce:  sync.Once{},
 	}
+	s.resendCond = sync.NewCond(&s.resendMutex)
 
 	var validatorSettings = defaultValidatorSettings
 	if settings.HasSetting(config.ValidateFieldsOutOfOrder) {