Repository: mostjs/core
Pull request #84 (2017-06-28 11:55:06)

docs(combinator-template): Add combinator doc template
This adds a new page, "Writing docs", where we could put helpful information for how to write docs.  Initially, it just includes a template that can be cut and pasted into api.rst when writing docs for an operator.


Patch:
diff --git a/.gitignore b/.gitignore
index d41c673..a90c38a 100644
--- a/.gitignore
+++ b/.gitignore
@@ -2,7 +2,9 @@ node_modules/
 .idea/
 *.iml
 .DS_Store
+.nyc_output/
 experiments/
+coverage/
 dist/
 docs/_build
 docs/_build_html
diff --git a/docs/conf.py b/docs/conf.py
index 188f79a..8510a12 100644
--- a/docs/conf.py
+++ b/docs/conf.py
@@ -30,7 +30,7 @@
 # Add any Sphinx extension module names here, as strings. They can be
 # extensions coming with Sphinx (named 'sphinx.ext.*') or your custom
 # ones.
-extensions = []
+extensions = ['sphinx.ext.todo']
 
 # Add any paths that contain templates here, relative to this directory.
 templates_path = ['_templates']
@@ -74,7 +74,7 @@ exclude_patterns = ['_build', 'Thumbs.db', '.DS_Store']
 pygments_style = 'sphinx'
 
 # If true, `todo` and `todoList` produce output, else they produce nothing.
-todo_include_todos = False
+todo_include_todos = True
 
 
 # -- Options for HTML output ----------------------------------------------
diff --git a/docs/index.rst b/docs/index.rst
index ff2eeed..6d5151d 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -11,7 +11,7 @@
 
 ``@most/core`` is a toolkit for reactive programming.  It helps you write highly interactive apps by composing streams of events, and without many of the hazards of side effects and mutable shared state.
 
-It features an ultra-high performance, low overhead architecture, and a small but powerful set of operations for merging, filtering, and transforming event streams.t
+It features an ultra-high performance, low overhead architecture, and a small but powerful set of operations for merging, filtering, and transforming event streams.
 
 .. toctree::
    :maxdepth: 2
@@ -20,6 +20,7 @@ It features an ultra-high performance, low overhead architecture, and a small bu
    notation
    concepts
    api
+   writing-docs
 
 Indices and tables
 ==================
diff --git a/docs/writing-docs.rst b/docs/writing-docs.rst
new file mode 100644
index 0000000..4cb05d8
--- /dev/null
+++ b/docs/writing-docs.rst
@@ -0,0 +1,44 @@
+Writing Docs
+============
+
+Tips and templates for writing ``@most/core`` docs.
+
+Combinator template
+-------------------
+
+.. note::
+  Template for writing docs for an API function.
+
+  #. Copy and paste into appropriate enclosing doc (probably api.rst)
+  #. Complete each todo
+  #. Make sure all todos have been removed before pushing
+
+.. todo:: Anchor link: function name with leading underscore
+.. _name:
+
+.. todo:: Function name heading
+
+name
+^^^^
+
+.. todo:: Type signature
+.. code-block:: haskell
+
+  name :: (a -> b) -> Stream a -> Stream b
+
+.. todo:: Short description
+
+Short description of what it does. Usually just 1-3 sentences, but may be multiple paragraphs if needed.::
+
+.. todo:: Ascii diagram. For example:
+  stream:         -a-b-c-d->
+  map(f, stream): -f(a)-f(b)-f(c)-f(d)->
+
+.. todo:: *Optionally* add a longer explanation, or delete if not necessary.
+
+Add more explanation *only if necessary*. Otherwise, just delete this.
+
+.. todo:: Simple Javascript usage example
+.. code-block:: javascript
+
+  map(x => x + 1, stream)
diff --git a/packages/core/package.json b/packages/core/package.json
index ff3654b..8a938d4 100644
--- a/packages/core/package.json
+++ b/packages/core/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@most/core",
-  "version": "0.9.1",
+  "version": "0.10.0",
   "description": "Reactive programming with lean, functions-only, curried, tree-shakeable API",
   "typings": "type-definitions/most.d.ts",
   "main": "dist/mostCore.js",
