[message] returning true/false instead of error when Ack/Nack (#40)

message/infrastructure/test_pubsub.go

@@ -604,7 +604,7 @@ func TestMessageCtx(t *testing.T, pubSub PubSub, features Features) {
 			// ok
 		}
 
-		require.NoError(t, msg.Ack())
+		require.True(t, msg.Ack())
 
 		select {
 		case <-ctx.Done():

message/message.go

@@ -4,8 +4,6 @@ import (
 	"bytes"
 	"context"
 	"sync"
-
-	"github.com/pkg/errors"
 )
 
 var closedchan = make(chan struct{})
@@ -14,11 +12,6 @@ func init() {
 	close(closedchan)
 }
 
-var (
-	ErrAlreadyAcked  = errors.New("message already acked")
-	ErrAlreadyNacked = errors.New("message already nacked")
-)
-
 type Payload []byte
 
 type Message struct {
@@ -86,16 +79,16 @@ func (m Message) Equals(toCompare *Message) bool {
 //
 // Ack is not blocking.
 // Ack is idempotent.
-// Error is returned, if Nack is already sent.
-func (m *Message) Ack() error {
+// False is returned, if Nack is already sent.
+func (m *Message) Ack() bool {
 	m.ackMutex.Lock()
 	defer m.ackMutex.Unlock()
 
 	if m.ackSentType == nack {
-		return ErrAlreadyNacked
+		return false
 	}
 	if m.ackSentType != noAckSent {
-		return nil
+		return true
 	}
 
 	m.ackSentType = ack
@@ -105,23 +98,23 @@ func (m *Message) Ack() error {
 		close(m.ack)
 	}
 
-	return nil
+	return true
 }
 
 // Nack sends message's negative acknowledgement.
 //
 // Nack is not blocking.
 // Nack is idempotent.
-// Error is returned, if Ack is already sent.
-func (m *Message) Nack() error {
+// False is returned, if Ack is already sent.
+func (m *Message) Nack() bool {
 	m.ackMutex.Lock()
 	defer m.ackMutex.Unlock()
 
 	if m.ackSentType == ack {
-		return ErrAlreadyAcked
+		return false
 	}
 	if m.ackSentType != noAckSent {
-		return nil
+		return true
 	}
 
 	m.ackSentType = nack
@@ -132,7 +125,7 @@ func (m *Message) Nack() error {
 		close(m.noAck)
 	}
 
-	return nil
+	return true
 }
 
 // Acked returns channel which is closed when acknowledgement is sent.

message/message_test.go

@@ -58,55 +58,55 @@ func TestMessage_Equals(t *testing.T) {
 
 func TestMessage_Ack(t *testing.T) {
 	msg := &message.Message{}
-	require.NoError(t, msg.Ack())
+	require.True(t, msg.Ack())
 
 	assertAcked(t, msg)
 	assertNoNack(t, msg)
 }
 
 func TestMessage_Ack_idempotent(t *testing.T) {
 	msg := &message.Message{}
-	require.NoError(t, msg.Ack())
-	require.NoError(t, msg.Ack())
+	require.True(t, msg.Ack())
+	require.True(t, msg.Ack())
 
 	assertAcked(t, msg)
 }
 
 func TestMessage_Ack_already_Nack(t *testing.T) {
 	msg := &message.Message{}
-	require.NoError(t, msg.Nack())
+	require.True(t, msg.Nack())
 
-	assert.Equal(t, message.ErrAlreadyNacked, msg.Ack())
+	assert.False(t, msg.Ack())
 }
 
 func TestMessage_Nack(t *testing.T) {
 	msg := &message.Message{}
-	require.NoError(t, msg.Nack())
+	require.True(t, msg.Nack())
 
 	assertNoAck(t, msg)
 	assertNacked(t, msg)
 }
 
 func TestMessage_Nack_idempotent(t *testing.T) {
 	msg := &message.Message{}
-	require.NoError(t, msg.Nack())
-	require.NoError(t, msg.Nack())
+	require.True(t, msg.Nack())
+	require.True(t, msg.Nack())
 
 	assertNacked(t, msg)
 }
 
 func TestMessage_Nack_already_Ack(t *testing.T) {
 	msg := &message.Message{}
-	require.NoError(t, msg.Ack())
+	require.True(t, msg.Ack())
 
-	assert.Equal(t, message.ErrAlreadyAcked, msg.Nack())
+	assert.False(t, msg.Nack())
 }
 
 func TestMessage_Copy(t *testing.T) {
 	msg := message.NewMessage("1", []byte("foo"))
 	msgCopy := msg.Copy()
 
-	require.NoError(t, msg.Ack())
+	require.True(t, msg.Ack())
 
 	assertAcked(t, msg)
 	assertNoAck(t, msgCopy)

message/router_test.go

@@ -249,7 +249,7 @@ func TestRouterNoPublisherHandler(t *testing.T) {
 		pubSub,
 		func(msg *message.Message) (producedMessages []*message.Message, err error) {
 			if msgReceived {
-				require.NoError(t, msg.Ack())
+				require.True(t, msg.Ack())
 				close(wait)
 				return nil, nil
 			}