Merge pull request #109 from thefrontside/tm/streamp-helpers-subject

feat(stream-helpers): Added behavior subject

Author: taras

stream-helpers/README.md

@@ -192,6 +192,53 @@ function* exampleWithCloseValue() {
 }
 ```
 
+### Subject (like RxJS BehaviorSubject)
+
+The `createSubject` 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).
+
+```typescript
+import { createSubject } from "@effectionx/stream-helpers";
+import { createChannel, spawn } from "effection";
+
+function* example() {
+  const subject = createSubject<number>();
+  const channel = createChannel<number, void>();
+  const downstream = subject(channel);
+
+  // First subscriber
+  const sub1 = yield* downstream;
+
+  yield* channel.send(1);
+  yield* channel.send(2);
+
+  console.log(yield* sub1.next()); // { done: false, value: 1 }
+  console.log(yield* sub1.next()); // { done: false, value: 2 }
+
+  // Late subscriber gets the latest value immediately
+  const sub2 = yield* downstream;
+  console.log(yield* sub2.next()); // { done: false, value: 2 }
+}
+```
+
+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,
+);
+```
+
 ### Passthrough Tracker
 
 Passthrough Tracker stream helper provides a way to know if all items that

stream-helpers/deno.json

@@ -1,6 +1,6 @@
 {
   "name": "@effectionx/stream-helpers",
-  "version": "0.4.1",
+  "version": "0.5.0",
   "imports": {
     "effection": "npm:effection@^3",
     "immutable": "npm:immutable@^5",

stream-helpers/subject.test.ts

@@ -0,0 +1,94 @@
+import {
+  createChannel,
+  type Operation,
+  type Stream,
+  type Subscription,
+} from "effection";
+import { beforeEach, describe, it } from "@effectionx/bdd";
+import { expect } from "@std/expect";
+
+import { createSubject } from "./subject.ts";
+
+function* next<T, TClose>(
+  subscription: Subscription<T, TClose>,
+): Operation<T | TClose> {
+  const item = yield* subscription.next();
+  if (item.done) {
+    return item.value;
+  } else {
+    return item.value;
+  }
+}
+
+describe("subject", () => {
+  let subject = createSubject<number>();
+  let upstream = createChannel<number, string>();
+  let downstream: Stream<number, string>;
+
+  beforeEach(function* () {
+    subject = createSubject();
+
+    upstream = createChannel();
+
+    downstream = subject(upstream);
+  });
+
+  it("allows multiple subscribers", function* () {
+    const subscriber1 = yield* downstream;
+    const subscriber2 = yield* downstream;
+
+    yield* upstream.send(1);
+    yield* upstream.send(2);
+
+    // 1 multicast to both
+    expect(yield* next(subscriber1)).toEqual(1);
+    expect(yield* next(subscriber2)).toEqual(1);
+
+    // 2 multicast to both
+    expect(yield* next(subscriber1)).toEqual(2);
+    expect(yield* next(subscriber2)).toEqual(2);
+  });
+
+  it("each later subscribers get latest value", function* () {
+    const subscriber1 = yield* downstream;
+    yield* upstream.send(1);
+    expect(yield* next(subscriber1)).toEqual(1);
+
+    yield* upstream.send(2);
+    expect(yield* next(subscriber1)).toEqual(2);
+
+    const subscriber2 = yield* downstream;
+    expect(yield* next(subscriber2)).toEqual(2);
+  });
+
+  it("sends closing value to all subscribers", function* () {
+    const subscriber1 = yield* downstream;
+    const subscriber2 = yield* downstream;
+
+    yield* upstream.send(1);
+    yield* upstream.close("bye");
+
+    // 1 multicast to both
+    expect(yield* next(subscriber1)).toEqual(1);
+    expect(yield* next(subscriber2)).toEqual(1);
+
+    // 2 multicast to both
+    expect(yield* next(subscriber1)).toEqual("bye");
+    expect(yield* next(subscriber2)).toEqual("bye");
+  });
+
+  it("subscriber after close receives last value and close value", function* () {
+    const subscriber1 = yield* downstream;
+
+    yield* upstream.send(1);
+    yield* upstream.close("bye");
+
+    // First subscriber gets value and close
+    expect(yield* next(subscriber1)).toEqual(1);
+    expect(yield* next(subscriber1)).toEqual("bye");
+
+    // Late subscriber after close should get last value and close value
+    const subscriber2 = yield* downstream;
+    expect(yield* next(subscriber2)).toEqual("bye");
+  });
+});

stream-helpers/subject.ts

@@ -0,0 +1,65 @@
+import type { Stream, 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).
+ *
+ * @returns A function that takes a stream and returns a multicast stream
+ *
+ * @example
+ * ```ts
+ * const subject = createSubject<number>();
+ * const downstream = subject(upstream);
+ *
+ * const sub1 = yield* downstream; // subscribes to upstream
+ * 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,
+ * ]);
+ * ```
+ */
+export function createSubject<T>(): <TClose>(
+  stream: Stream<T, TClose>,
+) => Stream<T, TClose> {
+  let current: IteratorResult<T> | undefined = undefined;
+
+  return <TClose>(stream: Stream<T, TClose>) => ({
+    *[Symbol.iterator]() {
+      let upstream = yield* stream;
+
+      let iterator: Subscription<T, TClose> = current
+        ? {
+          *next() {
+            iterator = upstream;
+            return current!;
+          },
+        }
+        : {
+          *next() {
+            return current = yield* upstream.next();
+          },
+        };
+
+      return {
+        next: () => iterator.next(),
+      };
+    },
+  });
+}