✨ Support initial value in createSubject()

stream-helpers/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@effectionx/stream-helpers",
   "description": "Type-safe stream operators like filter, map, reduce, and forEach",
-  "version": "0.7.3",
+  "version": "0.8.0",
   "type": "module",
   "main": "./dist/mod.js",
   "types": "./dist/mod.d.ts",

stream-helpers/subject.test.ts

@@ -76,6 +76,31 @@ describe("subject", () => {
     expect(yield* next(subscriber2)).toEqual("bye");
   });
 
+  it("yields initial value to first subscriber before any upstream values", function* () {
+    subject = createSubject(42);
+    downstream = subject(upstream);
+
+    const subscriber = yield* downstream;
+    expect(yield* next(subscriber)).toEqual(42);
+
+    yield* upstream.send(1);
+    expect(yield* next(subscriber)).toEqual(1);
+  });
+
+  it("yields upstream value to late subscriber once upstream has sent", function* () {
+    subject = createSubject(42);
+    downstream = subject(upstream);
+
+    const subscriber1 = yield* downstream;
+    expect(yield* next(subscriber1)).toEqual(42);
+
+    yield* upstream.send(1);
+    expect(yield* next(subscriber1)).toEqual(1);
+
+    const subscriber2 = yield* downstream;
+    expect(yield* next(subscriber2)).toEqual(1);
+  });
+
   it("subscriber after close receives last value and close value", function* () {
     const subscriber1 = yield* downstream;
 

stream-helpers/subject.ts

@@ -35,28 +35,33 @@ import type { Stream, Subscription } from "effection";
  * ]);
  * ```
  */
-export function createSubject<T>(): <TClose>(
-  stream: Stream<T, TClose>,
-) => Stream<T, TClose> {
-  let current: IteratorResult<T> | undefined = undefined;
+export function createSubject<T>(
+  initial?: T,
+): <TClose>(stream: Stream<T, TClose>) => 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;
 
+      let tracking: Subscription<T, TClose> = {
+        *next() {
+          current = yield* upstream.next();
+          return current;
+        },
+      };
+
       let iterator: Subscription<T, TClose> = current
         ? {
             *next() {
-              iterator = upstream;
+              iterator = tracking;
               return current!;
             },
           }
-        : {
-            *next() {
-              current = yield* upstream.next();
-              return current;
-            },
-          };
+        : tracking;
 
       return {
         next: () => iterator.next(),