[rstream] error handling, subscribe/transform cleanups

examples/adaptive-threshold/src/events.ts

@@ -61,7 +61,7 @@ export const defHandler = <E extends EventType>(
         error: console.warn,
     };
     return xform
-        ? eventProc.subscribeTopic(id, {}, {}).subscribe(sub, xform)
+        ? eventProc.subscribeTopic(id, sub, { xform })
         : eventProc.subscribeTopic(id, sub);
 };
 

examples/adaptive-threshold/src/index.ts

@@ -1,6 +1,5 @@
-import { peek } from "@thi.ng/arrays";
 import type { PackedBuffer } from "@thi.ng/pixel";
-import { fromRAF, sidechainPartition } from "@thi.ng/rstream";
+import { sidechainPartitionRAF } from "@thi.ng/rstream";
 import { map } from "@thi.ng/transducers";
 import { updateDOM } from "@thi.ng/transducers-hdom";
 import {
@@ -113,6 +112,4 @@ const app = (state: AppState) => {
 // sidechain to buffer intra-frame state updates. then only passes the
 // most recent one to `app()` and its resulting UI tree to the
 // `updateDOM()` transducer
-state
-    .subscribe(sidechainPartition<AppState, number>(fromRAF()))
-    .transform(map(peek), map(app), updateDOM());
+sidechainPartitionRAF(state).transform(map(app), updateDOM());

examples/bitmap-font/src/index.ts

@@ -1,7 +1,7 @@
 import type { IObjectOf } from "@thi.ng/api";
 import { dropdown } from "@thi.ng/hdom-components";
 import { clamp } from "@thi.ng/math";
-import { reactive, Stream, Subscription, sync } from "@thi.ng/rstream";
+import { ISubscriber, reactive, Stream, sync } from "@thi.ng/rstream";
 import {
     comp,
     map,
@@ -17,7 +17,7 @@ import { bits } from "@thi.ng/transducers-binary";
 import { updateDOM } from "@thi.ng/transducers-hdom";
 import { FONT } from "./font";
 
-const emitOnStream = (stream: Subscription<any, any>) => (e: Event) =>
+const emitOnStream = (stream: ISubscriber<any>) => (e: Event) =>
     stream.next((<HTMLSelectElement>e.target).value);
 
 // retrieve font bytes for given char

examples/canvas-dial/src/dial.ts

@@ -3,7 +3,7 @@ import { peek } from "@thi.ng/arrays";
 import { isString } from "@thi.ng/checks";
 import { canvas2D } from "@thi.ng/hdom-components";
 import { fitClamped } from "@thi.ng/math";
-import type { Subscription } from "@thi.ng/rstream";
+import type { ISubscription } from "@thi.ng/rstream";
 import { GestureEvent, gestureStream } from "@thi.ng/rstream-gestures";
 import { heading, sub2 } from "@thi.ng/vectors";
 
@@ -128,7 +128,7 @@ export const dial = (_opts: Partial<DialOpts>) => {
         font: "10px sans-serif",
         ..._opts,
     };
-    let events: Subscription<any, GestureEvent>;
+    let events: ISubscription<any, GestureEvent>;
     let cx: number, cy: number;
     const startTheta = opts.base + opts.gap / 2;
 

examples/crypto-chart/src/index.ts

@@ -14,10 +14,10 @@ import { resolve } from "@thi.ng/resolve-map";
 import {
     fromEvent,
     fromInterval,
+    ISubscriber,
     reactive,
     resolve as resolvePromise,
     stream,
-    Subscription,
     sync,
     trace,
 } from "@thi.ng/rstream";
@@ -173,11 +173,11 @@ const API_URL = (market: string, symbol: string, period: number) =>
 // stub for local testing
 // const API_URL = (..._) => `ohlc.json`;
 
-const emitOnStream = (stream: Subscription<any, any>) => (e: Event) =>
+const emitOnStream = (stream: ISubscriber<any>) => (e: Event) =>
     stream.next((<HTMLInputElement>e.target).value);
 
 const menu = (
-    stream: Subscription<any, any>,
+    stream: ISubscriber<any>,
     title: string,
     items: DropDownOption[]
 ) =>

examples/fft-synth/src/index.ts

@@ -3,9 +3,8 @@ import { fit, fitClamped } from "@thi.ng/math";
 import {
     fromAtom,
     fromDOMEvent,
-    fromRAF,
     merge,
-    sidechainPartition,
+    sidechainPartitionRAF,
     sync,
 } from "@thi.ng/rstream";
 import { gestureStream } from "@thi.ng/rstream-gestures";
@@ -82,7 +81,8 @@ const app = () => {
     };
 };
 
-main.subscribe(sidechainPartition<any, number>(fromRAF())).transform(
-    map(app()),
-    updateDOM()
-);
+// subscription & transformation of app state stream. uses a RAF
+// sidechain to buffer intra-frame state updates. then only passes the
+// most recent one to `app()` and its resulting UI tree to the
+// `updateDOM()` transducer
+sidechainPartitionRAF(main).transform(map(app()), updateDOM());

examples/hdom-benchmark/src/index.ts

@@ -72,13 +72,15 @@ const fpsCounter = (
                 );
             },
         },
