added a test to verify the reuse of the consumer (#192)

* Added a test to verify the reuse of the consumer
nats-io · Sep 3, 2021 · dad900f · dad900f
1 parent ddf5a41
commit dad900f
Show file tree

Hide file tree

Showing 3 changed files with 46 additions and 3 deletions.
diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
@@ -30,7 +30,7 @@ jobs:
           deno-version: ${{ matrix.deno-version }}
 
       - name: Set NATS Server Version
-        run: echo "NATS_VERSION=v2.3.3" >> $GITHUB_ENV
+        run: echo "NATS_VERSION=v2.4.0" >> $GITHUB_ENV
 
       - name: Get nats-server
         run: |

diff --git a/nats-base-client/jsclient.ts b/nats-base-client/jsclient.ts
@@ -368,8 +368,6 @@ export class JetStreamClientImpl extends BaseApiClient
 
     if (!jsi.attached) {
       jsi.config.filter_subject = subject;
-      // jsi.config.deliver_subject = jsi.config.deliver_subject ??
-      //   createInbox(this.nc.options.inboxPrefix);
     }
 
     jsi.deliver = jsi.config.deliver_subject ||

diff --git a/tests/jetstream_test.ts b/tests/jetstream_test.ts
@@ -41,6 +41,7 @@ import {
   NatsError,
   nuid,
   QueuedIterator,
+  RetentionPolicy,
   StringCodec,
 } from "../nats-base-client/internal_mod.ts";
 import {
@@ -1964,3 +1965,47 @@ Deno.test("jetstream - cleanup", async () => {
 
   await cleanup(ns, nc);
 });
+
+Deno.test("jetstream - reuse consumer", async () => {
+  const { ns, nc } = await setup(jetstreamServerConf({}, true));
+  const id = nuid.next();
+  const jsm = await nc.jetstreamManager();
+  await jsm.streams.add({
+    subjects: [`${id}.*`],
+    name: id,
+    retention: RetentionPolicy.Workqueue,
+  });
+
+  await jsm.consumers.add(id, {
+    "durable_name": "X",
+    "deliver_subject": "out",
+    "deliver_policy": DeliverPolicy.All,
+    "ack_policy": AckPolicy.Explicit,
+    "deliver_group": "queuea",
+  });
+
+  // second create should be OK, since it is idempotent
+  await jsm.consumers.add(id, {
+    "durable_name": "X",
+    "deliver_subject": "out",
+    "deliver_policy": DeliverPolicy.All,
+    "ack_policy": AckPolicy.Explicit,
+    "deliver_group": "queuea",
+  });
+
+  const js = nc.jetstream();
+  const opts = consumerOpts();
+  opts.ackExplicit();
+  opts.durable("X");
+  opts.deliverAll();
+  opts.deliverTo("out2");
+  opts.queue("queuea");
+
+  const sub = await js.subscribe(`${id}.*`, opts);
+  const ci = await sub.consumerInfo();
+  // the deliver subject we specified should be ignored
+  // the one specified by the consumer is used
+  assertEquals(ci.config.deliver_subject, "out");
+
+  await cleanup(ns, nc);
+});