Allow setting consumer replicas though options

example_test.go

@@ -625,6 +625,11 @@ func ExampleSubOpt() {
 	js.Subscribe("foo", func(msg *nats.Msg) {
 		fmt.Printf("Received a message: %s\n", string(msg.Data))
 	}, nats.ManualAck(), nats.MaxDeliver(2), nats.BackOff([]time.Duration{50 * time.Millisecond, 250 * time.Millisecond}))
+
+	// Set consumer replicas count for a durable while subscribing.
+	js.Subscribe("foo", func(msg *nats.Msg) {
+		fmt.Printf("Received a message: %s\n", string(msg.Data))
+	}, nats.Durable("FOO"), nats.ConsumerReplicas(1))
 }
 
 func ExampleMaxWait() {

js.go

@@ -1299,6 +1299,9 @@ func checkConfig(s, u *ConsumerConfig) error {
 	if u.Heartbeat > 0 && u.Heartbeat != s.Heartbeat {
 		return makeErr("heartbeat", u.Heartbeat, s.Heartbeat)
 	}
+	if u.Replicas > 0 && u.Replicas != s.Replicas {
+		return makeErr("replicas", u.Replicas, s.Replicas)
+	}
 	return nil
 }
 
@@ -2413,6 +2416,17 @@ func InactiveThreshold(threshold time.Duration) SubOpt {
 	})
 }
 
+// ConsumerReplicas sets the number of replica count for a consumer.
+func ConsumerReplicas(replicas int) SubOpt {
+	return subOptFn(func(opts *subOpts) error {
+		if replicas < 1 {
+			return fmt.Errorf("invalid ConsumerReplicas value (%v), needs to be greater than 0", replicas)
+		}
+		opts.cfg.Replicas = replicas
+		return nil
+	})
+}
+
 func (sub *Subscription) ConsumerInfo() (*ConsumerInfo, error) {
 	sub.mu.Lock()
 	// TODO(dlc) - Better way to mark especially if we attach.

js_test.go

@@ -1026,3 +1026,33 @@ func TestJetStreamClusterPlacement(t *testing.T) {
 		t.Fatalf("Unexpected tag: %q", v)
 	}
 }
+
+func TestConsumerReplicasOption(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, s)
+
+	nc, js := jsClient(t, s)
+	defer nc.Close()
+
+	if _, err := js.AddStream(&StreamConfig{Name: "CONSUMER_REPLICAS_TEST", Subjects: []string{"foo"}}); err != nil {
+		t.Fatalf("Error adding stream: %v", err)
+	}
+
+	cb := func(msg *Msg) {}
+	// Subscribe to the stream with a durable consumer "bar".
+	_, err := js.Subscribe("foo", cb, Durable("bar"), ConsumerReplicas(1))
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	// Get consumer info
+	consInfo, err := js.ConsumerInfo("CONSUMER_REPLICAS_TEST", "bar")
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	// Check if the number of replicas is the same as we provided.
+	if consInfo.Config.Replicas != 1 {
+		t.Fatalf("Expected consumer replica to be %v, got %+v", 1, consInfo.Config.Replicas)
+	}
+}