-        // stream transducer to compute the windowed moving avarage
-        comp(
-            benchmark(),
-            movingAverage(period),
-            map((x) => 1000 / x),
-            partition(width, 1, true)
-        )
+        {
+            // stream transducer to compute the windowed moving avarage
+            xform: comp(
+                benchmark(),
+                movingAverage(period),
+                map((x) => 1000 / x),
+                partition(width, 1, true)
+            ),
+        }
     );
     return [
         {

examples/hdom-canvas-shapes/src/index.ts

@@ -7,7 +7,7 @@ import { COMMENT, serialize } from "@thi.ng/hiccup";
 import { convertTree, svg } from "@thi.ng/hiccup-svg";
 import { sincos } from "@thi.ng/math";
 import { concat, skewX23, translation23 } from "@thi.ng/matrices";
-import { fromRAF, stream, Subscription, sync } from "@thi.ng/rstream";
+import { fromRAF, ISubscriber, stream, sync } from "@thi.ng/rstream";
 import { map, range, repeatedly } from "@thi.ng/transducers";
 import { updateDOM } from "@thi.ng/transducers-hdom";
 import { addN } from "@thi.ng/vectors";
@@ -297,7 +297,7 @@ const TESTS: any = {
 };
 
 // test case selection dropdown
-const choices = (_: any, target: Subscription<string, any>, id: string) => [
+const choices = (_: any, target: ISubscriber<string>, id: string) => [
     dropdown,
     {
         class: "w4 ma2",

examples/imgui/src/index.ts

@@ -33,9 +33,8 @@ import { setInManyUnsafe } from "@thi.ng/paths";
 import {
     fromAtom,
     fromDOMEvent,
-    fromRAF,
     merge,
-    sidechainPartition,
+    sidechainPartitionRAF,
     sync,
 } from "@thi.ng/rstream";
 import { gestureStream } from "@thi.ng/rstream-gestures";
@@ -552,10 +551,8 @@ const main = sync({
     },
 });
 
-// transform the stream:
-main
-    // group potentially higher frequency event updates & sync with RAF
-    // to avoid extraneous real DOM/Canvas updates
-    .subscribe(sidechainPartition<any, number>(fromRAF()))
-    // then apply main compoment function & apply hdom
-    .transform(map(app()), updateDOM());
+// subscription & transformation of app state stream. uses a RAF
+// sidechain to buffer intra-frame state updates. then only passes the
+// most recent one to `app()` and its resulting UI tree to the
+// `updateDOM()` transducer
+sidechainPartitionRAF(main).transform(map(app()), updateDOM());

examples/rdom-lissajous/src/index.ts

@@ -6,15 +6,14 @@ import { $canvas } from "@thi.ng/rdom-canvas";
 import {
     fromDOMEvent,
     fromRAF,
-    ISubscribable,
+    ISubscription,
     reactive,
-    Subscription,
     sync,
 } from "@thi.ng/rstream";
 import { map, slidingWindow } from "@thi.ng/transducers";
 
 const slider = (
-    dest: Subscription<number, number>,
+    dest: ISubscription<number, number>,
     desc: string,
     tooltip: string,
     attribs?: any
@@ -60,7 +59,7 @@ size.next(<any>null);
 
 // combine various reactive parameters
 // and transform via transducers
-const dots: ISubscribable<any[]> = sync({
+const dots: ISubscription<any, any[]> = sync({
     src: { a, b, scale, size, time: fromRAF() },
 }).transform(
     // compute next lissajous point

examples/rdom-search-docs/src/index.ts

@@ -1,7 +1,7 @@
 import { timed } from "@thi.ng/bench";
 import { anchor, div, inputText } from "@thi.ng/hiccup-html";
 import { $compile, $list, $text, Component, IComponent } from "@thi.ng/rdom";
-import { debounce, reactive, Stream, Subscription } from "@thi.ng/rstream";
+import { debounce, ISubscription, reactive, Stream } from "@thi.ng/rstream";
 import { map } from "@thi.ng/transducers";
 // @ts-ignore
 import { deserialize } from "@ygoe/msgpack";
@@ -20,7 +20,7 @@ class DocSearch extends Component {
     inner!: IComponent;
     pager!: Pagination<string[][]>;
     query!: Stream<string>;
-    queryResults!: Subscription<string, string[][]>;
+    queryResults!: ISubscription<string, string[][]>;
 
     updateQuery(e: InputEvent) {
         const term = (<HTMLInputElement>e.target).value;

examples/rdom-search-docs/src/pagination.ts

@@ -2,15 +2,15 @@ import type { IRelease } from "@thi.ng/api";
 import { equiv } from "@thi.ng/equiv";
 import { button, div } from "@thi.ng/hiccup-html";
 import { clamp } from "@thi.ng/math";
-import { reactive, Stream, Subscription, sync } from "@thi.ng/rstream";
+import { ISubscription, reactive, Stream, sync } from "@thi.ng/rstream";
 import { comp, dedupe, map, page } from "@thi.ng/transducers";
 
 export class Pagination<T extends any[]> implements IRelease {
     page: Stream<number>;
-    maxPage: Subscription<T, number>;
-    resultPage: Subscription<any, T>;
+    maxPage: ISubscription<T, number>;
+    resultPage: ISubscription<any, T>;
 
-    constructor(src: Subscription<any, T>, pageSize: number) {
+    constructor(src: ISubscription<any, T>, pageSize: number) {
         this.page = reactive(0);
         this.maxPage = src.transform(map((res) => ~~(res.length / pageSize)));
         // produce search result page using `page()` transducer

examples/rstream-event-loop/src/events.ts

@@ -50,9 +50,7 @@ export const defHandler = <E extends EventType>(
         next: <Fn<Event, void>>handler,
         error: console.warn,
     };
-    return xform
-        ? eventProc.subscribeTopic(id, {}, {}).subscribe(sub, xform)
-        : eventProc.subscribeTopic(id, sub);
+    return eventProc.subscribeTopic(id, sub, { xform });
 };
 
 /**

examples/rstream-event-loop/src/index.ts

@@ -1,5 +1,4 @@
-import { peek } from "@thi.ng/arrays";
-import { fromRAF, sidechainPartition } from "@thi.ng/rstream";
+import { sidechainPartitionRAF } from "@thi.ng/rstream";
 import { map } from "@thi.ng/transducers";
 import { updateDOM } from "@thi.ng/transducers-hdom";
 import { AppState, NEXT, PREV } from "./api";
@@ -24,10 +23,38 @@ const app = ({ pageID, isLoading }: AppState) =>
               // navigation buttons w/ event dispatch
               [
                   "div",
-                  ["button", { onclick: () => dispatch([PREV, 5]) }, "<<"],
-                  ["button", { onclick: () => dispatch([PREV, 1]) }, "<"],
-                  ["button", { onclick: () => dispatch([NEXT, 1]) }, ">"],
-                  ["button", { onclick: () => dispatch([NEXT, 5]) }, ">>"],
+                  [
+                      "button",
+                      {
+                          disabled: pageID < 5,
+                          onclick: () => dispatch([PREV, 5]),
+                      },
+                      "<<",
+                  ],
+                  [
+                      "button",
+                      {
+                          disabled: pageID === 0,
+                          onclick: () => dispatch([PREV, 1]),
+                      },
+                      "<",
+                  ],
+                  [
+                      "button",
+                      {
+                          disabled: pageID === 19,
+                          onclick: () => dispatch([NEXT, 1]),
+                      },
+                      ">",
+                  ],
+                  [
+                      "button",
+                      {
+                          disabled: pageID >= 15,
+                          onclick: () => dispatch([NEXT, 5]),
+                      },
+                      ">>",
+                  ],
               ],
               // only here to show timestamp of last DOM update
               ["div.mt3", new Date().toString()],
@@ -45,6 +72,4 @@ const page = (_: any, pageID: number) => ["h1", `Page: ${pageID}`];
 // sidechain to buffer intra-frame state updates. then only passes the
 // most recent one to `app()` and its resulting UI tree to the
 // `updateDOM()` transducer
-state
-    .subscribe(sidechainPartition<AppState, number>(fromRAF()))
-    .transform(map(peek), map(app), updateDOM());
+sidechainPartitionRAF(state).transform(map(app), updateDOM());

examples/rstream-hdom/src/index.ts

@@ -1,9 +1,7 @@
-import { peek } from "@thi.ng/arrays";
 import {
-    fromRAF,
     ISubscribable,
-    sidechainPartition,
-    Subscription,
+    ISubscriber,
+    sidechainPartitionRAF,
     subscription,
     sync,
 } from "@thi.ng/rstream";
@@ -47,9 +45,7 @@ const ctx = {
  * @param ctx user context object
  */
 const domUpdate = (root: HTMLElement, tree: ISubscribable<any>, ctx?: any) =>
-    tree
-        .subscribe(sidechainPartition<any, number>(fromRAF()))
-        .transform(map(peek), updateDOM({ root, ctx }));
+    sidechainPartitionRAF(tree).transform(updateDOM({ root, ctx }));
 
 /**
  * Generic button component.
@@ -70,7 +66,7 @@ const button = (ctx: any, onclick: EventListener, body: any) => [
  * @param _ hdom user context (unused)
  * @param stream counter stream
  */
-const clickButton = (_: any, stream: Subscription<boolean, number>) => [
+const clickButton = (_: any, stream: ISubscriber<boolean>) => [
     button,
     () => stream.next(true),
     stream.deref(),
@@ -82,7 +78,7 @@ const clickButton = (_: any, stream: Subscription<boolean, number>) => [
  * @param _ hdom user context (unused)
  * @param counters streams to reset
  */
-const resetButton = (_: any, counters: Subscription<boolean, number>[]) => [
+const resetButton = (_: any, counters: ISubscriber<boolean>[]) => [
     button,
     () => counters.forEach((c) => c.next(false)),
     "reset",

examples/webgl-msdf/src/index.ts

@@ -4,7 +4,7 @@ import { canvasWebGL } from "@thi.ng/hdom-components";
 import { fitClamped } from "@thi.ng/math";
 import { concat, lookAt, perspective, transform44 } from "@thi.ng/matrices";
 import { SYSTEM } from "@thi.ng/random";
-import { fromDOMEvent, Subscription } from "@thi.ng/rstream";
+import { fromDOMEvent, ISubscription } from "@thi.ng/rstream";
 import {
     $w,
     add,
@@ -219,7 +219,7 @@ const app = () => {
     const glyphs = convertGlyphs(GLYPHS);
     let stars: ModelSpec;
     let body: ModelSpec;
-    let mouse: Subscription<any, ReadonlyVec>;
+    let mouse: ISubscription<any, ReadonlyVec>;
     let bg = 0;
     const canvas = canvasWebGL({
         init: async (el, gl) => {

examples/xml-converter/src/ui.ts

@@ -1,4 +1,4 @@
-import type { Subscription } from "@thi.ng/rstream";
+import type { ISubscriber } from "@thi.ng/rstream";
 import { mapIndexed } from "@thi.ng/transducers";
 import { handleTab } from "./utils";
 
@@ -143,7 +143,7 @@ const iconButton = (
 const copyButton = (
     { copyButton }: any,
     attribs: any,
-    stream: Subscription<boolean, boolean>,
+    stream: ISubscriber<boolean>,
     text: string,
     delay = 500
 ) => [