Merge pull request #165 from bhaan/session-msg-handling

Session event loop follow up

Author: mgatny

registry.go

@@ -39,6 +39,8 @@ func SendToTarget(m Messagable, sessionID SessionID) error {
 	session, err := lookupSession(sessionID)
 	if err != nil {
 		return err
+	} else if session.toSend == nil {
+		return fmt.Errorf("Not logged on")
 	}
 
 	request := sendRequest{msg, make(chan error)}

session.go

@@ -187,7 +187,7 @@ func (s *session) resend(msg Message) {
 }
 
 //send should NOT be called outside of the run loop
-func (s *session) send(msg Message) error {
+func (s *session) send(msg Message) (err error) {
 	s.fillDefaultHeader(msg)
 
 	seqNum := s.store.NextSenderMsgSeqNum()
@@ -199,14 +199,18 @@ func (s *session) send(msg Message) error {
 	} else {
 		s.application.ToApp(msg, s.sessionID)
 	}
-	if msgBytes, err := msg.Build(); err != nil {
-		return err
-	} else {
-		s.store.SaveMessage(seqNum, msgBytes)
+
+	var msgBytes []byte
+	if msgBytes, err = msg.Build(); err != nil {
+		return
+	}
+
+	if err = s.store.SaveMessage(seqNum, msgBytes); err == nil {
 		s.sendBytes(msgBytes)
-		s.store.IncrNextSenderMsgSeqNum()
+		err = s.store.IncrNextSenderMsgSeqNum()
 	}
-	return nil
+
+	return
 }
 
 func (s *session) sendBytes(msg []byte) {
@@ -507,6 +511,7 @@ func (s *session) run(msgIn chan fixIn, msgOut chan []byte, quit chan bool) {
 	defer func() {
 		close(s.messageOut)
 		close(s.toSend)
+		s.toSend = nil
 		s.onDisconnect()
 	}()