feat(rstream): add fromView(), update fromAtom() docs, update re-exports

packages/rstream/src/from/atom.ts

@@ -3,14 +3,16 @@ import { ReadonlyAtom } from "@thi.ng/atom/api";
 import { Stream } from "../stream";
 
 /**
- * Yields stream of value changes in given atom / cursor.
- * Attaches watch to atom and checks for value changes with given `changed`
- * predicate (`!==` by default). If the predicate returns truthy result,
- * the atom change is emitted on the stream.
- * If `emitFirst` is true (default), also emits atom's current value
- * when first subscriber attaches to stream.
+ * Yields stream of value changes in given atom / cursor. Attaches watch
+ * to atom and checks for value changes with given `changed` predicate
+ * (`!==` by default). If the predicate returns truthy result, the new
+ * value is emitted on the stream. If `emitFirst` is true (default),
+ * also emits atom's current value when first subscriber attaches to
+ * stream.
  *
- * See: @thi.ng/atom
+ * See:
+ * - fromView()
+ * - @thi.ng/atom
  *
  * ```
  * db = new Atom({a: 23, b: 88});

packages/rstream/src/from/view.ts

@@ -0,0 +1,57 @@
+import { ReadonlyAtom, ViewTransform } from "@thi.ng/atom/api";
+import { View } from "@thi.ng/atom/view";
+import { Stream } from "../stream";
+import { Path } from "@thi.ng/paths";
+import { Predicate2 } from "@thi.ng/api/api";
+
+/**
+ * Similar to `fromAtom()`, but creates an eager derived view for a
+ * nested value in atom / cursor and yields stream of its value changes.
+ * Views are readonly versions of Cursors and more lightweight. The view
+ * checks for value changes with given `equiv` predicate
+ * (`@thi.ng/api/equiv` by default). If the predicate returns a falsy
+ * result, the new value is emitted on the stream. The first value
+ * emitted is always the (possibly transformed) current value at the
+ * stream's start time (i.e. when the first subscriber attaches).
+ *
+ * If the optional `tx` is given, the raw value is first passed to this
+ * transformer function and its result emitted on the stream.
+ *
+ * When the stream is cancelled the view is destroyed as well.
+ *
+ * See:
+ * - fromAtom()
+ * - @thi.ng/atom
+ *
+ * ```
+ * db = new Atom({a: 1, b: {c: 2}});
+ *
+ * fromView(db, "b.c", (x) => x != null ? x : "n/a").subscribe(trace("view:"))
+ * // view: 2
+ *
+ * db.swapIn("b.c", (x: number) => x + 1);
+ * // view: 3
+ *
+ * db.reset({a: 10});
+ * // view: n/a
+ * ```
+ *
+ * @param atom
+ * @param path
+ * @param tx
+ */
+export function fromView<T>(atom: ReadonlyAtom<any>, path: Path, tx?: ViewTransform<T>, equiv?: Predicate2<any>): Stream<T> {
+    return new Stream<T>((stream) => {
+        let isActive = true;
+        const view = new View<T>(
+            atom,
+            path,
+            tx ?
+                (x) => isActive && (x = tx(x), stream.next(x), x) :
+                (x) => isActive && (stream.next(x), x),
+            false,
+            equiv
+        );
+        return () => (isActive = false, view.release());
+    });
+}

packages/rstream/src/index.ts

@@ -11,6 +11,7 @@ export * from "./from/iterable";
 export * from "./from/promise";
 export * from "./from/promises";
 export * from "./from/raf";
+export * from "./from/view";
 export * from "./from/worker";
 
 export * from "./subs/bisect";