conat: more unit tests of dstream

Author: williamstein

src/packages/nats/sync/dstream.ts

@@ -48,7 +48,7 @@ import { COCALC_MESSAGE_ID_HEADER } from "./core-stream";
 
 const logger = getLogger("dstream");
 
-const MAX_PARALLEL = 250;
+const MAX_PARALLEL = 50;
 
 export interface DStreamOptions extends StreamOptions {
   noAutosave?: boolean;
@@ -75,7 +75,6 @@ export class DStream<T = any> extends EventEmitter {
 
   constructor(opts: DStreamOptions) {
     super();
-
     if (
       opts.noInventory ||
       opts.ephemeral ||
@@ -88,12 +87,10 @@ export class DStream<T = any> extends EventEmitter {
     this.noAutosave = !!opts.noAutosave;
     this.name = opts.name;
     this.stream =
-      opts.ephemeral || opts.persist
-        ? new CoreStream(opts)
-        : new Stream(opts);
+      opts.ephemeral || opts.persist ? new CoreStream(opts) : new Stream(opts);
     this.messages = this.stream.messages;
     this.raw = this.stream.raw;
-    if (!this.noAutosave) {
+    if (!opts.ephemeral && !this.noAutosave) {
       this.client = getClient();
       this.client.on("connected", this.save);
     }
@@ -242,7 +239,7 @@ export class DStream<T = any> extends EventEmitter {
       }
       return obj;
     }
-    return this.opts.env.jc.decode(this.opts.env.jc.encode(obj));
+    return obj;
   };
 
   publish = (
@@ -453,12 +450,20 @@ export class DStream<T = any> extends EventEmitter {
   );
 }
 
-export const cache = refCache<UserStreamOptions, DStream>({
+type CreateOptions = UserStreamOptions & {
+  noAutosave?: boolean;
+  noInventory?: boolean;
+  leader?: boolean;
+  ephemeral?: boolean;
+  persist?: boolean;
+};
+
+export const cache = refCache<CreateOptions, DStream>({
   name: "dstream",
   createKey: userStreamOptionsKey,
   createObject: async (options) => {
-    await waitUntilConnected();
-    if (options.env == null) {
+    if (options.env == null && !options.ephemeral) {
+      await waitUntilConnected();
       options.env = await getEnv();
     }
     const { account_id, project_id, name, valueType = "json" } = options;
@@ -483,14 +488,6 @@ export const cache = refCache<UserStreamOptions, DStream>({
   },
 });
 
-export async function dstream<T>(
-  options: UserStreamOptions & {
-    noAutosave?: boolean;
-    noInventory?: boolean;
-    leader?: boolean;
-    ephemeral?: boolean;
-    persist?: boolean;
-  },
-): Promise<DStream<T>> {
+export async function dstream<T>(options: CreateOptions): Promise<DStream<T>> {
   return await cache(options);
 }

src/packages/nats/sync/stream.ts

@@ -70,6 +70,7 @@ import jsonStableStringify from "json-stable-stringify";
 import { asyncDebounce } from "@cocalc/util/async-utils";
 import { waitUntilReady } from "@cocalc/nats/tiered-storage/client";
 import { COCALC_MESSAGE_ID_HEADER } from "./core-stream";
+import type { Client } from "@cocalc/nats/core/client";
 
 interface RawMsg extends Msg {
   timestamp: number;
@@ -921,13 +922,14 @@ export interface UserStreamOptions {
   noCache?: boolean;
   desc?: JSONValue;
   valueType?: ValueType;
+  client?: Client;
 }
 
 export function userStreamOptionsKey(options: UserStreamOptions) {
   if (!options.name) {
     throw Error("name must be specified");
   }
-  const { env, ...x } = options;
+  const { env, client, ...x } = options;
   return jsonStableStringify(x);
 }
 

src/packages/server/nats/test/core/core-stream-ephemeral.test.ts

@@ -7,6 +7,7 @@ import {
 import { wait } from "@cocalc/server/nats/test/util";
 import type { Client } from "@cocalc/nats/core/client";
 import { is_date as isDate } from "@cocalc/util/misc";
+import { dstream } from "@cocalc/nats/sync/dstream";
 
 beforeAll(before);
 
@@ -248,4 +249,56 @@ describe("test creating a non-leader first, then the leader", () => {
   });
 });
 
+describe("test using ephemeral dstream", () => {
+  let client;
+  let stream;
+  let name = `test-${Math.random()}`;
+
+  it("creates an ephemeral dstream", async () => {
+    client = connect();
+    stream = await dstream({ client, name, ephemeral: true, leader: true });
+    expect(stream.length).toBe(0);
+  });
+
+  it("publishes a value", () => {
+    stream.publish(0);
+    expect(stream.getAll()).toEqual([0]);
+  });
+
+  // [ ] TODO: this is be fast even for count=10000,
+  // but it is NOT. We use a smaller value for now.
+  const count = 100;
+  it(`publish ${count} messages`, async () => {
+    const v: number[] = [0];
+    for (let i = 1; i < count; i++) {
+      stream.publish(i);
+      v.push(i);
+      expect(stream.get(i)).toBe(i);
+      expect(stream.length).toBe(i + 1);
+    }
+    expect(stream.length).toBe(count);
+    expect(stream.getAll()).toEqual(v);
+    await stream.save();
+  });
+
+  let client2;
+  let stream2;
+  it("opens a second dstream non-leader", async () => {
+    client2 = connect();
+    stream2 = await dstream({
+      client: client2,
+      name,
+      ephemeral: true,
+      leader: true,
+    });
+    expect(stream2.length).toBe(count);
+  });
+
+  it("write to the second stream and see reflected in the first", async () => {
+    await stream2.publish("x");
+    wait({ until: () => stream.length == 101 });
+    expect(stream.get(stream.length - 1)).toBe("x");
+  });
+});
+
 afterAll(after);