feat(jetstream): handle "no results" status in direct APIs and add relevant tests

- Added `isNoResults` method to identify "no results" error in JetStream errors.
- Updated logic in `jsm_direct.ts` to stop iterators on "no results" status.
- Enhanced tests with scenarios for handling batches and direct message retrieval when no messages are available.

Signed-off-by: Alberto Ricart <[email protected]>

Author: aricart

jetstream/src/jserrors.ts

@@ -176,6 +176,10 @@ export class JetStreamStatus {
     return this.code === 404 && this.description === "message not found";
   }
 
+  isNoResults(): boolean {
+    return this.code === 404 && this.description === "no results";
+  }
+
   isMessageSizeExceedsMaxBytes(): boolean {
     return this.code === 409 &&
       this.description === "message size exceeds maxbytes";

jetstream/src/jsm_direct.ts

@@ -183,6 +183,11 @@ export class DirectStreamAPIImpl extends BaseApiClientImpl
         }
         const status = JetStreamStatus.maybeParseStatus(msg);
         if (status) {
+          if (status.isNoResults()) {
+            push({}, () => {
+              iter.stop();
+            });
+          }
           if (status.isEndOfBatch()) {
             push({}, () => {
               iter.stop();

jetstream/tests/jsm_direct_test.ts

@@ -492,13 +492,8 @@ Deno.test("direct - last message for", async (t) => {
   }
 
   await t.step("not matched filter", async () => {
-    await assertRejects(
-      async () => {
-        await assertBatch({ opts: { multi_last: ["c"] }, expect: [] });
-      },
-      JetStreamError,
-      "no results",
-    );
+      await assertBatch({ opts: { multi_last: ["c"] }, expect: [] });
+
   });
 
   await t.step("single filter", async () => {
@@ -580,3 +575,49 @@ Deno.test("direct - batch next_by_subj", async () => {
 
   await cleanup(ns, nc);
 });
+
+Deno.test("direct - batch no messages", async () => {
+  const { ns, nc } = await setup(jetstreamServerConf(), { debug: true });
+  if (await notCompatible(ns, nc, "2.11.0")) {
+    return;
+  }
+  const nci = nc as NatsConnectionImpl;
+  const jsm = await jetstreamManager(nci) as JetStreamManagerImpl;
+  await jsm.streams.add({
+    name: "A",
+    subjects: ["a", "b", "z"],
+    storage: StorageType.Memory,
+    allow_direct: true,
+  });
+
+  const iter = await jsm.direct.getLastMessagesFor("A", {
+    multi_last: ["a", "b", "z"],
+    batch: 100,
+  });
+  for await (const m of iter) {
+    console.log(m.seq);
+  }
+  assertEquals(iter.getProcessed(), 0);
+
+  await cleanup(ns, nc);
+});
+
+Deno.test("direct - get no messages", async () => {
+  const { ns, nc } = await setup(jetstreamServerConf(), { debug: true });
+  if (await notCompatible(ns, nc, "2.11.0")) {
+    return;
+  }
+  const nci = nc as NatsConnectionImpl;
+  const jsm = await jetstreamManager(nci) as JetStreamManagerImpl;
+  await jsm.streams.add({
+    name: "A",
+    subjects: ["a", "b", "z"],
+    storage: StorageType.Memory,
+    allow_direct: true,
+  });
+
+  const m = await jsm.direct.getMessage("A", { last_by_subj: "a" });
+  assertEquals(m, null);
+
+  await cleanup(ns, nc);
+});