Repository: mostjs/core
Pull request #91 (2017-07-13 12:00:19)

adds MulticastSource typings

briancavalier: The failing tests are due to test fakes [here](https://github.com/mostjs/core/blob/master/packages/core/test/combinator/multicast-test.js#L161) and [here](https://github.com/mostjs/core/blob/master/packages/core/test/combinator/multicast-test.js#L183) still having `._dispose` rather than `.dispose`.  I wonder if it'd better to switch the tests to using real MulticastSources instead of maintaining fakes.  Thoughts?
TylorS: Well we're already exporting `MulticastSource` so I think it'd be okay to switch the tests to use the real thing.
briancavalier: > Well we're already exporting MulticastSource so I think it'd be okay to switch the tests to use the real thing.

Sounds good to me :+1:

Patch:
diff --git a/docs/api.rst b/docs/api.rst
index c879e9b..0a51618 100644
--- a/docs/api.rst
+++ b/docs/api.rst
@@ -279,7 +279,7 @@ slice
 
   slice :: int -> int -> Stream a -> Stream a
 
-Create a new stream containing only events where start <= index < end, where index is the ordinal index of an event in stream.::
+Keep only events in a range, where start <= index < end, and index is the ordinal index of an event in stream.::
 
   stream:              -a-b-c-d-e-f->
   slice(1, 4, stream): ---b-c-d|
@@ -296,7 +296,7 @@ take
 
   take :: int -> Stream a -> Stream a
 
-Create a new stream containing at most n events from stream.::
+Keep at most the first n events from stream.::
 
   stream:          -a-b-c-d-e-f->
   take(3, stream): -a-b-c|
@@ -315,7 +315,7 @@ skip
 
   skip :: int -> Stream a -> Stream a
 
-Create a new stream that omits the first n events from stream.::
+Discard the first n events from stream.::
 
   stream:          -a-b-c-d-e-f->
   skip(3, stream): -------d-e-f->
@@ -328,17 +328,113 @@ Create a new stream that omits the first n events from stream.::
 
 If stream contains fewer than n events, the returned stream will be empty.
 
-takeWhile :: (a -> bool) -> Stream a -> Stream a
+takeWhile
+^^^^^^^^^
+
+.. code-block:: haskell
+
+  takeWhile :: (a -> bool) -> Stream a -> Stream a
+
+Keep all events until predicate returns false, and discard the rest.::
+
+  stream:                  -2-4-5-6-8->
+  takeWhile(even, stream): -2-4-|
+
+.. _skipWhile:
+
+skipWhile
+^^^^^^^^^
+
+.. code-block:: haskell
+
+  skipWhile :: (a -> bool) -> Stream a -> Stream a
+
+Discard all events until predicate returns false, and keep the rest.::
+
+  stream:                  -2-4-5-6-8->
+  skipWhile(even, stream): -----5-6-8->
+
+.. _skipAfter:
+
+skipAfter
+^^^^^^^^^
+
+.. code-block:: haskell
+
+  skipAfter :: (a -> bool) -> Stream a -> Stream a
+
+Discard all events after the first event for which predicate returns true.::
+
+  stream:                  -1-2-3-4-5-6-8->
+  skipAfter(even, stream): -1-2|
+
+until
+^^^^^
+
+.. code-block:: haskell
 
-skipWhile :: (a -> bool) -> Stream a -> Stream a
+  until :: Stream * -> Stream a -> Stream a
 
-skipAfter :: (a -> bool) -> Stream a -> Stream a
+Keep all events in one stream until the first event occurs in another.::
 
-until :: Stream * -> Stream a -> Stream a
+  stream:                   -a-b-c-d-e-f->
+  endSignal:                ------z->
+  until(endSignal, stream): -a-b-c|
+
+Note that if endSignal has no events, then the returned stream will be effectively equivalent to the original.
+
+.. code-block:: javascript
+
+  // Keep only 3 seconds of events, discard the rest
+  until(at(3000, null), stream)
+
+since
+^^^^^
+
+.. code-block:: haskell
+
+  since :: Stream * -> Stream a -> Stream a
+
+Discard all events in one stream until the first event occurs in another.::
+
+  stream:                     -a-b-c-d-e-f->
+  startSignal:                ------z->
+  since(startSignal, stream): -------d-e-f->
+
+Note that if startSignal is has no events, then the returned stream will be effectively equivalent to :ref:`never`.
+
+.. code-block:: javascript
+
+  // Discard events for 3 seconds, keep the rest
+  since(at(3000, null), stream)
+
+during
+^^^^^^
+
+.. code-block:: haskell
+
+  during :: Stream (Stream *) -> Stream a -> Stream a
+
+Keep events that occur during a time window defined by a higher-order stream.::
+
+  stream:                     -a-b-c-d-e-f-g->
+  timeWindow:                 -----s
+  s:                                -----x
+  during(timeWindow, stream): -----c-d-e-|
+
+This is similar to :ref:`slice`, but uses time rather than indices to "slice" the stream.
+
+.. code-block:: javascript
 
-since :: Stream * -> Stream a -> Stream a
+  // A time window that:
+  // 1. starts at time = 1 second
+  // 2. ends at time = 6 seconds (1 second + 5 seconds)
+  const timeWindow = at(1000, at(5000, null))
 
-during :: Stream (Stream *) -> Stream a -> Stream a
+  // 1. discard events for 1 second, then
+  // 2. keep events for 5 more seconds, then
+  // 3. discard all subsequent events
+  during(timeWindow, stream)
 
 delay :: int -> Stream a -> Stream a
 
diff --git a/packages/core/src/combinator/multicast.js b/packages/core/src/combinator/multicast.js
index 7d2271a..0faee9c 100644
--- a/packages/core/src/combinator/multicast.js
+++ b/packages/core/src/combinator/multicast.js
@@ -1,6 +1,6 @@
-import { append, remove, findIndex } from '@most/prelude'
+import { append, findIndex, remove } from '@most/prelude'
 import { disposeNone, disposeOnce } from '@most/disposable'
-import { tryEvent, tryEnd } from '../source/tryEvent'
+import { tryEnd, tryEvent } from '../source/tryEvent'
 
 export const multicast = stream =>
   stream instanceof Multicast ? stream : new Multicast(stream)
@@ -18,20 +18,20 @@ export class MulticastSource {
   constructor (source) {
     this.source = source
     this.sinks = []
-    this._disposable = disposeNone()
+    this.disposable = disposeNone()
   }
 
   run (sink, scheduler) {
     const n = this.add(sink)
     if (n === 1) {
-      this._disposable = this.source.run(this, scheduler)
+      this.disposable = this.source.run(this, scheduler)
     }
     return disposeOnce(new MulticastDisposable(this, sink))
   }
 
-  _dispose () {
-    const disposable = this._disposable
-    this._disposable = disposeNone()
+  dispose () {
+    const disposable = this.disposable
+    this.disposable = disposeNone()
     return disposable.dispose()
   }
 
@@ -83,7 +83,7 @@ export class MulticastDisposable {
 
   dispose () {
     if (this.source.remove(this.sink) === 0) {
-      this.source._dispose()
+      this.source.dispose()
     }
   }
 }
diff --git a/packages/core/src/index.js.flow b/packages/core/src/index.js.flow
index 3d5b508..e23a5bf 100644
--- a/packages/core/src/index.js.flow
+++ b/packages/core/src/index.js.flow
@@ -178,6 +178,25 @@ declare export function throwError <E:Error> (e: Error): Stream<any>
 
 declare export function multicast <A> (s: Stream<A>): Stream<A>
 
+declare export class MulticastSource <A> {
+  source: Stream<A>;
+  sinks: Array<Sink<A>>;
+  disposable: Disposable;
+
+  constructor(Stream<A>): MulticastSource<A>;
+
+  run(Sink<A>, Scheduler): Disposable;
+
+  event(Time, A): void;
+  error(Time, Error): void;
+  end(Time): void;
+
+  add(Sink<A>): number;
+  remove(Sink<A>): number;
+
+  dispose(): void;
+}
+
 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
index 0d6c006..8ed1d5b 100644
--- a/packages/core/type-definitions/combinator/multicast.d.ts
+++ b/packages/core/type-definitions/combinator/multicast.d.ts
@@ -1,3 +1,22 @@
-import { Stream } from '@most/types';
+import { Disposable, Scheduler, Sink, Stream, Time } from '@most/types';
 
 export function multicast<A>(s: Stream<A>): Stream<A>;
+
+export class MulticastSource<A> implements Stream<A>, Sink<A>, Disposable {
+  public source: Stream<A>;
+  public sinks: Array<Sink<A>>;
+  public disposable: Disposable;
+
+  constructor(source: Stream<A>);
+
+  public run(sink: Sink<A>, scheduler: Scheduler): Disposable;
+
+  public event(time: Time, value: A): void;
+  public error(time: Time, error: Error): void;
+  public end(time: Time): void;
+
+  public add(sink: Sink<A>): number;
+  public remove(sink: Sink<A>): number;
+
+  public dispose(): void;
+}


Test patch:
diff --git a/packages/core/test/combinator/multicast-test.js b/packages/core/test/combinator/multicast-test.js
index 43f3ca0..6596003 100644
--- a/packages/core/test/combinator/multicast-test.js
+++ b/packages/core/test/combinator/multicast-test.js
@@ -1,14 +1,14 @@
-import { describe, it } from 'mocha'
+import { MulticastDisposable, MulticastSource, multicast } from '../../src/combinator/multicast'
 import { assert, eq } from '@briancavalier/assert'
-import { spy } from 'sinon'
+import { collectEvents, makeEvents, ticks } from '../helper/testEnv'
+import { describe, it } from 'mocha'
+import { map, tap } from '../../src/combinator/transform'
 
+import FakeDisposeStream from '../helper/FakeDisposeStream'
 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'
+import { spy } from 'sinon'
 
 describe('multicast', () => {
   it('should be identity for already-multicasted stream', () => {
@@ -158,10 +158,10 @@ describe('MulticastDisposable', () => {
         return 0
       },
 
-      _dispose () {}
+      dispose () {}
     }
 
-    const disposed = spy(source, '_dispose')
+    const disposed = spy(source, 'dispose')
     const removed = spy(source, 'remove')
 
     const md = new MulticastDisposable(source, expectedSink)
@@ -180,10 +180,10 @@ describe('MulticastDisposable', () => {
         return 1
       },
 
-      _dispose () {}
+      dispose () {}
     }
 
-    const disposed = spy(source, '_dispose')
+    const disposed = spy(source, 'dispose')
     const removed = spy(source, 'remove')
 
     const md = new MulticastDisposable(source, expectedSink)