@@ -12,7 +12,7 @@
   ],
   "scripts": {
     "test": "npm run test:lint && npm run test:unit && npm run test:flow",
-    "test:unit": "mocha -r buba/register --reporter dot --recursive 'test/**/*-test.js'",
+    "test:unit": "nyc --reporter=text-summary mocha -r buba/register --reporter dot --recursive 'test/**/*-test.js'",
     "test:lint": "eslint --fix --ignore-path ../../.eslintignore src test",
     "test:flow": "flow check",
     "build": "npm run build:dist && npm run build:min && npm run build:flow",
@@ -48,6 +48,7 @@
     "eslint": "^3.15.0",
     "flow-bin": "^0.47.0",
     "mocha": "^3.4.2",
+    "nyc": "^11.0.2",
     "rollup": "^0.43.0",
     "rollup-plugin-buble": "^0.15.0",
     "rollup-plugin-node-resolve": "^2.0.0",
@@ -55,9 +56,9 @@
     "uglify-js": "^2.7.5"
   },
   "dependencies": {
-    "@most/disposable": "^0.9.1",
-    "@most/prelude": "^1.5.3",
-    "@most/scheduler": "^0.9.1",
+    "@most/disposable": "^0.10.0",
+    "@most/prelude": "^1.6.1",
+    "@most/scheduler": "^0.10.0",
     "@most/types": "^0.9.1"
   }
 }
diff --git a/packages/core/src/combinator/multicast.js b/packages/core/src/combinator/multicast.js
new file mode 100644
index 0000000..7d2271a
--- /dev/null
+++ b/packages/core/src/combinator/multicast.js
@@ -0,0 +1,89 @@
+import { append, remove, findIndex } from '@most/prelude'
+import { disposeNone, disposeOnce } from '@most/disposable'
+import { tryEvent, tryEnd } from '../source/tryEvent'
+
+export const multicast = stream =>
+  stream instanceof Multicast ? stream : new Multicast(stream)
+
+class Multicast {
+  constructor (source) {
+    this.source = new MulticastSource(source)
+  }
+  run (sink, scheduler) {
+    return this.source.run(sink, scheduler)
+  }
+}
+
+export class MulticastSource {
+  constructor (source) {
+    this.source = source
+    this.sinks = []
+    this._disposable = disposeNone()
+  }
+
+  run (sink, scheduler) {
+    const n = this.add(sink)
+    if (n === 1) {
+      this._disposable = this.source.run(this, scheduler)
+    }
+    return disposeOnce(new MulticastDisposable(this, sink))
+  }
+
+  _dispose () {
+    const disposable = this._disposable
+    this._disposable = disposeNone()
+    return disposable.dispose()
+  }
+
+  add (sink) {
+    this.sinks = append(sink, this.sinks)
+    return this.sinks.length
+  }
+
+  remove (sink) {
+    const i = findIndex(sink, this.sinks)
+    // istanbul ignore next
+    if (i >= 0) {
+      this.sinks = remove(i, this.sinks)
+    }
+
+    return this.sinks.length
+  }
+
+  event (time, value) {
+    const s = this.sinks
+    if (s.length === 1) {
+      return s[0].event(time, value)
+    }
+    for (let i = 0; i < s.length; ++i) {
+      tryEvent(time, value, s[i])
+    }
+  }
+
+  end (time) {
+    const s = this.sinks
+    for (let i = 0; i < s.length; ++i) {
+      tryEnd(time, s[i])
+    }
+  }
+
+  error (time, err) {
+    const s = this.sinks
+    for (let i = 0; i < s.length; ++i) {
+      s[i].error(time, err)
+    }
+  }
+}
+
+export class MulticastDisposable {
+  constructor (source, sink) {
+    this.source = source
+    this.sink = sink
+  }
+
+  dispose () {
+    if (this.source.remove(this.sink) === 0) {
+      this.source._dispose()
+    }
+  }
+}
diff --git a/packages/core/src/index.js b/packages/core/src/index.js
index 72c3cdf..1b09531 100644
--- a/packages/core/src/index.js
+++ b/packages/core/src/index.js
@@ -164,6 +164,12 @@ import { recoverWith as _recoverWith, throwError } from './combinator/errors'
 export const recoverWith = curry2(_recoverWith)
 export { throwError }
 
