createReplaySignal to handle thunk registration

Author: jbolda

src/fx/replay-signal.ts

@@ -0,0 +1,83 @@
+import type { Resolve, Subscription } from "effection";
+import { action, resource } from "effection";
+
+export function createReplaySignal<T, TClose>() {
+  const subscribers = new Set<Subscription<T, TClose>>();
+  // single shared durable queue storage
+  const queue = createDurableQueue<T, TClose>();
+
+  // each subscriber gets its own iterator over the shared items by
+  // calling `queue.subscribe()` which returns a Stream
+  const subscribe = resource<Subscription<T, TClose>>(function* (provide) {
+    const queued = queue.stream();
+    subscribers.add(queued);
+
+    try {
+      yield* provide({ next: queued.next });
+    } finally {
+      subscribers.delete(queued);
+    }
+  });
+
+  function send(value: T) {
+    queue.add(value);
+  }
+
+  function close(value?: TClose) {
+    queue.close(value);
+  }
+
+  return { ...subscribe, send, close };
+}
+
+function createDurableQueue<T, TClose = never>() {
+  type Item = IteratorResult<T, TClose>;
+
+  const items: Item[] = [];
+
+  // a set of active subscribers; each subscription has its own iterator
+  // and its own waiting notifier set
+  const subscribers = new Set<{
+    notify: Set<Resolve<Item>>;
+  }>();
+
+  function enqueue(item: Item) {
+    items.push(item);
+    for (const sub of subscribers) {
+      if (sub.notify.size > 0) {
+        const [resolve] = sub.notify;
+        // use resolve from yield* action to notify waiting subscribers
+        resolve(item);
+      }
+    }
+  }
+
+  function stream(): Subscription<T, TClose> {
+    const iter = items[Symbol.iterator]();
+    const notify = new Set<Resolve<Item>>();
+    const sub = { notify };
+    subscribers.add(sub);
+
+    return {
+      *next() {
+        const item = iter.next().value;
+        // item will be `undefined` when we've iterated to the end of the
+        // current `items` array; in that case we wait for new items to be
+        // enqueued and the resolve will be called with the new `Item`.
+        if (item !== undefined) {
+          return item;
+        }
+        return yield* action<Item>((resolve) => {
+          notify.add(resolve);
+          return () => notify.delete(resolve);
+        });
+      },
+    };
+  }
+
+  return {
+    add: (value: T) => enqueue({ done: false, value }),
+    close: (value?: TClose) => enqueue({ done: true, value: value as TClose }),
+    stream,
+  };
+}

src/query/thunk.ts

@@ -4,7 +4,6 @@ import {
   type Operation,
   type Scope,
   createContext,
-  createSignal,
   each,
   ensure,
   useScope,
@@ -15,6 +14,7 @@ import { supervise } from "../fx/index.js";
 import { createKey } from "./create-key.js";
 import { isFn, isObject } from "./util.js";
 
+import { createReplaySignal } from "../fx/replay-signal.js";
 import { IdContext } from "../store/store.js";
 import type { ActionWithPayload, AnyAction, Next, Payload } from "../types.js";
 import type {
@@ -139,10 +139,9 @@ export function createThunks<Ctx extends ThunkCtx = ThunkCtx<any>>(
   } = { supervisor: takeEvery },
 ): ThunksApi<Ctx> {
   const storeRegistration = new Set();
-  const watch = createSignal<Visors>();
+  const watch = createReplaySignal<Visors>();
 
   const middleware: Middleware<Ctx>[] = [];
-  const visors: { [key: string]: Visors } = {};
   const middlewareMap: { [key: string]: Middleware<Ctx> } = {};
   let dynamicMiddlewareMap: { [key: string]: Middleware<Ctx> } = {};
   const actionMap: {
@@ -174,7 +173,7 @@ export function createThunks<Ctx extends ThunkCtx = ThunkCtx<any>>(
   }
 
   function create(name: string, ...args: any[]) {
-    if (visors[name]) {
+    if (actionMap[name]) {
       const msg = `[${name}] already exists, do you have two thunks with the same name?`;
       console.warn(msg);
     }
@@ -216,10 +215,6 @@ export function createThunks<Ctx extends ThunkCtx = ThunkCtx<any>>(
     function* curVisor(): Operation<void> {
       yield* tt(type, onApi);
     }
-
-    // maintains a history for any future registration
-    visors[name] = () => supervise(curVisor);
-    // signals for any stores already listening
     watch.send(() => supervise(curVisor));
 
     const errMsg = `[${name}] is being called before its thunk has been registered. Run \`store.run(thunks.register)\` where \`thunks\` is the name of your \`createThunks\` or \`createApi\` variable.`;
@@ -262,9 +257,6 @@ export function createThunks<Ctx extends ThunkCtx = ThunkCtx<any>>(
       return kickoff;
     }
 
-    // maintains a history for any future registration
-    visors[name] = curVisor;
-    // signals for any stores already listening
     watch.send(curVisor);
 
     // returns to the user can use this resource from
@@ -285,11 +277,6 @@ export function createThunks<Ctx extends ThunkCtx = ThunkCtx<any>>(
       storeRegistration.delete(parentStoreId);
     });
 
-    // Register any thunks created before listening to signal
-    for (const created of Object.values(visors)) {
-      yield* scope.spawn(created(scope));
-    }
-
     // wait for further thunk create
     for (const watched of yield* each(watch)) {
       yield* scope.spawn(watched(scope));

src/test/replay-signal.test.ts

@@ -0,0 +1,41 @@
+import { createReplaySignal } from "../fx/replay-signal.js";
+import { each, run, sleep, spawn } from "../index.js";
+import { expect, test } from "../test.js";
+
+test("should call the generator function", async () => {
+  const thunk = createReplaySignal();
+  const checkOne = [] as string[];
+  const checkTwo = [] as string[];
+  const checkThree = [] as string[];
+
+  function* valPush(arrKey: string, arr: string[]) {
+    for (const val of yield* each(thunk)) {
+      console.log(arrKey, "received", val);
+      arr.push(val as string);
+      yield* each.next();
+    }
+  }
+
+  await run(function* () {
+    console.log("spawning consumers");
+    const v1 = yield* spawn(() => valPush("checkOne", checkOne));
+    yield* sleep(0); // allow spawns to start
+    thunk.send("first");
+    const v2 = yield* spawn(() => valPush("checkTwo", checkTwo));
+    yield* sleep(0); // allow spawns to start
+    thunk.send("second");
+    const v3 = yield* spawn(() => valPush("checkThree", checkThree));
+    yield* sleep(0); // allow spawns to start
+    thunk.send("third");
+    yield* sleep(0); // allow spawns to start
+    console.log("sending values");
+    thunk.close();
+    yield* v1;
+    yield* v2;
+    yield* v3;
+  });
+
+  expect(checkOne).toHaveLength(3);
+  expect(checkTwo).toHaveLength(3);
+  expect(checkThree).toHaveLength(3);
+});