Merge pull request #560 from nats-io/test-by-start-reset

[FIX] ordered consumer max_redeliver set to 1, and to use mem_storage

jetstream/consumer.ts

@@ -896,6 +896,8 @@ export class OrderedPullConsumerImpl implements Consumer {
     // reset the consumer sequence as JetStream will renumber from 1
     this.cursor.deliver_seq = 0;
     const config = this.getConsumerOpts(seq);
+    config.max_deliver = 1;
+    config.mem_storage = true;
     let ci;
     // FIXME: replace with general jetstream retry logic
     while (true) {

jetstream/jsmstream_api.ts

@@ -97,12 +97,6 @@ export class ConsumersImpl implements Consumers {
   }
 
   checkVersion(): Promise<void> {
-    if (!this.notified) {
-      this.notified = true;
-      console.log(
-        `\u001B[33m >> consumers framework is beta functionality \u001B[0m`,
-      );
-    }
     const fv = (this.api as ConsumerAPIImpl).nc.features.get(
       Feature.JS_SIMPLIFICATION,
     );

jetstream/tests/consumersordered_test.ts

@@ -706,3 +706,33 @@ Deno.test("ordered - headers only", async () => {
 
   await cleanup(ns, nc);
 });
+
+Deno.test("ordered - max deliver", async () => {
+  const { ns, nc } = await setup(jetstreamServerConf());
+  const js = nc.jetstream();
+
+  const jsm = await nc.jetstreamManager();
+  await jsm.streams.add({ name: "test", subjects: ["test.*"] });
+
+  const oc = await js.consumers.get("test");
+  const ci = await oc.info();
+  assertExists(ci);
+  assertEquals(ci.config.max_deliver, 1);
+
+  await cleanup(ns, nc);
+});
+
+Deno.test("ordered - mem", async () => {
+  const { ns, nc } = await setup(jetstreamServerConf());
+  const js = nc.jetstream();
+
+  const jsm = await nc.jetstreamManager();
+  await jsm.streams.add({ name: "test", subjects: ["test.*"] });
+
+  const oc = await js.consumers.get("test");
+  const ci = await oc.info();
+  assertExists(ci);
+  assertEquals(ci.config.mem_storage, true);
+
+  await cleanup(ns, nc);
+});