+// -----------------------------------------------------------------------
+// Multicasting
+
+export { multicast, MulticastSource } from './combinator/multicast'
+
+// ----------------------------------------------------------------------
 import {
   propagateTask as _propagateTask,
   propagateEventTask as _propagateEventTask,
diff --git a/packages/core/src/index.js.flow b/packages/core/src/index.js.flow
index 0fda667..3d5b508 100644
--- a/packages/core/src/index.js.flow
+++ b/packages/core/src/index.js.flow
@@ -176,6 +176,8 @@ declare export function recoverWith <A, E:Error> (f: (E) => Stream<A>): (s: Stre
 
 declare export function throwError <E:Error> (e: Error): Stream<any>
 
+declare export function multicast <A> (s: Stream<A>): Stream<A>
+
 export type PropagateTask<A> = Task & {
   value: A,
   sink: Sink<A>,
diff --git a/packages/core/src/scheduler/runWithLocalTime.js b/packages/core/src/scheduler/runWithLocalTime.js
index 1c0e2e7..b2245d1 100644
--- a/packages/core/src/scheduler/runWithLocalTime.js
+++ b/packages/core/src/scheduler/runWithLocalTime.js
@@ -3,5 +3,18 @@
 import RelativeSink from '../sink/RelativeSink'
 import { schedulerRelativeTo } from '@most/scheduler'
 
+// Run a stream with its own localized clock
+// This transforms time from the provided scheduler's clock to a stream-local
+// clock (which starts at 0), and then *back* to the scheduler's clock before
+// propagating events to sink.  IOW, stream.run will see local times, and sink
+// will see scheduler times.
 export const runWithLocalTime = (origin, stream, sink, scheduler) =>
-  stream.run(new RelativeSink(origin, sink), schedulerRelativeTo(origin, scheduler))
+  stream.run(relativeSink(origin, sink), schedulerRelativeTo(origin, scheduler))
+
+// Accumulate offsets instead of nesting RelativeSinks, which can happen
+// with higher-order stream and combinators like continueWith when they're
+// applied recursively.
+export const relativeSink = (origin, sink) =>
+  sink instanceof RelativeSink
+    ? new RelativeSink(origin + sink.offset, sink.sink)
+    : new RelativeSink(origin, sink)
diff --git a/packages/core/type-definitions/combinator/multicast.d.ts b/packages/core/type-definitions/combinator/multicast.d.ts
new file mode 100644
index 0000000..0d6c006
--- /dev/null
+++ b/packages/core/type-definitions/combinator/multicast.d.ts
@@ -0,0 +1,3 @@
+import { Stream } from '@most/types';
+
+export function multicast<A>(s: Stream<A>): Stream<A>;
diff --git a/packages/core/type-definitions/most.d.ts b/packages/core/type-definitions/most.d.ts
index 254f722..bff4e33 100644
--- a/packages/core/type-definitions/most.d.ts
+++ b/packages/core/type-definitions/most.d.ts
@@ -13,6 +13,7 @@ export * from './combinator/limit';
 export * from './combinator/loop';
 export * from './combinator/merge';
 export * from './combinator/mergeConcurrently';
+export * from './combinator/multicast';
 export * from './combinator/promises';
 export * from './combinator/sample';
 export * from './combinator/scan';
diff --git a/packages/disposable/package.json b/packages/disposable/package.json
index 653d088..2edbbad 100644
--- a/packages/disposable/package.json
+++ b/packages/disposable/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@most/disposable",
-  "version": "0.9.1",
+  "version": "0.10.0",
   "description": "Reactive programming with lean, functions-only, curried, tree-shakeable API",
   "typings": "type-definitions/index.d.ts",
   "main": "dist/index.js",
@@ -12,7 +12,7 @@
   ],
   "scripts": {
     "test": "npm run test:lint && npm run test:unit && npm run test:flow",
-    "test:unit": "mocha -r buba/register --reporter dot --recursive 'test/**/*-test.js'",
+    "test:unit": "nyc --reporter=text-summary mocha -r buba/register --reporter dot --recursive 'test/**/*-test.js'",
     "test:lint": "eslint --fix --ignore-path ../../.eslintignore src test",
     "test:flow": "flow check",
     "build": "npm run build:dist && npm run build:flow",
@@ -47,6 +47,7 @@
     "eslint": "^3.15.0",
     "flow-bin": "^0.47.0",
     "mocha": "^3.4.2",
+    "nyc": "^11.0.2",
     "rollup": "^0.43.0",
     "rollup-plugin-buble": "^0.15.0",
     "rollup-plugin-node-resolve": "^2.0.0",
@@ -54,7 +55,7 @@
     "uglify-js": "^2.7.5"
   },
   "dependencies": {
-    "@most/prelude": "^1.5.3",
+    "@most/prelude": "^1.6.1",
     "@most/types": "^0.9.1"
   }
 }
diff --git a/packages/prelude/package.json b/packages/prelude/package.json
index 6528fd0..a622053 100644
--- a/packages/prelude/package.json
+++ b/packages/prelude/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@most/prelude",
-  "version": "1.6.0",
+  "version": "1.6.1",
   "description": "prelude",
   "main": "dist/index.js",
   "module": "dist/index.es.js",
@@ -17,7 +17,7 @@
     "prepublish": "npm run build",
     "test": "npm run test:lint && npm run test:unit && npm run test:flow",
     "test:lint": "eslint src test",
-    "test:unit": "nyc mocha -r buba/register",
+    "test:unit": "nyc --reporter=text-summary mocha -r buba/register --reporter dot",
     "test:flow": "flow check"
   },
   "repository": {
@@ -37,7 +37,7 @@
     "cpy-cli": "^1.0.1",
     "flow-bin": "^0.47.0",
     "mocha": "^3.4.2",
-    "nyc": "^10.3.2",
+    "nyc": "^11.0.2",
     "rollup": "^0.43.0",
     "rollup-plugin-buble": "^0.15.0"
   }
