[fix] redelivered property on js message and added a test

.github/workflows/test.yml

@@ -30,7 +30,7 @@ jobs:
           deno-version: ${{ matrix.deno-version }}
 
       - name: Set NATS Server Version
-        run: echo "NATS_VERSION=v2.5.0" >> $GITHUB_ENV
+        run: echo "NATS_VERSION=v2.6.1" >> $GITHUB_ENV
 
       - name: Get nats-server
         run: |
@@ -61,11 +61,11 @@ jobs:
       - name: Generate lcov
         run: deno coverage --unstable --lcov ./cov > cov.lcov
 
-#      - name: Upload coverage
-#        uses: coverallsapp/[email protected]
-#        with:
-#          github-token: ${{ secrets.github_token }}
-#          path-to-lcov: ./cov.lcov
+      - name: Upload coverage
+        uses: coverallsapp/[email protected]
+        with:
+          github-token: ${{ secrets.github_token }}
+          path-to-lcov: ./cov.lcov
 
       - name: Release
         uses: softprops/action-gh-release@v1

nats-base-client/jsmsg.ts

@@ -60,6 +60,7 @@ export function parseInfo(s: string): DeliveryInfo {
   di.stream = tokens[4];
   di.consumer = tokens[5];
   di.redeliveryCount = parseInt(tokens[6], 10);
+  di.redelivered = di.redeliveryCount > 1;
   di.streamSequence = parseInt(tokens[7], 10);
   di.deliverySequence = parseInt(tokens[8], 10);
   di.timestampNanos = parseInt(tokens[9], 10);

tests/jetstream_test.ts

@@ -43,6 +43,7 @@ import {
   nuid,
   QueuedIterator,
   RetentionPolicy,
+  StorageType,
   StringCodec,
 } from "../nats-base-client/internal_mod.ts";
 import {
@@ -2119,3 +2120,102 @@ Deno.test("jetstream - pull sub - multiple consumers", async () => {
 
   await cleanup(ns, nc);
 });
+
+Deno.test("jetstream - source", async () => {
+  const { ns, nc } = await setup(jetstreamServerConf({}, true));
+
+  const stream = nuid.next();
+  const subj = `${stream}.*`;
+  const jsm = await nc.jetstreamManager();
+  await jsm.streams.add(
+    { name: stream, subjects: [subj] },
+  );
+
+  const js = nc.jetstream();
+
+  for (let i = 0; i < 10; i++) {
+    await js.publish(`${stream}.A`);
+    await js.publish(`${stream}.B`);
+  }
+
+  await jsm.streams.add({
+    name: "work",
+    storage: StorageType.File,
+    retention: RetentionPolicy.Workqueue,
+    sources: [
+      { name: stream, filter_subject: ">" },
+    ],
+  });
+
+  const ci = await jsm.consumers.add("work", {
+    ack_policy: AckPolicy.Explicit,
+    durable_name: "worker",
+    filter_subject: `${stream}.B`,
+    deliver_subject: createInbox(),
+  });
+
+  const sub = await js.subscribe(`${stream}.B`, { config: ci.config });
+  for await (const m of sub) {
+    console.log(m.seq, m.subject);
+    m.ack();
+    if (m.info.pending === 0) {
+      break;
+    }
+  }
+
+  const si = await jsm.streams.info("work");
+  // stream still has all the 'A' messages
+  assertEquals(si.state.messages, 10);
+
+  await cleanup(ns, nc);
+});
+
+Deno.test("jetstream - redelivery property works", async () => {
+  const { ns, nc } = await setup(jetstreamServerConf({}, true));
+  if (await notCompatible(ns, nc, "2.3.5")) {
+    return;
+  }
+  const { stream, subj } = await initStream(nc);
+  const js = nc.jetstream();
+
+  let r = 0;
+
+  const opts = consumerOpts();
+  opts.ackAll();
+  opts.queue("q");
+  opts.durable("n");
+  opts.deliverTo(createInbox());
+  opts.callback((_err, m) => {
+    if (m) {
+      if (m.info.redelivered) {
+        r++;
+      }
+      if (m.seq === 100) {
+        m.ack();
+      }
+      if (m.seq % 3 === 0) {
+        m.nak();
+      }
+    }
+  });
+
+  const sub = await js.subscribe(subj, opts);
+  const sub2 = await js.subscribe(subj, opts);
+
+  for (let i = 0; i < 100; i++) {
+    await js.publish(subj, Empty);
+  }
+  await nc.flush();
+  await sub.drain();
+  await sub2.drain();
+
+  assert(sub.getProcessed() > 0);
+  assert(sub2.getProcessed() > 0);
+  assertEquals(sub.getProcessed() + sub2.getProcessed(), 100 + r);
+
+  const jsm = await nc.jetstreamManager();
+  const ci = await jsm.consumers.info(stream, "n");
+  assertEquals(ci.delivered.consumer_seq, 100 + r);
+
+  await cleanup(ns, nc);
+});