Merge pull request #64 from cbusbey/send_on_close_fix

misc refactoring, actual fix for #59

Author: cbusbey

event_timer.go

@@ -16,6 +16,8 @@ func (t *eventTimer) Reset(timeout time.Duration) (ok bool) {
 		ok = true
 	}
 
-	t.timer = time.AfterFunc(timeout, t.Task)
+	if t.Task != nil {
+		t.timer = time.AfterFunc(timeout, t.Task)
+	}
 	return
 }

in_session.go

@@ -8,6 +8,9 @@ import (
 type inSession struct {
 }
 
+func (state inSession) String() string   { return "In Session" }
+func (state inSession) IsLoggedOn() bool { return true }
+
 func (state inSession) FixMsgIn(session *session, msg Message) (nextState sessionState) {
 	var msgType FIXString
 	if err := msg.Header.GetField(tagMsgType, &msgType); err == nil {
@@ -196,7 +199,7 @@ func (state inSession) processReject(session *session, msg Message, rej MessageR
 	switch TypedError := rej.(type) {
 	case targetTooHigh:
 
-		switch session.currentState.(type) {
+		switch session.sessionState.(type) {
 		default:
 			session.doTargetTooHigh(TypedError)
 		case resendState:

latent_state.go

@@ -2,6 +2,9 @@ package quickfix
 
 type latentState struct{}
 
+func (state latentState) String() string   { return "Latent State" }
+func (state latentState) IsLoggedOn() bool { return false }
+
 func (state latentState) FixMsgIn(session *session, msg Message) (nextState sessionState) {
 	session.log.OnEventf("Invalid Session State: Unexpected Msg %v while in Latent state", msg)
 	return state

logon_state.go

@@ -2,6 +2,9 @@ package quickfix
 
 type logonState struct{}
 
+func (state logonState) String() string { return "Logon State" }
+func (s logonState) IsLoggedOn() bool   { return false }
+
 func (s logonState) FixMsgIn(session *session, msg Message) (nextState sessionState) {
 	var msgType FIXString
 	if err := msg.Header.GetField(tagMsgType, &msgType); err == nil && string(msgType) == "A" {

logout_state.go

@@ -3,6 +3,9 @@ package quickfix
 type logoutState struct {
 }
 
+func (state logoutState) String() string { return "Logout State" }
+func (s logoutState) IsLoggedOn() bool   { return false }
+
 func (state logoutState) FixMsgIn(session *session, msg Message) (nextState sessionState) {
 	return state
 }

registry.go

@@ -41,7 +41,9 @@ func SendToTarget(m Marshaler, sessionID SessionID) error {
 		return err
 	}
 
-	session.send(msg)
+	//NOTE: must queue for send here. otherwise, if not executed in same goroutine as session run loop,
+	//message may be sent on closed channel or sent outside of valid state
+	session.queueForSend(msg)
 
 	return nil
 }

resend_state.go

@@ -8,8 +8,7 @@ func (state resendState) FixMsgIn(session *session, msg Message) (nextState sess
 	for ok := true; ok; {
 		nextState = state.inSession.FixMsgIn(session, msg)
 
-		switch nextState.(type) {
-		case logoutState, latentState, resendState:
+		if !nextState.IsLoggedOn() {
 			return
 		}
 

session.go

@@ -5,6 +5,7 @@ import (
 	"github.com/quickfixgo/quickfix/config"
 	"github.com/quickfixgo/quickfix/datadictionary"
 	"github.com/quickfixgo/quickfix/enum"
+	"sync"
 	"time"
 )
 
@@ -16,11 +17,15 @@ type session struct {
 	sessionID SessionID
 
 	messageOut chan []byte
-	toSend     chan Message
 
-	sessionEvent            chan event
-	application             Application
-	currentState            sessionState
+	//application messages are queued up to be sent during the run loop.
+	toSend []Message
+	//mutex for access to toSend
+	sendMutex sync.Mutex
+
+	sessionEvent chan event
+	application  Application
+	sessionState
 	stateTimer              eventTimer
 	peerTimer               eventTimer
 	messageStash            map[int]Message
@@ -95,12 +100,10 @@ func createSession(sessionID SessionID, storeFactory MessageStoreFactory, settin
 		return err
 	}
 
-	session.toSend = make(chan Message)
 	session.sessionEvent = make(chan event)
 	session.application = application
 	session.stateTimer = eventTimer{Task: func() { session.sessionEvent <- needHeartbeat }}
 	session.peerTimer = eventTimer{Task: func() { session.sessionEvent <- peerTimeout }}
-
 	application.OnCreate(session.sessionID)
 	sessions.newSession <- session
 
@@ -109,7 +112,7 @@ func createSession(sessionID SessionID, storeFactory MessageStoreFactory, settin
 
 //kicks off session as an initiator
 func (s *session) initiate(msgIn chan fixIn, msgOut chan []byte, quit chan bool) {
-	s.currentState = logonState{}
+	s.sessionState = logonState{}
 	s.messageStash = make(map[int]Message)
 	s.initiateLogon = true
 
@@ -118,7 +121,7 @@ func (s *session) initiate(msgIn chan fixIn, msgOut chan []byte, quit chan bool)
 
 //kicks off session as an acceptor
 func (s *session) accept(msgIn chan fixIn, msgOut chan []byte, quit chan bool) {
-	s.currentState = logonState{}
+	s.sessionState = logonState{}
 	s.messageStash = make(map[int]Message)
 
 	s.run(msgIn, msgOut, quit)
@@ -161,6 +164,30 @@ func (s *session) resend(msg Message) {
 	s.sendBytes(msg.rawMessage)
 }
 
+//queueForSend will queue up a message to be sent by the session during the next iteration of the run loop
+func (s *session) queueForSend(msg Message) {
+	s.sendMutex.Lock()
+	defer s.sendMutex.Unlock()
+	s.toSend = append(s.toSend, msg)
+}
+
+//sends queued messages if session is logged on
+func (s *session) sendQueued() {
+	if !s.IsLoggedOn() {
+		return
+	}
+
+	s.sendMutex.Lock()
+	defer s.sendMutex.Unlock()
+
+	for _, msg := range s.toSend {
+		s.send(msg)
+	}
+
+	s.toSend = s.toSend[:0]
+}
+
+//send should NOT be called outside of the run loop
 func (s *session) send(msg Message) {
 	s.fillDefaultHeader(msg)
 
@@ -183,7 +210,6 @@ func (s *session) send(msg Message) {
 }
 
 func (s *session) sendBytes(msg []byte) {
-
 	s.log.OnOutgoing(string(msg))
 	s.messageOut <- msg
 	s.stateTimer.Reset(time.Duration(s.heartBeatTimeout))
@@ -469,13 +495,13 @@ type fixIn struct {
 
 func (s *session) run(msgIn chan fixIn, msgOut chan []byte, quit chan bool) {
 	s.messageOut = msgOut
+
 	defer func() {
 		close(s.messageOut)
 		s.onDisconnect()
 	}()
 
 	if s.initiateLogon {
-
 		if s.resetOnLogon {
 			s.store.Reset()
 		}
@@ -500,11 +526,13 @@ func (s *session) run(msgIn chan fixIn, msgOut chan []byte, quit chan bool) {
 
 	for {
 
-		switch s.currentState.(type) {
+		switch s.sessionState.(type) {
 		case latentState:
 			return
 		}
 
+		s.sendQueued()
+
 		select {
 		case fixIn, ok := <-msgIn:
 			if ok {
@@ -513,21 +541,18 @@ 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
-					s.currentState = s.currentState.FixMsgIn(s, msg)
+					s.sessionState = s.FixMsgIn(s, msg)
 				}
 			} else {
 				return
 			}
 			s.peerTimer.Reset(time.Duration(int64(1.2 * float64(s.heartBeatTimeout))))
 
-		case msg := <-s.toSend:
-			s.send(msg)
-
 		case <-quit:
 			return
 
 		case evt := <-s.sessionEvent:
-			s.currentState = s.currentState.Timeout(s, evt)
+			s.sessionState = s.Timeout(s, evt)
 		}
 	}
 }

session_state.go

@@ -1,6 +1,20 @@
 package quickfix
 
+import "fmt"
+
+//sessionState is the current state of the session state machine. The session state determines how the session responds to
+//incoming messages, timeouts, and requests to send application messages.
 type sessionState interface {
+	//FixMsgIn is called by the session on incoming messages from the counter party.  The return type is the next session state
+	//following message processing
 	FixMsgIn(*session, Message) (nextState sessionState)
+
+	//Timeout is called by the session on a timeout event.
 	Timeout(*session, event) (nextState sessionState)
+
+	//IsLoggedOn returns true if state is logged on an in session, false otherwise
+	IsLoggedOn() bool
+
+	//debugging convenience
+	fmt.Stringer
 }

session_state_internal_test.go

@@ -0,0 +1,26 @@
+package quickfix
+
+import "testing"
+
+func TestSessionState_IsLoggedOn(t *testing.T) {
+
+	var tests = []struct {
+		sessionState
+		expectIsLoggedOn bool
+	}{
+		{latentState{}, false},
+		{logonState{}, false},
+		{logoutState{}, false},
+		{inSession{}, true},
+		{resendState{}, true},
+	}
+
+	for _, test := range tests {
+		switch {
+		case test.expectIsLoggedOn && !test.IsLoggedOn():
+			t.Errorf("'%v' should be LoggedOn", test.sessionState)
+		case !test.expectIsLoggedOn && test.IsLoggedOn():
+			t.Errorf("'%v' should not be LoggedOn", test.sessionState)
+		}
+	}
+}