diff --git a/packages/scheduler/package.json b/packages/scheduler/package.json
index 5772c16..43ad396 100644
--- a/packages/scheduler/package.json
+++ b/packages/scheduler/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@most/scheduler",
-  "version": "0.9.1",
+  "version": "0.10.0",
   "description": "Reactive programming with lean, functions-only, curried, tree-shakeable API",
   "typings": "type-definitions/index.d.ts",
   "main": "dist/index.js",
@@ -12,7 +12,7 @@
   ],
   "scripts": {
     "test": "npm run test:lint && npm run test:unit && npm run test:flow",
-    "test:unit": "mocha -r buba/register --reporter dot --recursive 'test/**/*-test.js'",
+    "test:unit": "nyc --reporter=text-summary mocha -r buba/register --reporter dot --recursive 'test/**/*-test.js'",
     "test:lint": "eslint --fix --ignore-path ../../.eslintignore src test",
     "test:flow": "flow check",
     "build": "npm run build:dist && npm run build:min && npm run build:flow",
@@ -48,6 +48,7 @@
     "eslint": "^3.15.0",
     "flow-bin": "^0.47.0",
     "mocha": "^3.4.2",
+    "nyc": "^11.0.2",
     "rollup": "^0.43.0",
     "rollup-plugin-buble": "^0.15.0",
     "rollup-plugin-node-resolve": "^2.0.0",
@@ -55,7 +56,7 @@
     "uglify-js": "^2.7.5"
   },
   "dependencies": {
-    "@most/prelude": "^1.5.3",
+    "@most/prelude": "^1.6.1",
     "@most/types": "^0.9.1"
   }
 }


