feat(rstream): re-implement bisect() using PubSub, update tests

packages/rstream/src/subs/bisect.ts

@@ -1,17 +1,41 @@
 import { Predicate } from "@thi.ng/api/api";
 
 import { ISubscriber } from "../api";
-import { Subscription } from "../subscription";
+import { PubSub } from "../pubsub";
 
-export function bisect<T>(pred: Predicate<T>, a?: ISubscriber<T>, b?: ISubscriber<T>) {
-    return new Subscription<T, T>({
-        next(x) {
-            const sub = pred(x) ? a : b;
-            sub.next && sub.next(x);
-        },
-        done() {
-            a.done && a.done();
-            b.done && b.done();
-        }
-    });
+/**
+ * Returns a new `PubSub` instance using given predicate `pred` as
+ * boolean topic function and `a` & `b` as subscribers for truthy (`a`)
+ * and falsey `b` values.
+ *
+ * ```
+ * rs.fromIterable([1,2,3,4]).subscribe(
+ *   rs.bisect(
+ *     (x) => !(x & 1),
+ *     { next: (x) => console.log("even", x) },
+ *     { next: (x) => console.log("odd", x) }
+ *   )
+ * );
+ * ```
+ *
+ * If `a` or `b` need to be subscribed to directly, then `a` / `b` MUST
+ * be converted into a `Subscription` instance (if not already) and a
+ * reference kept prior to calling `bisect()`.
+ *
+ * ```
+ * const even = new rs.Subscription({next: (x) => console.log("even", x) });
+ * const odd = new rs.Subscription({next: (x) => console.log("odd", x) });
+ *
+ * rs.fromIterable([1,2,3,4]).subscribe(rs.bisect((x) => !(x & 1), even, odd));
+ * ```
+ *
+ * @param pred
+ * @param a
+ * @param b
+ */
+export function bisect<T>(pred: Predicate<T>, a?: ISubscriber<T>, b?: ISubscriber<T>): PubSub<T, T> {
+    const sub = new PubSub<T, T>({ topic: pred });
+    sub.subscribeTopic(true, a);
+    sub.subscribeTopic(false, b);
+    return sub;
 }

packages/rstream/test/bisect.ts

@@ -13,11 +13,12 @@ describe("bisect", () => {
     it("raw subscribers", (done) => {
         const odds = [], evens = [];
         src.subscribe(
-            rs.bisect((x) => !!(x & 1),
+            rs.bisect<number>((x) => !!(x & 1),
                 { next(x) { odds.push(x) } },
                 { next(x) { evens.push(x) } }
             )
-        ).subscribe({
+        );
+        src.subscribe({
             done() {
                 assert.deepEqual(odds, [1, 3]);
                 assert.deepEqual(evens, [2, 4]);
@@ -37,14 +38,14 @@ describe("bisect", () => {
             tx.map<number, number>(x => x * 100)
         );
         let doneCount = 0;
-        src.subscribe(rs.bisect((x) => !!(x & 1), subo, sube))
-            .subscribe({
-                done() {
-                    assert.deepEqual(odds, [10, 30]);
-                    assert.deepEqual(evens, [200, 400]);
-                    assert.equal(doneCount, 2);
-                    done();
-                }
-            });
+        src.subscribe(rs.bisect((x) => !!(x & 1), subo, sube));
+        src.subscribe({
+            done() {
+                assert.deepEqual(odds, [10, 30]);
+                assert.deepEqual(evens, [200, 400]);
+                assert.equal(doneCount, 2);
+                done();
+            }
+        });
     });
 });