Merge pull request #185 from cbusbey/donotsend_fix

fixes donotsend logic

Author: cbusbey

session.go

@@ -218,12 +218,12 @@ func (s *session) resend(msg Message) error {
 }
 
 //queueForSend will validate, persist, and queue the message for send
-func (s *session) queueForSend(msg Message) (err error) {
+func (s *session) queueForSend(msg Message) error {
 	s.sendMutex.Lock()
 	defer s.sendMutex.Unlock()
 
-	if err = s.prepMessageForSend(&msg); err != nil {
-		return
+	if doNotSend, err := s.prepMessageForSend(&msg); err != nil || doNotSend {
+		return err
 	}
 
 	s.toSend = append(s.toSend, msg)
@@ -233,69 +233,74 @@ func (s *session) queueForSend(msg Message) (err error) {
 	default:
 	}
 
-	return
+	return nil
 }
 
 //send will validate, persist, queue the message and send all messages in the queue
-func (s *session) send(msg Message) (err error) {
+func (s *session) send(msg Message) error {
 	s.sendMutex.Lock()
 	defer s.sendMutex.Unlock()
 
-	if err = s.prepMessageForSend(&msg); err != nil {
-		return
+	if doNotSend, err := s.prepMessageForSend(&msg); err != nil || doNotSend {
+		return err
 	}
 
 	s.toSend = append(s.toSend, msg)
 	s.sendQueued()
 
-	return
+	return nil
 }
 
 //dropAndSend will optionally reset the store, validate and persist the message, then drops the send queue and sends the message.
-func (s *session) dropAndSend(msg Message, resetStore bool) (err error) {
+func (s *session) dropAndSend(msg Message, resetStore bool) error {
 
 	s.sendMutex.Lock()
 	defer s.sendMutex.Unlock()
 
 	if resetStore {
-		if err = s.store.Reset(); err != nil {
+		if err := s.store.Reset(); err != nil {
 			return err
 		}
 	}
 
-	if err = s.prepMessageForSend(&msg); err != nil {
-		return
+	if doNotSend, err := s.prepMessageForSend(&msg); err != nil || doNotSend {
+		return err
 	}
 
 	s.dropQueued()
 	s.toSend = append(s.toSend, msg)
 	s.sendQueued()
 
-	return
+	return nil
 }
 
-func (s *session) prepMessageForSend(msg *Message) (err error) {
+func (s *session) prepMessageForSend(msg *Message) (doNotSend bool, err error) {
 	s.fillDefaultHeader(*msg)
 	seqNum := s.store.NextSenderMsgSeqNum()
 	msg.Header.SetField(tagMsgSeqNum, FIXInt(seqNum))
 
 	var msgType FIXString
 	if err = msg.Header.GetField(tagMsgType, &msgType); err != nil {
-		return err
+		return
 	}
 
 	if isAdminMessageType(string(msgType)) {
 		s.application.ToAdmin(*msg, s.sessionID)
 	} else {
-		s.application.ToApp(*msg, s.sessionID)
+		if doNotSendErr := s.application.ToApp(*msg, s.sessionID); doNotSendErr != nil {
+			s.log.OnEventf("Do Not Send: %v", doNotSendErr)
+			doNotSend = true
+			return
+		}
 	}
 
 	var msgBytes []byte
 	if msgBytes, err = msg.Build(); err != nil {
 		return
 	}
 
-	return s.persist(seqNum, msgBytes)
+	err = s.persist(seqNum, msgBytes)
+	return
 }
 
 func (s *session) persist(seqNum int, msgBytes []byte) error {
@@ -492,7 +497,11 @@ func (s *session) verifySelect(msg Message, checkTooHigh bool, checkTooLow bool)
 
 func (s *session) fromCallback(msg Message) MessageRejectError {
 	var msgType FIXString
-	if msg.Header.GetField(tagMsgType, &msgType); isAdminMessageType(string(msgType)) {
+	if err := msg.Header.GetField(tagMsgType, &msgType); err != nil {
+		return err
+	}
+
+	if isAdminMessageType(string(msgType)) {
 		return s.application.FromAdmin(msg, s.sessionID)
 	}
 

session_test.go

@@ -1,6 +1,7 @@
 package quickfix
 
 import (
+	"fmt"
 	"testing"
 	"time"
 
@@ -252,6 +253,7 @@ func TestSession_CheckTargetTooLow(t *testing.T) {
 }
 
 type TestClient struct {
+	rejectToApp   bool
 	adminMessages []Message
 	appMessages   []Message
 }
@@ -275,6 +277,11 @@ func (e *TestClient) ToAdmin(msg Message, sessionID SessionID) {
 
 func (e *TestClient) ToApp(msg Message, sessionID SessionID) (err error) {
 	e.appMessages = append(e.appMessages, msg)
+
+	if e.rejectToApp {
+		return fmt.Errorf("Rejecting ToApp")
+	}
+
 	return nil
 }
 
@@ -331,6 +338,17 @@ func (suite *SessionSendTestSuite) Logon() Message {
 	return msg
 }
 
+func (suite *SessionSendTestSuite) rejectToApp() {
+	suite.TestClient.rejectToApp = true
+}
+
+func (suite *SessionSendTestSuite) shouldNotPersistMessage() {
+	suite.Equal(1, suite.store.NextSenderMsgSeqNum(), "The next sender sequence number should not be incremented")
+	persistedMessages, err := suite.store.GetMessages(1, 1)
+	suite.Nil(err)
+	suite.Len(persistedMessages, 0, "The message should not be persisted")
+}
+
 func (suite *SessionSendTestSuite) shouldPersistMessage() {
 	suite.Equal(2, suite.store.NextSenderMsgSeqNum(), "The next sender sequence number should be incremented")
 	persistedMessages, err := suite.store.GetMessages(1, 1)
@@ -399,6 +417,18 @@ func (suite *SessionSendTestSuite) TestQueueForSendAppMessage() {
 	suite.shouldNotSendMessage()
 }
 
+func (suite *SessionSendTestSuite) TestQueueForSendDoNotSendAppMessage() {
+	suite.rejectToApp()
+	require.Nil(suite.T(), suite.queueForSend(suite.NewOrderSingle()))
+
+	suite.shouldCallToApp()
+	suite.shouldNotPersistMessage()
+	suite.shouldNotSendMessage()
+
+	require.Nil(suite.T(), suite.send(suite.Heartbeat()))
+	suite.shouldSendMessages(1)
+}
+
 func (suite *SessionSendTestSuite) TestQueueForSendAdminMessage() {
 	require.Nil(suite.T(), suite.queueForSend(suite.Heartbeat()))
 
@@ -415,6 +445,15 @@ func (suite *SessionSendTestSuite) TestSendAppMessage() {
 	suite.shouldSendMessage()
 }
 
+func (suite *SessionSendTestSuite) TestSendAppDoNotSendMessage() {
+	suite.rejectToApp()
+	require.Nil(suite.T(), suite.send(suite.NewOrderSingle()))
+
+	suite.shouldCallToApp()
+	suite.shouldNotPersistMessage()
+	suite.shouldNotSendMessage()
+}
+
 func (suite *SessionSendTestSuite) TestSendAdminMessage() {
 	require.Nil(suite.T(), suite.send(suite.Heartbeat()))