Test patch:
diff --git a/packages/core/test/combinator/multicast-test.js b/packages/core/test/combinator/multicast-test.js
new file mode 100644
index 0000000..43f3ca0
--- /dev/null
+++ b/packages/core/test/combinator/multicast-test.js
@@ -0,0 +1,196 @@
+import { describe, it } from 'mocha'
+import { assert, eq } from '@briancavalier/assert'
+import { spy } from 'sinon'
+
+import { now } from '../../src/source/now'
+import { map, tap } from '../../src/combinator/transform'
+import { multicast, MulticastSource, MulticastDisposable } from '../../src/combinator/multicast'
+import { runEffects } from '../../src/runEffects'
+import { ticks, makeEvents, collectEvents } from '../helper/testEnv'
+import { sinkSpy } from '../helper/sinkSpy'
+import FakeDisposeStream from '../helper/FakeDisposeStream'
+
+describe('multicast', () => {
+  it('should be identity for already-multicasted stream', () => {
+    const s = multicast(now(1))
+    eq(s, multicast(s))
+  })
+
+  it('should call mapper function once when there are > 1 observer', () => {
+    const observer1Spy = spy()
+    const observer2Spy = spy()
+    const observer3Spy = spy()
+    const scheduler = ticks(1)
+
+    const f = x => x + 1
+    const x = Math.random()
+    const s = now(x)
+
+    const mapperSpy = spy(f)
+    const mapped = map(mapperSpy, s)
+    const multicasted = multicast(mapped)
+
+    const o1 = runEffects(tap(observer1Spy, multicasted), scheduler)
+    const o2 = runEffects(tap(observer2Spy, multicasted), scheduler)
+    const o3 = runEffects(tap(observer3Spy, multicasted), scheduler)
+
+    return Promise.all([o1, o2, o3]).then(() => {
+      assert(mapperSpy.calledOnce)
+
+      const expected = f(x)
+
+      assert(observer1Spy.calledOnce)
+      eq(expected, observer1Spy.getCall(0).args[0])
+
+      assert(observer2Spy.calledOnce)
+      eq(expected, observer2Spy.getCall(0).args[0])
+
+      assert(observer3Spy.calledOnce)
+      eq(expected, observer3Spy.getCall(0).args[0])
+    })
+  })
+})
+
+const sentinel = { value: sentinel }
+
+describe('MulticastSource', () => {
+  it('should call producer on first observer', () => {
+    const eventSpy = spy()
+    const s = new MulticastSource({run: () => {}})
+    const scheduler = ticks(1)
+
+    s.run({event: eventSpy}, scheduler)
+    s.event(sentinel)
+
+    eq(true, eventSpy.calledOnce)
+  })
+
+  it('should call producer ONLY on the first observer', () => {
+    const sourceSpy = spy()
+    const s = new MulticastSource({run: sourceSpy})
+    const scheduler = ticks(1)
+
+    return Promise.all([
+      s.run({ event: () => {} }, scheduler),
+      s.run({ event: () => {} }, scheduler),
+      s.run({ event: () => {} }, scheduler),
+      s.run({ event: () => {} }, scheduler)
+    ]).then(() => {
+      eq(true, sourceSpy.calledOnce)
+    })
+  })
+
+  it('should publish events to all observers', () => {
+    const s = new MulticastSource(makeEvents(1, 2))
+    const scheduler = ticks(2)
+
+    return Promise.all([
+      collectEvents(s, scheduler),
+      collectEvents(s, scheduler)
+    ]).then(eq([
+      [{ time: 0, value: 0 }, { time: 1, value: 1 }],
+      [{ time: 0, value: 0 }, { time: 1, value: 1 }]
+    ]))
+  })
+
+  it('should propagate errors', () => {
+    const fakeSource = {
+      run () {}
+    }
+    const scheduler = ticks(1)
+
+    const ms = new MulticastSource(fakeSource)
+    const s1 = sinkSpy()
+    ms.run(s1, scheduler)
+
+    const e1 = new Error()
+    ms.error(1, e1)
+    eq(1, s1.errorCalled)
+    eq(1, s1.errorTime)
+    eq(e1, s1.errorValue)
+
+    const s2 = sinkSpy()
+    ms.run(s2, scheduler)
+
+    const e2 = new Error()
+    ms.error(2, e2)
+    eq(2, s1.errorCalled)
+    eq(2, s1.errorTime)
+    eq(e2, s1.errorValue)
+
+    eq(1, s2.errorCalled)
+    eq(2, s2.errorTime)
+    eq(e2, s2.errorValue)
+  })
+
+  it('should propagate errors only to errored observer', () => {
+    const s = new MulticastSource(makeEvents(1, 2))
+    const error = new Error()
+    const scheduler = ticks(2)
+
+    const p1 = runEffects(s, scheduler)
+    const p2 = runEffects(map(() => { throw error }, s), scheduler)
+        .catch(e => e)
+
+    return Promise.all([p1, p2]).then(([a, b]) => {
+      eq(undefined, a)
+      eq(error, b)
+    })
+  })
+
+  it('should call dispose if all observers disconnect', () => {
+    const disposer = spy()
+    const s = new MulticastSource(FakeDisposeStream.from(disposer, now(sentinel)))
+    const scheduler = ticks(1)
+
+    return Promise.all([runEffects(s, scheduler), runEffects(s, scheduler)]).then(() => {
+      eq(true, disposer.calledOnce)
+    })
+  })
+})
+
+describe('MulticastDisposable', () => {
+  it('should dispose when sinks reaches zero', () => {
+    const expectedSink = {}
+    const source = {
+      remove (sink) {
+        eq(expectedSink, sink)
+        return 0
+      },
+
+      _dispose () {}
+    }
+
+    const disposed = spy(source, '_dispose')
+    const removed = spy(source, 'remove')
+
+    const md = new MulticastDisposable(source, expectedSink)
+
+    md.dispose()
+
+    assert(removed.calledOnce)
+    assert(disposed.calledOnce)
+  })
+
+  it('should not dispose when sinks > 0', () => {
+    const expectedSink = {}
+    const source = {
+      remove (sink) {
+        eq(expectedSink, sink)
+        return 1
+      },
+
+      _dispose () {}
+    }
+
+    const disposed = spy(source, '_dispose')
+    const removed = spy(source, 'remove')
+
+    const md = new MulticastDisposable(source, expectedSink)
+
+    md.dispose()
+
+    assert(removed.calledOnce)
+    assert(disposed.notCalled)
+  })
+})
diff --git a/packages/core/test/helper/sinkSpy.js b/packages/core/test/helper/sinkSpy.js
new file mode 100644
index 0000000..5c2040a
--- /dev/null
+++ b/packages/core/test/helper/sinkSpy.js
@@ -0,0 +1,49 @@
+'use strict'
+
+export const sinkSpy = () => new SinkSpy(noop, noop, noop)
+
+export const eventErrorSinkSpy = e => new SinkSpy(() => { throw e }, noop, noop)
+
+export const endErrorSinkSpy = e => new SinkSpy(noop, () => { throw e }, noop)
+
+const noop = () => {}
+
+class SinkSpy {
+  constructor (event, end, error) {
+    this.eventCalled = 0
+    this.eventTime = NaN
+    this.eventValue = undefined
+    this._event = event
+
+    this.endCalled = 0
+    this.endTime = NaN
+    this.endValue = undefined
+    this._end = end
+
+    this.errorCalled = 0
+    this.errorTime = NaN
+    this.errorValue = undefined
+    this._error = error
+  }
+
+  event (t, x) {
+    this.eventCalled += 1
+    this.eventTime = t
+    this.eventValue = x
+    return this._event(t, x)
+  }
+
+  end (t, x) {
+    this.endCalled += 1
+    this.endTime = t
+    this.endValue = x
+    return this._end(t, x)
+  }
+
+  error (t, e) {
+    this.errorCalled += 1
+    this.errorTime = t
+    this.errorValue = e
+    return this._error(t, e)
+  }
+}