♻️ Move replay semantics from signals to stream-helpers

Remove `emitCurrentOnSubscribe` from `createValueSignal` — replay
behavior belongs in `createSubject()`, not the signal primitive.

Rewrite `createSubject()` as a resource with an active drain that
tracks upstream values synchronously via a custom SignalQueueFactory
queue and multicasts them through a relay signal. Late subscribers
always receive the latest value, even when `signal.send()` is called
within a running operation.

Breaking: `createSubject()` now returns `Operation<Stream>` instead
of `Stream`. Bump `@effectionx/stream-helpers` 0.8.3 → 0.9.0.

Compose `createSubject` with value signal in worker's
`createWorkerStatesSignal` to preserve replay behavior.

Author: Taras Mankovski

pnpm-lock.yaml

@@ -2,7 +2,9 @@ import type { Stream } from "effection";
 
 /**
  * A signal is a stream with set, update, and valueOf methods.
- * Subscribing to a signal will yield the current value of the signal.
+ * Subscribing to a signal yields values as they change.
+ * Use {@link createSubject} from `@effectionx/stream-helpers` to replay
+ * the latest value to new subscribers.
  */
 export interface ValueSignal<T> extends Stream<T, void> {
   /**

signals/types.ts

@@ -12,20 +12,13 @@ export interface CreateValueSignalOptions<T> {
    * Defaults to `Object.is`.
    */
   equals?: (current: T, next: T) => boolean;
-
-  /**
-   * Replays the current value to subscribers when they attach.
-   *
-   * This is disabled by default.
-   */
-  emitCurrentOnSubscribe?: boolean;
 }
 
 /**
  * Creates a value-backed signal with configurable equality semantics.
  *
  * @param initial - Initial signal value.
- * @param options - Equality and subscription behavior overrides.
+ * @param options - Equality overrides.
  * @returns A value signal resource.
  */
 export function createValueSignal<T>(
@@ -50,13 +43,7 @@ export function createValueSignal<T>(
 
     try {
       yield* provide({
-        [Symbol.iterator]: options.emitCurrentOnSubscribe
-          ? function* () {
-              const subscription = yield* signal;
-              signal.send(ref.current);
-              return subscription;
-            }
-          : signal[Symbol.iterator],
+        [Symbol.iterator]: signal[Symbol.iterator],
         set,
         update(updater) {
           return set(updater(ref.current));

signals/value.ts

@@ -368,14 +368,19 @@ Subject helper converts any stream into a multicast stream that replays the
 latest value to new subscribers. It's analogous to
 [RxJS BehaviorSubject](https://www.learnrxjs.io/learn-rxjs/subjects/behaviorsubject).
 
+Applying the subject to a stream returns a resource. Yielding that resource
+starts an internal drain that actively tracks the upstream, so late subscribers
+always receive the most recent value — even if no other subscriber has pulled
+it.
+
 ```typescript
 import { createSubject } from "@effectionx/stream-helpers";
-import { createChannel, spawn } from "effection";
+import { createChannel } from "effection";
 
 function* example() {
   const subject = createSubject<number>();
   const channel = createChannel<number, void>();
-  const downstream = subject(channel);
+  const downstream = yield* subject(channel);
 
   // First subscriber
   const sub1 = yield* downstream;
@@ -392,22 +397,8 @@ function* example() {
 }
 ```
 
-Use it with a pipe operator to convert any stream into a behavior subject:
-
-```typescript
-import { createSubject, map } from "@effectionx/stream-helpers";
-import { pipe } from "remeda";
-
-const subject = createSubject<string>();
-
-const stream = pipe(
-  source,
-  map(function* (x) {
-    return x.toString();
-  }),
-  subject,
-);
-```
+> **Note:** `createSubject()` returns a resource, so it must be yielded with
+> `yield*`. It cannot be used directly as a pipe transformer.
 
 ### Passthrough Tracker
 

stream-helpers/README.md

@@ -1,7 +1,7 @@
 {
   "name": "@effectionx/stream-helpers",
   "description": "Type-safe stream operators like filter, map, reduce, and forEach",
-  "version": "0.8.3",
+  "version": "0.9.0",
   "keywords": ["streams"],
   "type": "module",
   "main": "./dist/mod.js",

stream-helpers/package.json

@@ -1,6 +1,8 @@
 import {
   createChannel,
+  createSignal,
   type Operation,
+  scoped,
   type Stream,
   type Subscription,
 } from "effection";
@@ -29,7 +31,7 @@ describe("subject", () => {
 
     upstream = createChannel();
 
-    downstream = subject(upstream);
+    downstream = yield* subject(upstream);
   });
 
   it("allows multiple subscribers", function* () {
@@ -78,7 +80,7 @@ describe("subject", () => {
 
   it("yields initial value to first subscriber before any upstream values", function* () {
     subject = createSubject(42);
-    downstream = subject(upstream);
+    downstream = yield* subject(upstream);
 
     const subscriber = yield* downstream;
     expect(yield* next(subscriber)).toEqual(42);
@@ -89,7 +91,7 @@ describe("subject", () => {
 
   it("yields upstream value to late subscriber once upstream has sent", function* () {
     subject = createSubject(42);
-    downstream = subject(upstream);
+    downstream = yield* subject(upstream);
 
     const subscriber1 = yield* downstream;
     expect(yield* next(subscriber1)).toEqual(42);
@@ -115,4 +117,39 @@ describe("subject", () => {
     const subscriber2 = yield* downstream;
     expect(yield* next(subscriber2)).toEqual("bye");
   });
+
+  it("tracks latest value even when no subscriber has pulled", function* () {
+    const source = createSignal<number, string>();
+    downstream = yield* createSubject<number>(0)(source);
+
+    const sub1 = yield* downstream;
+    expect(yield* next(sub1)).toEqual(0);
+
+    // Upstream emits — sub1 does NOT pull these
+    source.send(1);
+    source.send(2);
+
+    // Late subscriber gets latest, not initial
+    const sub2 = yield* downstream;
+    expect(yield* next(sub2)).toEqual(2);
+  });
+
+  it("continues tracking after first subscriber exits", function* () {
+    const source = createSignal<number, string>();
+    downstream = yield* createSubject<number>()(source);
+
+    // First subscriber in a scoped block — exits after reading
+    yield* scoped(function* () {
+      const sub = yield* downstream;
+      source.send(1);
+      expect(yield* next(sub)).toEqual(1);
+    });
+    // sub's scope is gone, but drain lives in the resource scope
+
+    source.send(2);
+
+    // New subscriber still works
+    const sub2 = yield* downstream;
+    expect(yield* next(sub2)).toEqual(2);
+  });
 });

stream-helpers/subject.test.ts

@@ -1,71 +1,115 @@
-import type { Stream, Subscription } from "effection";
+import {
+  createQueue,
+  createSignal,
+  resource,
+  SignalQueueFactory,
+  spawn,
+  type Operation,
+  type Stream,
+  type Subscription,
+} from "effection";
 
 /**
- * Converts any stream into a multicast stream that produces latest value
- * to new subscribers. It's designed to be analagous in function to [RxJS
- * BehaviorSubject](https://www.learnrxjs.io/learn-rxjs/subjects/behaviorsubject).
+ * Converts any stream into a multicast stream that replays the latest value
+ * to new subscribers. Analogous to
+ * [RxJS BehaviorSubject](https://www.learnrxjs.io/learn-rxjs/subjects/behaviorsubject).
  *
- * @returns A function that takes a stream and returns a multicast stream
+ * Applying the subject to a stream returns a resource. Yielding that resource
+ * starts an internal drain that actively tracks the upstream, so late
+ * subscribers always receive the most recent value — even if no other
+ * subscriber has pulled it.
+ *
+ * @returns A function that takes a stream and returns a resource providing
+ *   a multicast stream.
  *
  * @example
  * ```ts
  * const subject = createSubject<number>();
- * const downstream = subject(upstream);
+ * const downstream = yield* subject(upstream);
  *
- * const sub1 = yield* downstream; // subscribes to upstream
+ * const sub1 = yield* downstream; // subscribes
  * yield* upstream.send(1);
  * yield* sub1.next(); // { done: false, value: 1 }
  *
  * const sub2 = yield* downstream; // late subscriber
- * yield* sub2.next(); // { done: false, value: 1 } - gets latest value
- * ```
- *
- * Use it with a pipe operator to convert any stream into a behavior subject.
- *
- * @example
- * ```
- * let source = createChannel<string, void>();
- * let subject = createSubject<string>();
- *
- * let pipeline = pipe([
- *  top,
- *  transform1,
- *  transform2,
- *  subject,
- * ]);
+ * yield* sub2.next(); // { done: false, value: 1 } — gets latest value
  * ```
  */
 export function createSubject<T>(
   initial?: T,
-): <TClose>(stream: Stream<T, TClose>) => Stream<T, TClose> {
+): <TClose>(stream: Stream<T, TClose>) => Operation<Stream<T, TClose>> {
   let current: IteratorResult<T> | undefined =
     typeof initial !== "undefined"
       ? { done: false, value: initial }
       : undefined;
 
-  return <TClose>(stream: Stream<T, TClose>) => ({
-    *[Symbol.iterator]() {
-      let upstream = yield* stream;
+  return <TClose>(stream: Stream<T, TClose>) =>
+    resource<Stream<T, TClose>>(function* (provide) {
+      const relay = createSignal<T, TClose>();
+      let closed = false;
 
-      let tracking: Subscription<T, TClose> = {
-        *next() {
-          current = yield* upstream.next();
-          return current;
-        },
-      };
+      // Install a custom queue that updates `current` synchronously
+      // when values arrive — before the drain task gets scheduled.
+      // This guarantees late subscribers always see the latest value
+      // even when signal.send() is called within a running operation.
+      yield* SignalQueueFactory.set(<U, UClose = never>() => {
+        const queue = createQueue<U, UClose>();
+        return {
+          ...queue,
+          add(value: U) {
+            current = { done: false, value: value as unknown as T };
+            queue.add(value);
+          },
+          close(value: UClose) {
+            current = { done: true, value } as unknown as IteratorResult<T>;
+            closed = true;
+            queue.close(value);
+          },
+        };
+      });
 
-      let iterator: Subscription<T, TClose> = current
-        ? {
-            *next() {
-              iterator = tracking;
-              return current!;
-            },
+      // Subscribe to upstream eagerly so its queue starts buffering
+      // before provide() hands back control to callers.
+      const upstream = yield* stream;
+
+      // Drain owns the read loop that multicasts values to relay subscribers.
+      // Lives in the resource scope — outlives any individual subscriber.
+      yield* spawn(function* () {
+        let result = yield* upstream.next();
+        while (!result.done) {
+          relay.send(result.value);
+          result = yield* upstream.next();
+        }
+        relay.close(result.value);
+      });
+
+      yield* provide({
+        *[Symbol.iterator]() {
+          // Post-close: Effection signals are stateless — subscribing after
+          // relay.close() would hang. Return captured close result directly.
+          if (closed && current) {
+            let snapshot = current;
+            return {
+              *next() {
+                return snapshot as IteratorResult<T, TClose>;
+              },
+            };
           }
-        : tracking;
 
-      return {
-        next: () => iterator.next(),
-      };
-    },
-  });
+          let subscription: Subscription<T, TClose> = yield* relay;
+          let snapshot = current;
+          let replayed = !snapshot;
+
+          return {
+            *next() {
+              if (!replayed) {
+                replayed = true;
+                return snapshot! as IteratorResult<T, TClose>;
+              }
+              return yield* subscription.next();
+            },
+          };
+        },
+      });
+    });
 }

stream-helpers/subject.ts

@@ -29,6 +29,7 @@
   "sideEffects": false,
   "dependencies": {
     "@effectionx/signals": "workspace:*",
+    "@effectionx/stream-helpers": "workspace:*",
     "@effectionx/timebox": "workspace:*",
     "web-worker": "^1"
   },

worker/package.json

@@ -13,6 +13,9 @@
     {
       "path": "../signals"
     },
+    {
+      "path": "../stream-helpers"
+    },
     {
       "path": "../timebox"
     },