feat: implement state

packages/stream/src/combinators/at.ts

@@ -0,0 +1,5 @@
+/**
+ * Creates a stream that emits a value after a given amount of time.
+ * @name at<A>(time: Time, value: A): Stream<A>
+ */
+export { at } from '@most/core'

packages/stream/src/combinators/index.ts

@@ -1,4 +1,5 @@
 export * from './ap'
+export * from './at'
 export * from './awaitPromises'
 export * from './chain'
 export * from './combine'
@@ -33,6 +34,7 @@ export * from './skipRepeatsWith'
 export * from './skipWhile'
 export * from './slice'
 export * from './startWith'
+export * from './state'
 export * from './switchCombine'
 export * from './switchMap'
 export * from './switchMerge'

packages/stream/src/combinators/state.test.ts

@@ -0,0 +1,25 @@
+import { Test, describe, given, it } from '@typed/test'
+import { at, mergeArray, observe, state } from './'
+
+import { Stream } from '@motorcycle/types'
+import { add } from '167'
+
+export const test: Test = describe(`state`, [
+  given(`(a -> b -> a) -> Stream a -> Stream b`, [
+    it(`returns Stream a`, ({ equal }) => {
+      const a$ = mergeArray([at(0, 0), at(100, 100), at(200, 200)])
+      const b$ = mergeArray([at(1, 1), at(2, 2)])
+
+      const expected = [0, 1, 3, 100, 101, 103, 200, 201, 203]
+      const sut = state(add, a$, b$)
+
+      return collectEvents(sut).then(equal(expected))
+    }),
+  ]),
+])
+
+function collectEvents<A>(stream: Stream<A>): Promise<Array<A>> {
+  const events: Array<A> = []
+
+  return observe(x => events.push(x), stream).then(() => events)
+}

packages/stream/src/combinators/state.ts

@@ -0,0 +1,53 @@
+import { Stream } from '@motorcycle/types'
+import { curry3 } from '167'
+import { scan } from './scan'
+import { switchMap } from './switchMap'
+
+/**
+ * Especially useful when keeping local state that also needs to be updated
+ * from an outside source.
+ * @name state<A, B>(f: (acc: A, value: B) => A, seed$: Stream<A>, values$: Stream<B>): Stream<A>
+ * @example 
+ * import { Stream } from '@motorcycle/types'
+ * import { query, dragOverEvent, dragstartEvent, dropEvent } from '@motorcycle/dom'
+ * import { sample, map, state, mapList } from '@motorcycle/stream'
+ * import { move } from '@typed/prelude'
+ * 
+ * export function ReorderableList(sources) {
+ *   const { list$, dom } = sources
+ *   const li = query('li', dom)
+ *   const dragOver$ = dragOverEvent(li)
+ *   const dragStart$ = dragstartEvent(li)
+ *   const drop$ = dropEvent(li)
+ *   const reducer$: Stream<(list: Array<string>) => Array<string>> = 
+ *     sample((to, from) => move(from, to), map(getKey, drop$), map(getKey, dragStart$))
+ *   const reorderedList$ = state((x, f) => f(x), list$, reducer$)
+ *   // create all of our <li> tags
+ *   const childViews$ = mapList(listItem, reorderedList$)
+ *   // create our <ul> containgin our <li> tags
+ *   const view$ = map(view, childViews$)
+ *
+ *   return {
+ *     view$,
+ *     preventDefault$: dragOver$,
+ *   }
+ * }
+ */
+export const state: State = curry3(__state)
+
+function __state<A, B>(
+  f: (accumulator: B, value: A) => B,
+  seed$: Stream<B>,
+  values$: Stream<A>
+): Stream<B> {
+  return switchMap(seed => scan(f, seed, values$), seed$)
+}
+
+export interface State {
+  <A, B>(f: (accumulator: A, value: B) => A, seed$: Stream<A>, values$: Stream<B>): Stream<A>
+  <A, B>(f: (accumulator: A, value: B) => A, seed$: Stream<A>): (values$: Stream<B>) => Stream<A>
+  <A, B>(f: (accumulator: A, value: B) => A): {
+    (seed$: Stream<A>, values$: Stream<B>): Stream<A>
+    (seed$: Stream<A>): (values$: Stream<B>) => Stream<A>
+  }
+}