rework to @effectionx/process and createReplaySignal

Author: jbolda

package-lock.json

@@ -27,7 +27,7 @@
   ],
   "scripts": {
     "clean": "echo skip",
-    "test": "node --import tsx --test test/*.test.ts",
+    "test": "node --test-timeout=60000 --import tsx --test test/*.test.ts",
     "prepack": "npm run build",
     "build": "tsdown",
     "lint": "echo noop",
@@ -36,9 +36,9 @@
   },
   "dependencies": {
     "effection": "^4.0.0",
-    "@effectionx/timebox": "^0.3.1",
-    "@effectionx/tinyexec": "^0.3.1",
-    "@effectionx/context-api": "^0.2.1"
+    "@effectionx/context-api": "^0.2.1",
+    "@effectionx/process": "^0.6.2",
+    "@effectionx/timebox": "^0.3.1"
   },
   "devDependencies": {},
   "exports": {

packages/server/package.json

@@ -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,
+  };
+}

packages/server/src/createReplaySignal.ts

@@ -5,6 +5,10 @@ export const stdio = createApi("@simulacrum/logging", {
   *stdout(line: string): Operation<void> {
     console.log(line);
   },
+  *stderr(line: string): Operation<void> {
+    console.log(line);
+  },
 });
 
 export const { stdout } = stdio.operations;
+export const { stderr } = stdio.operations;