Fix ack failure for payload with header (#137)

adjust · Feb 8, 2023 · 028d635 · 028d635
1 parent a48807d
commit 028d635
Show file tree

Hide file tree

Showing 2 changed files with 21 additions and 14 deletions.
diff --git a/delivery.go b/delivery.go
@@ -21,14 +21,15 @@ var (
 )
 
 type redisDelivery struct {
-	ctx         context.Context
-	payload     string
-	header      http.Header
-	unackedKey  string
-	rejectedKey string
-	pushKey     string
-	redisClient RedisClient
-	errChan     chan<- error
+	ctx          context.Context
+	payload      string
+	clearPayload string
+	header       http.Header
+	unackedKey   string
+	rejectedKey  string
+	pushKey      string
+	redisClient  RedisClient
+	errChan      chan<- error
 }
 
 func (delivery *redisDelivery) Header() http.Header {
@@ -56,19 +57,19 @@ func newDelivery(
 
 	var err error
 
-	if rd.header, rd.payload, err = ExtractHeaderAndPayload(payload); err != nil {
+	if rd.header, rd.clearPayload, err = ExtractHeaderAndPayload(payload); err != nil {
 		return nil, err
 	}
 
 	return &rd, nil
 }
 
 func (delivery *redisDelivery) String() string {
-	return fmt.Sprintf("[%s %s]", delivery.payload, delivery.unackedKey)
+	return fmt.Sprintf("[%s %s]", delivery.clearPayload, delivery.unackedKey)
 }
 
 func (delivery *redisDelivery) Payload() string {
-	return delivery.payload
+	return delivery.clearPayload
 }
 
 // blocking versions of the functions below with the following behavior:

diff --git a/queue_test.go b/queue_test.go
@@ -208,16 +208,21 @@ func TestConsumerCommon(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Nil(t, consumer.Last())
 
-	assert.NoError(t, queue1.Publish("cons-d1"))
+	assert.NoError(t, queue1.Publish(PayloadWithHeader("cons-d1", http.Header{"foo": []string{"bar1"}})))
 	eventuallyReady(t, queue1, 0)
 	eventuallyUnacked(t, queue1, 1)
 	require.NotNil(t, consumer.Last())
 	assert.Equal(t, "cons-d1", consumer.Last().Payload())
+	assert.Equal(t, http.Header{"foo": []string{"bar1"}}, consumer.Last().(WithHeader).Header())
 
-	assert.NoError(t, queue1.Publish("cons-d2"))
+	assert.NoError(t, queue1.Publish(PayloadWithHeader("cons-d2", http.Header{"foo": []string{"bar2"}})))
 	eventuallyReady(t, queue1, 0)
 	eventuallyUnacked(t, queue1, 2)
 	assert.Equal(t, "cons-d2", consumer.Last().Payload())
+	assert.Equal(t, http.Header{"foo": []string{"bar2"}}, consumer.Last().(WithHeader).Header())
+
+	assert.Regexp(t, `\[cons-d2 rmq::connection::cons-conn-\w{6}::queue::\[cons-q\]::unacked\]`,
+		fmt.Sprintf("%s", consumer.Last()))
 
 	assert.NoError(t, consumer.Deliveries()[0].Ack())
 	eventuallyReady(t, queue1, 0)
@@ -229,11 +234,12 @@ func TestConsumerCommon(t *testing.T) {
 
 	assert.Equal(t, ErrorNotFound, consumer.Deliveries()[0].Ack())
 
-	assert.NoError(t, queue1.Publish("cons-d3"))
+	assert.NoError(t, queue1.Publish(PayloadWithHeader("cons-d3", http.Header{"foo": []string{"bar3"}})))
 	eventuallyReady(t, queue1, 0)
 	eventuallyUnacked(t, queue1, 1)
 	eventuallyRejected(t, queue1, 0)
 	assert.Equal(t, "cons-d3", consumer.Last().Payload())
+	assert.Equal(t, http.Header{"foo": []string{"bar3"}}, consumer.Last().(WithHeader).Header())
 	assert.NoError(t, consumer.Last().Reject())
 	eventuallyReady(t, queue1, 0)
 	eventuallyUnacked(t, queue1, 0)