Merge pull request #1064 from chapurlatn/patch-1

fix: pubsub ack must rely on protocol.IsAck
cloudevents · Jul 18, 2024 · 5f0cc73 · 5f0cc73
2 parents 42a298d + b94c458
commit 5f0cc73
Show file tree

Hide file tree

Showing 3 changed files with 32 additions and 13 deletions.
diff --git a/protocol/pubsub/v2/message.go b/protocol/pubsub/v2/message.go
@@ -14,6 +14,7 @@ import (
 	"github.com/cloudevents/sdk-go/v2/binding"
 	"github.com/cloudevents/sdk-go/v2/binding/format"
 	"github.com/cloudevents/sdk-go/v2/binding/spec"
+	"github.com/cloudevents/sdk-go/v2/protocol"
 )
 
 const (
@@ -119,15 +120,13 @@ func (m *Message) GetExtension(name string) interface{} {
 	return m.internal.Attributes[prefix+name]
 }
 
-// Finish marks the message to be forgotten.
-// If err is nil, the underlying Pubsub message will be acked;
-// otherwise nacked and return the error.
+// Finish marks the message to be forgotten and returns the provided error without modification.
+// If err is nil or of type protocol.ResultACK the PubSub message will be acknowledged, otherwise nack-ed.
 func (m *Message) Finish(err error) error {
-	if err != nil {
+	if protocol.IsACK(err) {
+		m.internal.Ack()
+	} else {
 		m.internal.Nack()
-		return err
 	}
-
-	m.internal.Ack()
-	return nil
+	return err
 }
diff --git a/protocol/pubsub/v2/message_test.go b/protocol/pubsub/v2/message_test.go
@@ -13,6 +13,7 @@ import (
 	"cloud.google.com/go/pubsub"
 	"github.com/cloudevents/sdk-go/v2/binding"
 	"github.com/cloudevents/sdk-go/v2/event"
+	"github.com/cloudevents/sdk-go/v2/protocol"
 )
 
 func TestReadStructured(t *testing.T) {
@@ -62,6 +63,22 @@ func TestFinish(t *testing.T) {
 			err:     fmt.Errorf("error"),
 			wantErr: true,
 		},
+		{
+			name: "result not acked",
+			pm: &pubsub.Message{
+				ID: "testid",
+			},
+			err:     protocol.NewReceipt(false, "error"),
+			wantErr: true,
+		},
+		{
+			name: "result acked",
+			pm: &pubsub.Message{
+				ID: "testid",
+			},
+			err:     protocol.NewReceipt(true, "error"),
+			wantErr: true,
+		},
 		{
 			name: "no errors",
 			pm: &pubsub.Message{

diff --git a/v2/client/client.go b/v2/client/client.go
@@ -38,15 +38,18 @@ type Client interface {
 	// * func()
 	// * func() error
 	// * func(context.Context)
-	// * func(context.Context) protocol.Result
+	// * func(context.Context) error
 	// * func(event.Event)
-	// * func(event.Event) protocol.Result
+	// * func(event.Event) error
 	// * func(context.Context, event.Event)
-	// * func(context.Context, event.Event) protocol.Result
+	// * func(context.Context, event.Event) error
 	// * func(event.Event) *event.Event
-	// * func(event.Event) (*event.Event, protocol.Result)
+	// * func(event.Event) (*event.Event, error)
 	// * func(context.Context, event.Event) *event.Event
-	// * func(context.Context, event.Event) (*event.Event, protocol.Result)
+	// * func(context.Context, event.Event) (*event.Event, error)
+	// The error returned may impact the messages processing made by the protocol
+	// used (example: message acknowledgement). Please refer to each protocol's
+	// package documentation of the function "Finish(err error) error".
 	StartReceiver(ctx context.Context, fn interface{}) error
 }