TRD-1749: Escape send queued when blocked on connection side (#18)

Author: AlexandrosKyriakakis

session.go

@@ -235,12 +235,16 @@ func (s *session) queueForSend(msg *Message) error {
 
 	s.toSend = append(s.toSend, msgBytes)
 
+	s.notifyMessageOut()
+
+	return nil
+}
+
+func (s *session) notifyMessageOut() {
 	select {
 	case s.messageEvent <- true:
 	default:
 	}
-
-	return nil
 }
 
 // send will validate, persist, queue the message. If the session is logged on, send all messages in the queue.
@@ -347,8 +351,12 @@ func (s *session) persist(seqNum int, msgBytes []byte) error {
 }
 
 func (s *session) sendQueued() {
-	for _, msgBytes := range s.toSend {
-		s.sendBytes(msgBytes)
+	for i, msgBytes := range s.toSend {
+		if !s.sendBytes(msgBytes) {
+			s.toSend = s.toSend[i:]
+			s.notifyMessageOut()
+			return
+		}
 	}
 
 	s.dropQueued()
@@ -366,15 +374,20 @@ func (s *session) EnqueueBytesAndSend(msg []byte) {
 	s.sendQueued()
 }
 
-func (s *session) sendBytes(msg []byte) {
+func (s *session) sendBytes(msg []byte) bool {
 	if s.messageOut == nil {
 		s.log.OnEventf("Failed to send: disconnected")
-		return
+		return false
 	}
 
-	s.log.OnOutgoing(msg)
-	s.messageOut <- msg
-	s.stateTimer.Reset(s.HeartBtInt)
+	select {
+	case s.messageOut <- msg:
+		s.log.OnOutgoing(msg)
+		s.stateTimer.Reset(s.HeartBtInt)
+		return true
+	default:
+		return false
+	}
 }
 
 func (s *session) doTargetTooHigh(reject targetTooHigh) (nextState resendState, err error) {