Merge pull request #121 from go-stomp/fix-issue-120

fix: wrong header sent for explicit acknowledgment

Author: worg

conn.go

@@ -2,12 +2,13 @@ package stomp
 
 import (
 	"errors"
-	"github.com/go-stomp/stomp/v3/frame"
 	"io"
 	"net"
 	"strconv"
 	"sync"
 	"time"
+
+	"github.com/go-stomp/stomp/v3/frame"
 )
 
 // Default time span to add to read/write heart-beat timeouts
@@ -750,10 +751,12 @@ func (c *Conn) createAckNackFrame(msg *Message, ack bool) (*frame.Frame, error)
 			return nil, missingHeader(frame.MessageId)
 		}
 	case V12:
-		if ack, ok := msg.Header.Contains(frame.Id); ok {
+		// message frame contains ack header
+		if ack, ok := msg.Header.Contains(frame.Ack); ok {
+			// ack frame should reference it as id
 			f.Header.Add(frame.Id, ack)
 		} else {
-			return nil, missingHeader(frame.Id)
+			return nil, missingHeader(frame.Ack)
 		}
 	}
 

conn_test.go

@@ -32,7 +32,7 @@ func (rw *fakeReaderWriter) Close() error {
 
 func (s *StompSuite) Test_conn_option_set_logger(c *C) {
 	fc1, fc2 := testutil.NewFakeConn(c)
-		go func() {
+	go func() {
 
 		defer func() {
 			fc2.Close()
@@ -384,8 +384,8 @@ func subscribeHelper(c *C, ackMode AckMode, version Version, opts ...func(*frame
 				frame.Subscription, id,
 				frame.MessageId, messageId,
 				frame.Destination, destination)
-			if version == V12 {
-				f4.Header.Add(frame.Id, messageId)
+			if version == V12 && ackMode.ShouldAck() {
+				f4.Header.Add(frame.Ack, messageId)
 			}
 			f4.Body = []byte(bodyText)
 			err = rw.Write(f4)
@@ -395,7 +395,7 @@ func subscribeHelper(c *C, ackMode AckMode, version Version, opts ...func(*frame
 				f5, _ := rw.Read()
 				c.Assert(f5.Command, Equals, "ACK")
 				if version == V12 {
-					c.Assert(f5.Header.Get("id"), Equals, messageId)
+					c.Assert(f5.Header.Get(frame.Id), Equals, messageId)
 				} else {
 					c.Assert(f5.Header.Get("subscription"), Equals, id)
 					c.Assert(f5.Header.Get("message-id"), Equals, messageId)
@@ -431,6 +431,9 @@ func subscribeHelper(c *C, ackMode AckMode, version Version, opts ...func(*frame
 		c.Assert(msg.Body, DeepEquals, []byte(bodyText))
 		c.Assert(msg.Destination, Equals, "/queue/test-1")
 		c.Assert(msg.Header.Get(frame.MessageId), Equals, messageId)
+		if version == V12 && ackMode.ShouldAck() {
+			c.Assert(msg.Header.Get(frame.Ack), Equals, messageId)
+		}
 
 		c.Assert(msg.ShouldAck(), Equals, ackMode.ShouldAck())
 		if msg.ShouldAck() {
@@ -491,8 +494,8 @@ func subscribeTransactionHelper(c *C, ackMode AckMode, version Version, abort bo
 				frame.Subscription, id,
 				frame.MessageId, messageId,
 				frame.Destination, destination)
-			if version == V12 {
-				f4.Header.Add(frame.Id, messageId)
+			if version == V12 && ackMode.ShouldAck() {
+				f4.Header.Add(frame.Ack, messageId)
 			}
 			f4.Body = []byte(bodyText)
 			err = rw.Write(f4)
@@ -514,7 +517,7 @@ func subscribeTransactionHelper(c *C, ackMode AckMode, version Version, abort bo
 					c.Assert(f5.Command, Equals, "ACK")
 				}
 				if version == V12 {
-					c.Assert(f5.Header.Get("id"), Equals, messageId)
+					c.Assert(f5.Header.Get(frame.Id), Equals, messageId)
 				} else {
 					c.Assert(f5.Header.Get("subscription"), Equals, id)
 					c.Assert(f5.Header.Get("message-id"), Equals, messageId)

examples/client_test/client_test

@@ -406,18 +406,19 @@ func (c *Conn) cleanupSubChannel() {
 
 // Send a frame to the client, allocating necessary headers prior.
 func (c *Conn) allocateMessageId(f *frame.Frame, sub *Subscription) {
-	if f.Command == frame.MESSAGE {
+	if f.Command == frame.MESSAGE || f.Command == frame.ACK {
 		// allocate the value of message-id for this frame
 		c.lastMsgId++
 		messageId := strconv.FormatUint(c.lastMsgId, 10)
 		f.Header.Set(frame.MessageId, messageId)
+		f.Header.Set(frame.Id, messageId)
 
 		// if there is any requirement by the client to acknowledge, set
 		// the ack header as per STOMP 1.2
 		if sub == nil || sub.ack == frame.AckAuto {
-			f.Header.Del(frame.Id)
+			f.Header.Del(frame.Ack)
 		} else {
-			f.Header.Set(frame.Id, messageId)
+			f.Header.Set(frame.Ack, messageId)
 		}
 	}
 }
@@ -659,7 +660,7 @@ func (c *Conn) handleAck(f *frame.Frame) error {
 	var err error
 	var msgId string
 
-	if ack, ok := f.Header.Contains(frame.Id); ok {
+	if ack, ok := f.Header.Contains(frame.Ack); ok {
 		msgId = ack
 	} else if msgId, ok = f.Header.Contains(frame.MessageId); !ok {
 		return missingHeader(frame.MessageId)
@@ -702,7 +703,7 @@ func (c *Conn) handleNack(f *frame.Frame) error {
 	var err error
 	var msgId string
 
-	if ack, ok := f.Header.Contains(frame.Id); ok {
+	if ack, ok := f.Header.Contains(frame.Ack); ok {
 		msgId = ack
 	} else if msgId, ok = f.Header.Contains(frame.MessageId); !ok {
 		return missingHeader(frame.MessageId)

server/client/conn.go

@@ -50,27 +50,28 @@ func (s *ServerSuite) TestConnectAndDisconnect(c *C) {
 func (s *ServerSuite) TestHeartBeatingTolerance(c *C) {
 	// Heart beat should not close connection exactly after not receiving message after cx
 	//  it should add a pretty decent amount of time to counter network delay of other timing issues
-	addr := ":59092"
-	l, err := net.Listen("tcp", addr)
+	l, err := net.Listen("tcp", `127.0.0.1:0`)
 	c.Assert(err, IsNil)
 	defer func() { l.Close() }()
 	serv := Server{
-		Addr:          "",
+		Addr:          l.Addr().String(),
 		Authenticator: nil,
 		QueueStorage:  nil,
 		HeartBeat:     5 * time.Millisecond,
 	}
 	go serv.Serve(l)
 
-	conn, err := net.Dial("tcp", "127.0.0.1"+addr)
+	conn, err := net.Dial("tcp", l.Addr().String())
 	c.Assert(err, IsNil)
 	defer conn.Close()
 
-	client, err := stomp.Connect(conn, stomp.ConnOpt.HeartBeat(5 * time.Millisecond, 5 * time.Millisecond))
+	client, err := stomp.Connect(conn, 
+		stomp.ConnOpt.HeartBeat(5 * time.Millisecond, 5 * time.Millisecond),
+	)
 	c.Assert(err, IsNil)
 	defer client.Disconnect()
 
-	time.Sleep(serv.HeartBeat * 50) // let it go for some time to allow client and server to exchange some heart beat
+	time.Sleep(serv.HeartBeat * 20) // let it go for some time to allow client and server to exchange some heart beat
 
 	// Ensure the server has not closed his readChannel
 	err = client.Send("/topic/whatever", "text/plain", []byte("hello"))