add resend mutex and unit test for blocking sends during resend request

Author: hlibman-connamara

in_session_test.go

@@ -19,6 +19,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/suite"
 
 	"github.com/quickfixgo/quickfix/internal"
@@ -373,40 +374,29 @@ func (s *InSessionTestSuite) TestFIXMsgInResendRequestDoNotSendApp() {
 	s.State(inSession{})
 }
 
-func (s *InSessionTestSuite) TestSendBlockedWhenResendRequestActive() {
+func (s *InSessionTestSuite) TestFIXMsgInResendRequestBlocksSend() {
 	s.MockApp.On("ToApp").Return(nil)
+	s.Require().Nil(s.session.send(s.NewOrderSingle()))
+	s.LastToAppMessageSent()
+	s.MockApp.AssertNumberOfCalls(s.T(), "ToApp", 1)
+	s.NextSenderMsgSeqNum(2)
 
-	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.MockStore.On("IterateMessages", mock.Anything, mock.Anything, mock.AnythingOfType("func([]byte) error")).
+		Run(func(args mock.Arguments) {
+			s.Require().Nil(s.session.send(s.NewOrderSingle()))
+		}).
+		Return(nil)
 
-	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")
-	}
+	s.MockApp.On("FromAdmin").Return(nil)
+	go s.fixMsgIn(s.session, s.ResendRequest(1))
+
+	s.MockApp.AssertNumberOfCalls(s.T(), "ToApp", 1)
+	s.NextSenderMsgSeqNum(2)
+
+	s.Require().Nil(s.session.send(s.NewOrderSingle()))
+	s.LastToAppMessageSent()
+	s.MockApp.AssertNumberOfCalls(s.T(), "ToApp", 2)
+	s.NextSenderMsgSeqNum(3)
 }
 
 func (s *InSessionTestSuite) TestFIXMsgInTargetTooLow() {

session.go

@@ -40,7 +40,7 @@ type session struct {
 	toSend [][]byte
 
 	// Mutex for access to toSend.
-	sendMutex sync.Mutex
+	sendMutex sync.RWMutex
 
 	resendRequestActive bool
 	resendMutex         sync.Mutex
@@ -275,8 +275,8 @@ func (s *session) resend(msg *Message) bool {
 
 // queueForSend will validate, persist, and queue the message for send.
 func (s *session) queueForSend(msg *Message) error {
-	s.sendMutex.Lock()
-	defer s.sendMutex.Unlock()
+	s.sendMutex.RLock()
+	defer s.sendMutex.RUnlock()
 
 	msgBytes, err := s.prepMessageForSend(msg, nil)
 	if err != nil {
@@ -312,8 +312,8 @@ func (s *session) sendInReplyTo(msg *Message, inReplyTo *Message) error {
 	}
 	s.resendMutex.Unlock()
 
-	s.sendMutex.Lock()
-	defer s.sendMutex.Unlock()
+	s.sendMutex.RLock()
+	defer s.sendMutex.RUnlock()
 
 	msgBytes, err := s.prepMessageForSend(msg, inReplyTo)
 	if err != nil {
@@ -328,8 +328,8 @@ func (s *session) sendInReplyTo(msg *Message, inReplyTo *Message) error {
 
 // dropAndReset will drop the send queue and reset the message store.
 func (s *session) dropAndReset() error {
-	s.sendMutex.Lock()
-	defer s.sendMutex.Unlock()
+	s.sendMutex.RLock()
+	defer s.sendMutex.RUnlock()
 
 	s.dropQueued()
 	return s.store.Reset()
@@ -340,8 +340,8 @@ func (s *session) dropAndSend(msg *Message) error {
 	return s.dropAndSendInReplyTo(msg, nil)
 }
 func (s *session) dropAndSendInReplyTo(msg *Message, inReplyTo *Message) error {
-	s.sendMutex.Lock()
-	defer s.sendMutex.Unlock()
+	s.sendMutex.RLock()
+	defer s.sendMutex.RUnlock()
 
 	msgBytes, err := s.prepMessageForSend(msg, inReplyTo)
 	if err != nil {
@@ -423,8 +423,8 @@ func (s *session) dropQueued() {
 }
 
 func (s *session) EnqueueBytesAndSend(msg []byte) {
-	s.sendMutex.Lock()
-	defer s.sendMutex.Unlock()
+	s.sendMutex.RLock()
+	defer s.sendMutex.RUnlock()
 
 	s.toSend = append(s.toSend, msg)
 	s.sendQueued(true)