synchronize msg sends with session loop (wip)

Author: bhaan

registry.go

@@ -41,11 +41,9 @@ func SendToTarget(m Messagable, sessionID SessionID) error {
 		return err
 	}
 
-	//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
+	request := sendRequest{msg, make(chan error)}
+	session.toSend <- request
+	return <-request.err
 }
 
 type sessionActivate struct {

session.go

@@ -2,7 +2,6 @@ package quickfix
 
 import (
 	"fmt"
-	"sync"
 	"time"
 
 	"github.com/quickfixgo/quickfix/config"
@@ -18,11 +17,8 @@ type session struct {
 	sessionID SessionID
 
 	messageOut chan []byte
-
-	//application messages are queued up to be sent during the run loop.
-	toSend []Message
-	//mutex for access to toSend
-	sendMutex sync.Mutex
+	messageIn  chan fixIn
+	toSend     chan sendRequest
 
 	sessionEvent chan event
 	messageEvent chan bool
@@ -189,35 +185,8 @@ 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)
-	select {
-	case s.messageEvent <- true:
-	default:
-	}
-}
-
-//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) {
+func (s *session) send(msg Message) error {
 	s.fillDefaultHeader(msg)
 
 	seqNum := s.store.NextSenderMsgSeqNum()
@@ -230,12 +199,13 @@ func (s *session) send(msg Message) {
 		s.application.ToApp(msg, s.sessionID)
 	}
 	if msgBytes, err := msg.Build(); err != nil {
-		panic(err)
+		return err
 	} else {
 		s.store.SaveMessage(seqNum, msgBytes)
 		s.sendBytes(msgBytes)
 		s.store.IncrNextSenderMsgSeqNum()
 	}
+	return nil
 }
 
 func (s *session) sendBytes(msg []byte) {
@@ -511,7 +481,13 @@ type fixIn struct {
 	receiveTime time.Time
 }
 
+type sendRequest struct {
+	msg Message
+	err chan error
+}
+
 func (s *session) run(msgIn chan fixIn, msgOut chan []byte, quit chan bool) {
+	s.messageIn = msgIn
 	s.messageOut = msgOut
 
 	defer func() {
@@ -549,9 +525,13 @@ func (s *session) run(msgIn chan fixIn, msgOut chan []byte, quit chan bool) {
 			return
 		}
 
-		s.sendQueued()
-
 		select {
+		case request := <-s.toSend:
+			if s.IsLoggedOn() {
+				request.err <- s.send(request.msg)
+			} else {
+				request.err <- fmt.Errorf("Not logged on")
+			}
 		case fixIn, ok := <-msgIn:
 			if ok {
 				s.log.OnIncoming(string(fixIn.bytes))
@@ -575,7 +555,6 @@ func (s *session) run(msgIn chan fixIn, msgOut chan []byte, quit chan bool) {
 			}
 		case evt := <-s.sessionEvent:
 			s.sessionState = s.Timeout(s, evt)
-		case <-s.messageEvent:
 		}
 	}
 }

session_test.go

@@ -326,56 +326,3 @@ func TestSession_CheckToAppCalled(t *testing.T) {
 		t.Error("Toadmin should not have been called, instead was called", app.adminCalled, "times")
 	}
 }
-
-func TestSession_sendQueued(t *testing.T) {
-	app := new(TestClient)
-	otherEnd := make(chan []byte)
-	go func() {
-		for {
-			_, ok := <-otherEnd
-			if !ok {
-				return
-			}
-		}
-	}()
-	session := session{
-		store:       new(memoryStore),
-		application: app,
-		messageOut:  otherEnd,
-		log:         nullLog{}}
-	session.queueForSend(buildMessage())
-	session.queueForSend(buildMessage())
-	session.queueForSend(buildMessage())
-
-	if len(session.toSend) != 3 {
-		t.Errorf("Expected %v queued messages, got %v", 3, len(session.toSend))
-	}
-
-	var tests = []struct {
-		sessionState
-	}{
-		{logonState{}},
-		{logoutState{}},
-	}
-
-	for _, test := range tests {
-		session.sessionState = test.sessionState
-		session.sendQueued()
-
-		if app.appCalled != 0 {
-			t.Fatalf("session state %v should not allow send but sent %v times", session.sessionState, app.appCalled)
-		}
-	}
-
-	session.sessionState = inSession{}
-
-	session.sendQueued()
-
-	if app.appCalled != 3 {
-		t.Errorf("Toapp should have been called %v times, instead was called %v times", 3, app.appCalled)
-	}
-
-	if len(session.toSend) != 0 {
-		t.Errorf("Expected no queued messages, got %v", len(session.toSend))
-	}
-}