feat(rstream): add atom dep, add fromAtom() & docs

thi-ng · Jan 29, 2018 · ca3994a · ca3994a
1 parent 04c3d59
commit ca3994a
Show file tree

Hide file tree

Showing 3 changed files with 41 additions and 1 deletion.
diff --git a/packages/rstream/package.json b/packages/rstream/package.json
@@ -24,6 +24,7 @@
     },
     "dependencies": {
         "@thi.ng/api": "^1.3.0",
+        "@thi.ng/atom": "^0.0.1",
         "@thi.ng/transducers": "^1.0.0"
     },
     "keywords": [
@@ -40,4 +41,4 @@
     "publishConfig": {
         "access": "public"
     }
-}
+}
diff --git a/packages/rstream/src/from/atom.ts b/packages/rstream/src/from/atom.ts
@@ -0,0 +1,38 @@
+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 compares values with `===`.
+ * If `emitFirst` is true (default), also emits atom's
+ * current value when first subscriber attaches to stream.
+ *
+ * See: @thi.ng/atom
+ *
+ * ```
+ * db = new Atom({a: 23, b: 88});
+ * cursor = new Cursor(db, (s) => s.a, (s, x)=> ({...s, a: x}))
+ *
+ * rs.fromAtom(cursor).subscribe(rs.trace("cursor val:"))
+ * // cursor val: 23
+ *
+ * cursor.reset(42);
+ * // cursor val: 42
+ *
+ * db.reset({a: 66})
+ * // cursor val: 66
+ * ```
+ *
+ * @param atom
+ */
+export function fromAtom<T>(atom: ReadonlyAtom<T>, emitFirst = true): Stream<T> {
+    return new Stream<T>((stream) => {
+        atom.addWatch(stream.id, (_, prev, curr) => {
+            if (curr !== prev) {
+                stream.next(curr);
+            }
+        });
+        emitFirst && stream.next(atom.deref());
+        return () => atom.removeWatch(stream.id);
+    });
+}
diff --git a/packages/rstream/src/index.ts b/packages/rstream/src/index.ts
@@ -3,6 +3,7 @@ export * from "./stream";
 export * from "./stream-merge";
 export * from "./subscription";
 
+export * from "./from/atom";
 export * from "./from/event";
 export * from "./from/interval";
 export * from "./from/iterable";