Repository: mostjs/core
Pull request #77 (2017-06-16 01:11:15)

feat(multicast): adds multicast back to the core
It seems like we need to figure out if we are going with this direction (as this PR has it) and keeping `@most/multicast` published and on npm
or are we all leaning to just bringing it back and storing it with the combinators.

axefrog: Worth noting that the current version of multicast still targets the old version of Most. I imagine that would need to be fixed prior to this PR being accepted.
axefrog: Heh looks like I saw this earlier and forgot about it :D

Patch:
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/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/prelude/package.json b/packages/prelude/package.json
index 83dce9f..6528fd0 100644
--- a/packages/prelude/package.json
+++ b/packages/prelude/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@most/prelude",
-  "version": "1.5.3",
+  "version": "1.6.0",
   "description": "prelude",
   "main": "dist/index.js",
   "module": "dist/index.es.js",


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)
+  }
+}