update resendMutex and unit test

Author: hlibman-connamara

in_session.go

@@ -231,15 +231,7 @@ func (state inSession) resendMessages(session *session, beginSeqNo, endSeqNo int
 	}
 
 	session.resendMutex.Lock()
-	session.resendRequestActive = true
-	session.resendMutex.Unlock()
-
-	defer func() {
-		session.resendMutex.Lock()
-		session.resendRequestActive = false
-		session.resendMutex.Unlock()
-		session.resendCond.Broadcast()
-	}()
+	defer session.resendMutex.Unlock()
 
 	seqNum := beginSeqNo
 	nextSeqNum := seqNum

in_session_test.go

@@ -388,15 +388,9 @@ func (s *InSessionTestSuite) TestFIXMsgInResendRequestBlocksSend() {
 		Return(nil)
 
 	s.MockApp.On("FromAdmin").Return(nil)
-	go s.fixMsgIn(s.session, s.ResendRequest(1))
+	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() {

quickfix_test.go

@@ -16,7 +16,6 @@
 package quickfix
 
 import (
-	"sync"
 	"time"
 
 	"github.com/stretchr/testify/mock"
@@ -222,7 +221,6 @@ 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

@@ -40,11 +40,8 @@ type session struct {
 	toSend [][]byte
 
 	// Mutex for access to toSend.
-	sendMutex sync.RWMutex
-
-	resendRequestActive bool
-	resendMutex         sync.Mutex
-	resendCond          *sync.Cond
+	sendMutex   sync.Mutex
+	resendMutex sync.RWMutex
 
 	sessionEvent chan internal.Event
 	messageEvent chan bool
@@ -275,8 +272,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.RLock()
-	defer s.sendMutex.RUnlock()
+	s.sendMutex.Lock()
+	defer s.sendMutex.Unlock()
 
 	msgBytes, err := s.prepMessageForSend(msg, nil)
 	if err != nil {
@@ -306,14 +303,11 @@ 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.resendMutex.RLock()
+	defer s.resendMutex.RUnlock()
 
-	s.sendMutex.RLock()
-	defer s.sendMutex.RUnlock()
+	s.sendMutex.Lock()
+	defer s.sendMutex.Unlock()
 
 	msgBytes, err := s.prepMessageForSend(msg, inReplyTo)
 	if err != nil {
@@ -328,8 +322,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.RLock()
-	defer s.sendMutex.RUnlock()
+	s.sendMutex.Lock()
+	defer s.sendMutex.Unlock()
 
 	s.dropQueued()
 	return s.store.Reset()
@@ -340,8 +334,8 @@ func (s *session) dropAndSend(msg *Message) error {
 	return s.dropAndSendInReplyTo(msg, nil)
 }
 func (s *session) dropAndSendInReplyTo(msg *Message, inReplyTo *Message) error {
-	s.sendMutex.RLock()
-	defer s.sendMutex.RUnlock()
+	s.sendMutex.Lock()
+	defer s.sendMutex.Unlock()
 
 	msgBytes, err := s.prepMessageForSend(msg, inReplyTo)
 	if err != nil {
@@ -423,8 +417,8 @@ func (s *session) dropQueued() {
 }
 
 func (s *session) EnqueueBytesAndSend(msg []byte) {
-	s.sendMutex.RLock()
-	defer s.sendMutex.RUnlock()
+	s.sendMutex.Lock()
+	defer s.sendMutex.Unlock()
 
 	s.toSend = append(s.toSend, msg)
 	s.sendQueued(true)

session_factory.go

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