Send tx messages to subscribers in same order as received on broker

Signed-off-by: Tejas Wanjari <[email protected]>

Author: Tejas Wanjari

README.md

@@ -7,17 +7,24 @@ Includes:
 2. `stompd`: STOMP Broker
 3. `stomper`: Interactive CLI for STOMP Client
 
-## `stomp`
+## stomp
 Package import:
-```
+```shell
 go get -u github.com/tjs-w/go-proto-stomp
 ```
 
 ## stomper
 
+```shell
+stomper -p tcp
+```
+
 ![stomper demo](stomper.gif "stomper")
 
 ## stompd
+```shell
+stompd -p tcp <host> <port>
+```
 
 ## STOMP Library Documentation
 

pkg/stomp/subscribe.go

@@ -84,19 +84,27 @@ func publish(frame *Frame, txID string) error {
 		return errorMsg(errBrokerStateMachine, "Missing entry in destToSubsMap, for key: "+dest)
 	}
 
+	sendIt := func(subsID string, info *subsInfo) {
+		info.Lock()
+		defer info.Unlock()
+
+		if err := info.sessionHandler.sendMessage(dest, subsID, info.nextAckNum, txID,
+			frame.headers, frame.body); err != nil {
+			fmt.Println(err)
+			return
+		}
+		info.pendingAckBitmap.Add(info.nextAckNum)
+		info.nextAckNum++
+	}
+
 	for subsID, info := range destToSubsMap[dest] {
-		go func(subsID string, info *subsInfo) {
-			info.Lock()
-			defer info.Unlock()
-
-			if err := info.sessionHandler.sendMessage(dest, subsID, info.nextAckNum, txID,
-				frame.headers, frame.body); err != nil {
-				fmt.Println(err)
-				return
-			}
-			info.pendingAckBitmap.Add(info.nextAckNum)
-			info.nextAckNum++
-		}(subsID, info)
+		if txID == "" {
+			// Parallelize sending non-tx messages
+			go sendIt(subsID, info)
+		} else {
+			// Send tx messages in same sequence (do not parallelize to maintain order)
+			sendIt(subsID, info)
+		}
 	}
 
 	return nil