Add pure and sequence

1 files changed, 79 insertions, 5 deletions

diff --git a/src/rx.ts b/src/rx.ts
index ea4f6a4..a83c915 100644
--- a/src/rx.ts
+++ b/src/rx.ts
@@ -112,6 +112,21 @@ export class Rx<A> {
   }
 }
 
+class Pure<A> extends Rx<A> {
+  readonly type: 'Pure'
+  readonly value: A
+
+  constructor(value: A) {
+    super()
+    this.type = 'Pure'
+    this.value = value
+  }
+}
+
+export function pure<A>(value: A): Rx<A> {
+  return new Pure(value)
+}
+
 class Var<A> extends Rx<A> {
   readonly type: 'Var'
   readonly id: string
@@ -151,6 +166,28 @@ class FlatMap<A, B> extends Rx<B> {
   }
 }
 
+export function sequence<A>(xs: Array<Rx<A>>): Sequence<A> {
+  return new Sequence<A>(xs)
+}
+
+export function sequence2<A>(xs: Array<Rx<A>>): Rx<Array<A>> {
+  return xs.reduce(
+    (acc: Rx<Array<A>>, x: Rx<A>) => acc.flatMap(ys => x.map(y => [y, ...ys])),
+    new Pure([])
+  )
+}
+
+class Sequence<A> extends Rx<Array<A>> {
+  readonly type: 'Sequence'
+  readonly xs: Array<Rx<A>>
+
+  constructor(xs: Array<Rx<A>>) {
+    super()
+    this.type = 'Sequence'
+    this.xs = xs
+  }
+}
+
 // Mount
 
 export function mount(html: Html): Cancelable {
@@ -222,8 +259,15 @@ const voidRemove = () => {}
 
 // Rx run
 
-function rxRun<A>(state: State, rx: Rx<A>, effect: (value: A) => void): Cancelable {
-  if (isVar(rx)) {
+function rxRun<A>(
+  state: State,
+  rx: Rx<A>,
+  effect: (value: A) => void
+): Cancelable {
+  if (isPure<A>(rx)) {
+    effect(rx.value)
+    return voidCancel
+  } else if (isVar(rx)) {
     const cancel = state.subscribe(rx, effect)
     effect(state.get(rx))
     return cancel
@@ -239,13 +283,39 @@ function rxRun<A>(state: State, rx: Rx<A>, effect: (value: A) => void): Cancelab
       cancel2()
       cancel1()
     }
+  } else if (isSequence(rx)) {
+    const cancels = Array(rx.xs.length).fill(voidCancel)
+    const xs = Array(rx.xs.length).fill(undefined)
+    let initEnded = false
+
+    rx.xs.forEach((rxChild, i) => {
+      cancels[i] = rxRun(
+        state,
+        rxChild,
+        (value: A) => {
+          xs[i] = value
+          if (initEnded) {
+            // @ts-ignore
+            effect(xs)
+          }
+        }
+      )
+    })
+    // @ts-ignore
+    effect(xs)
+    initEnded = true
+    return () => cancels.forEach(cancel => cancel())
   } else {
     throw new Error(`Unrecognized rx: ${rx}`)
   }
 }
 
 function isRx<A>(x: any): x is Rx<A> {
-  return x !== undefined && x.type !== undefined && (x.type === "Var" || x.type === "Map" || x.type === "FlatMap")
+  return x !== undefined && x.type !== undefined && (x.type === "Var" || x.type === "Map" || x.type === "FlatMap" || x.type === 'Sequence' || x.type === 'Pure')
+}
+
+function isPure<A>(x: any): x is Pure<A> {
+  return x.type === 'Pure'
 }
 
 function isVar<A>(x: any): x is Var<A> {
@@ -253,11 +323,15 @@ function isVar<A>(x: any): x is Var<A> {
 }
 
 function isMap<A, B>(x: any): x is Map<A, B> {
-    return x.type === "Map"
+  return x.type === "Map"
 }
 
 function isFlatMap<A, B>(x: any): x is FlatMap<A, B> {
-    return x.type === "FlatMap"
+  return x.type === "FlatMap"
+}
+
+function isSequence<A>(x: any): x is Sequence<A> {
+  return x.type === "Sequence"
 }
 
 // Append