Add use_existing_consumer option for NATS async consumer (#1074)

Author: b43

internal/configtypes/types.go

@@ -925,6 +925,12 @@ type NatsJetStreamConsumerConfig struct {
 	Token string `mapstructure:"token" json:"token" toml:"token" yaml:"token"`
 	// StreamName is the name of the NATS JetStream stream to use.
 	StreamName string `mapstructure:"stream_name" json:"stream_name" toml:"stream_name" yaml:"stream_name"`
+	// UseExistingConsumer when enabled tells Centrifugo to use an existing consumer with
+	// durable_consumer_name instead of creating a new one. When on, these fields are ignored:
+	// deliver_policy, subjects, max_ack_pending, and all other consumer-creation-related options
+	// which may be added later (like ack wait, etc.). The existing consumer's configuration defines
+	// all behavior, and Centrifugo will fail to start if the consumer does not already exist.
+	UseExistingConsumer bool `mapstructure:"use_existing_consumer" default:"false" json:"use_existing_consumer" toml:"use_existing_consumer" yaml:"use_existing_consumer"`
 	// Subjects is the list of NATS subjects (topics) to filter.
 	Subjects []string `mapstructure:"subjects" json:"subjects" toml:"subjects" yaml:"subjects"`
 	// DurableConsumerName sets the name of the durable JetStream consumer to use.
@@ -971,8 +977,10 @@ func (cfg NatsJetStreamConsumerConfig) Validate() error {
 	if cfg.DurableConsumerName == "" {
 		return errors.New("durable_consumer_name is required for consumer")
 	}
-	if cfg.DeliverPolicy != "new" && cfg.DeliverPolicy != "all" {
-		return errors.New("deliver_policy must be either 'new' or 'all'")
+	if !cfg.UseExistingConsumer {
+		if cfg.DeliverPolicy != "new" && cfg.DeliverPolicy != "all" {
+			return errors.New("deliver_policy must be either 'new' or 'all'")
+		}
 	}
 	if cfg.PublicationDataMode.Enabled && cfg.PublicationDataMode.ChannelsHeader == "" {
 		return errors.New("channels_header is required for publication data mode")

internal/consuming/nats_jetstream.go

@@ -106,6 +106,17 @@ func createJetStreamConsumer(ctx context.Context, nc *nats.Conn, cfg NatsJetStre
 	if err != nil {
 		return nil, fmt.Errorf("failed to create JetStream context: %w", err)
 	}
+
+	if cfg.UseExistingConsumer {
+		// If UseExistingConsumer is enabled, just look up the existing consumer by DurableConsumerName
+		consumer, err := js.Consumer(ctx, cfg.StreamName, cfg.DurableConsumerName)
+		if err != nil {
+			return nil, fmt.Errorf("failed to lookup existing consumer %q: %w", cfg.DurableConsumerName, err)
+		}
+		return consumer, nil
+	}
+
+	// By default, create or update the consumer
 	deliverPolicy := jetstream.DeliverNewPolicy
 	if cfg.DeliverPolicy == "all" {
 		deliverPolicy = jetstream.DeliverAllPolicy

internal/consuming/nats_jetstream_test.go

@@ -18,19 +18,21 @@ import (
 func TestNatsJetStreamConsumer(t *testing.T) {
 	t.Parallel()
 	testCases := []struct {
-		name string
+		name                string
+		useExistingConsumer bool
 	}{
-		{name: "basic"},
+		{name: "basic", useExistingConsumer: false},
+		{name: "use_existing_consumer", useExistingConsumer: true},
 	}
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
-			testNatsJetStreamConsumer(t)
+			testNatsJetStreamConsumer(t, tc.useExistingConsumer)
 		})
 	}
 }
 
-func testNatsJetStreamConsumer(t *testing.T) {
+func testNatsJetStreamConsumer(t *testing.T, useExistingConsumer bool) {
 	url := "nats://localhost:4222"
 	subject := "test.subject." + uuid.NewString()
 	durableConsumerName := "test-durable-" + uuid.NewString()
@@ -50,6 +52,15 @@ func testNatsJetStreamConsumer(t *testing.T) {
 	})
 	require.NoError(t, err)
 
+	// If useExistingConsumer is true, create the consumer manually first
+	if useExistingConsumer {
+		_, err = js.AddConsumer(streamName, &nats.ConsumerConfig{
+			Durable: durableConsumerName,
+			Name:    durableConsumerName,
+		})
+		require.NoError(t, err)
+	}
+
 	var receivedNum atomic.Int64
 
 	createConsumer := func(name string, done chan struct{}) *NatsJetStreamConsumer {
@@ -68,6 +79,7 @@ func testNatsJetStreamConsumer(t *testing.T) {
 			DurableConsumerName: durableConsumerName, // shared durable name
 			DeliverPolicy:       "new",
 			MethodHeader:        "test-method",
+			UseExistingConsumer: useExistingConsumer,
 		}
 
 		consumer, err := NewNatsJetStreamConsumer(cfg, dispatcher, testCommon(prometheus.NewRegistry()))