jetstream: support async acknowledgement in subscriber (#22)

ThreeDotsLabs · Mar 22, 2024 · 423b241 · 423b241
1 parent 69da773
commit 423b241
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 2 deletions.
diff --git a/pkg/jetstream/config.go b/pkg/jetstream/config.go
@@ -55,6 +55,9 @@ type SubscriberConfig struct {
 
 	// ConsumeOptions is the option that adjusts consume behavior
 	ConsumeOptions []natsJS.PullConsumeOpt
+
+	// AckAsync enables asynchronous acknowledgement
+	AckAsync bool
 }
 
 // setDefaults sets default values needed for a subscriber if unset

diff --git a/pkg/jetstream/message_handler.go b/pkg/jetstream/message_handler.go
@@ -55,8 +55,11 @@ func (s *Subscriber) handleMsg(ctx context.Context, msg jetstream.Msg, output ch
 
 	select {
 	case <-m.Acked():
-		// TODO: enable async ack via config
-		err = msg.DoubleAck(ctx)
+		if s.ackAsync {
+			err = msg.Ack()
+		} else {
+			err = msg.DoubleAck(ctx)
+		}
 
 		if err != nil {
 			s.logger.Error("Cannot send ack", err, messageLogFields)

diff --git a/pkg/jetstream/subscriber.go b/pkg/jetstream/subscriber.go
@@ -33,6 +33,7 @@ type Subscriber struct {
 	configureStream   StreamConfigurator
 	configureConsumer ConsumerConfigurator
 	consumeOptions    []jetstream.PullConsumeOpt
+	ackAsync          bool
 }
 
 // NewSubscriber creates a new watermill JetStream subscriber.
@@ -72,6 +73,7 @@ func newSubscriber(nc *nats.Conn, config *SubscriberConfig) (*Subscriber, error)
 		configureStream:   config.ConfigureStream,
 		configureConsumer: config.ConfigureConsumer,
 		consumeOptions:    config.ConsumeOptions,
+		ackAsync:          config.AckAsync,
 	}, nil
 }