Repository: mostjs/core
Pull request #120 (2017-08-31 22:46:37)

feat(core): Add runStream as function from of stream.run
Add `runStream` as the function form of `stream.run`.

### Open questions

Argument ordering.  It's not clear what the "right" argument ordering is.  Since the subject of the function is a Stream, the obvious choice is to make it last.  After that, though, it's less clear.

Consider that we have two other, similar functions, and it might be nice to have some coherent criteria for ordering the arguments across `runStream` and these:

- [runEffects](https://github.com/mostjs/core/blob/master/packages/core/src/runEffects.js#L5) - `runEffects :: Stream -> Scheduler`, and
- [runWithLocalTime](https://github.com/mostjs/core/blob/master/packages/core/src/scheduler/runWithLocalTime.js#L11) - `runWithLocalTime :: Offset -> Stream a -> Sink a -> Scheduler`
    - I believe my thinking here was that it made sense for the last two args to mimic the argument ordering of `stream.run`, i.e. sink, scheduler.  Maybe it makes sense to move the Stream to the last arg to end up with `Offset -> Sink a -> Scheduler -> Stream a`, which is similar to `runStream` with the extra `Offset` arg prepended.

Thoughts and suggestions welcome!

### Todo

- [x] docs


davidchase: I like stream last approach, in a similar to data last... given you can make a simple sink and have a defaultScheduler but not yet know what the stream is going to be entirely. If that makes any sense 😄 
briancavalier: Ok, I found a coherent argument ordering for runStream and runStreamWithLocalTime (newly renamed from runWithLocalTime) that I think makes sense.  I'm not really happy with the name runStreamWithLocalTime, though.  It'd be nice to find something shorter and at least as clear.

Ideas?

I'm inclined to keep runEffects' current "flipped" argument order (i.e. scheduler last, `Stream a -> Scheduler`).  I'm having trouble coming up with compelling reasons to justify that other than trying to provide some guidance about calling runEffects once and at the top level, instead of partially applying a Scheduler first and then scattering calls to runEffects around a codebase.  Obviously someone can just flip(runEffects), but at least they have to make that conscious decision.
davidchase: why not just leave it `runWithLocalTime` and `run` or `runStream` ?
briancavalier: My thinking is that the name `run` is a little less helpful than `runStream`, but maybe it's ok, since the types make it obvious.

The "WithLocalTime" suffix, while very explicit, seems a bit wordy.  I wonder if there's a way to say the same thing in less than 3 camel humps, like maybe "WithOrigin" or just "At": runWithOrigin, runAt, or runStreamWithOrigin, runStreamAt.  Or maybe "Relative": runRelative or runStreamRelative.

Do any of those seem appealing or give you other ideas?

I'm not totally against runWithLocalTime or runStreamWithLocalTime since they will tend to be used less often than runEffects, but if we _could_ find something more concise, that'd be great.
davidchase: I like run because like you mention types and the fact this is a stream library but maybe that's me with that said I like runWithOrigin or runWithTime if you really want make it obvious .. though for some reason it almost makes sense to make it runWithEffects ?
So something like
run
runWithEffects
runWithOrigin or runWithTime

Thoughts ?
briancavalier: > though for some reason it almost makes sense to make it runWithEffects

I'm not sure I understand.  We started with `runEffects` as the primary application-level API for activating a stream graph, and this PR adds two functions, currently named `runStream` and `runStreamWithLocalTime`, primarily to be used when authoring custom stream types. Are you suggesting we rename that to `runWithEffects`, or that we rename the `runStream` in this PR to `runWithEffects`?

One tricky thing now is that all 3 of these functions, `runEffects`, `runStream`, and `runStreamWithLocalTime`, run effects.  That is, they all activate a stream and begin consuming all its events and performing any side-effects.  They differ in their parameters and return values.

I can even imagine other functions for activating a stream:

```hs
-- Provide a handler for error, and a handler for end
cleverName :: (Error -> void) -> (() -> void) -> Stream a -> Scheduler -> void

-- Provide a handler for error only
cleverName2 :: (Error -> void) -> Stream a -> Scheduler -> void
```

Which maybe opens the can of worms about whether returning a promise is even worthwhile.  Given the first variant above, someone could easily do this to get themselves a promise:

```js
new Promise((resolve, reject) => cleverName(resolve, reject, stream, scheduler))
```





briancavalier: A few thoughts on short names (even though I don't think short is a necessity here since these functions won't be called all that often):

run
runAt

runStream
runStreamAt

run
runWithTime
(or runWithOrigin)
Frikki: `runLocalized`
briancavalier: See 46a6511, which splits stream clock localization into its own withLocalTime combinator.  Instead of needing 3 run-like functions, it means we only need 2: the existing runEffects(), and run(), where run is just the trivial function version of stream.run: `run :: Sink a -> Scheduler -> Stream a -> Disposable`.

briancavalier: If we like this direction, then there are a couple remaining todos:

- [ ] Add unit tests for withLocalTime
- [ ] Add docs for run and withLocalTime
davidchase: Love it 😍 
briancavalier: - [ ] Update TS and Flow type defs

Patch:
diff --git a/docs/api.rst b/docs/api.rst
index 10335b9..d4aad7c 100644
--- a/docs/api.rst
+++ b/docs/api.rst
@@ -189,6 +189,17 @@ runEffects
 
 Activate an event :ref:`Stream` and consume all its events.
 
+.. _runStream:
+
+run
+```
+
+.. code-block:: haskell
+
+  run :: Sink a -> Scheduler -> Stream a -> void
+
+Run a :ref:`Stream`, sending all events to the provided :ref:`Sink`.  The Stream's :ref:`Time` values come from the provided :ref:`Scheduler`.
+
 Construction
 ^^^^^^^^^^^^
 
@@ -553,7 +564,7 @@ To control concurrency, ``mergeConcurrently`` must maintain an internal queue of
 .. _mergeMapConcurrently:
 
 mergeMapConcurrently
-```````````````````
+````````````````````
 
 .. code-block:: haskell
 
@@ -925,6 +936,9 @@ This is similar to :ref:`slice`, but uses time rather than indices to "slice" th
   // 3. discard all subsequent events.
   during(timeWindow, stream)
 
+Dealing with time
+`````````````````
+
 .. _delay:
 
 delay
@@ -942,6 +956,22 @@ Timeshift a :ref:`Stream` by *n* milliseconds. ::
 
 Delaying a :ref:`Stream` timeshifts all the events by the same amount. It doesn't change the time *between* events.
 
+.. _withLocalTime:
+
+withLocalTime
+`````````````
+
+.. code-block:: haskell
+
+  withLocalTime :: Time -> Stream a -> Stream a
+
+Create a Stream with localized :ref:`Time` values, whose origin (i.e., time 0) is at the specified Time on the :ref:`Scheduler` provided when the Stream is observed with :ref:`runEffects` or :ref:`run`.
+
+When implementing custom higher-order :ref:`Stream` combinators, such as :ref:`chain`, you should use ``withLocalTime`` to localize "inner" Streams before running them.
+
+Rate limiting
+`````````````
+
 .. _throttle:
 
 throttle
diff --git a/packages/core/src/combinator/continueWith.js b/packages/core/src/combinator/continueWith.js
index df3241c..f91abc7 100644
--- a/packages/core/src/combinator/continueWith.js
+++ b/packages/core/src/combinator/continueWith.js
@@ -3,7 +3,8 @@
 /** @author John Hann */
 
 import Pipe from '../sink/Pipe'
-import { runWithLocalTime } from '../scheduler/runWithLocalTime'
+import { run } from '../run'
+import { withLocalTime } from '../combinator/withLocalTime'
 import { disposeOnce, tryDispose } from '@most/disposable'
 
 export const continueWith = (f, stream) =>
@@ -55,7 +56,7 @@ class ContinueWithSink extends Pipe {
   }
 
   _continue (f, t, sink) {
-    return runWithLocalTime(t, f(), sink, this.scheduler)
+    return run(sink, this.scheduler, withLocalTime(t, f()))
   }
 
   dispose () {
diff --git a/packages/core/src/combinator/errors.js b/packages/core/src/combinator/errors.js
index 880df4f..7370e85 100644
--- a/packages/core/src/combinator/errors.js
+++ b/packages/core/src/combinator/errors.js
@@ -7,7 +7,8 @@ import { tryDispose } from '@most/disposable'
 import { asap } from '@most/scheduler'
 import { tryEvent, tryEnd } from '../source/tryEvent'
 import { propagateErrorTask } from '../scheduler/PropagateTask'
-import { runWithLocalTime } from '../scheduler/runWithLocalTime'
+import { run } from '../run'
+import { withLocalTime } from './withLocalTime'
 
 /**
  * If stream encounters an error, recover and continue with items from stream
@@ -81,7 +82,7 @@ class RecoverWithSink {
   }
 
   _continue (f, t, x, sink) {
-    return runWithLocalTime(t, f(x), sink, this.scheduler)
+    return run(sink, this.scheduler, withLocalTime(t, f(x)))
   }
 
   dispose () {
diff --git a/packages/core/src/scheduler/runWithLocalTime.js b/packages/core/src/combinator/withLocalTime.js
similarity index 53%
rename from packages/core/src/scheduler/runWithLocalTime.js
rename to packages/core/src/combinator/withLocalTime.js
index b2245d1..dfce9a8 100644
--- a/packages/core/src/scheduler/runWithLocalTime.js
+++ b/packages/core/src/combinator/withLocalTime.js
@@ -1,15 +1,24 @@
-/** @license MIT License (c) copyright 2010-2017 original author or authors */
-
 import RelativeSink from '../sink/RelativeSink'
 import { schedulerRelativeTo } from '@most/scheduler'
 
-// Run a stream with its own localized clock
+// Create a stream with its own local 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(relativeSink(origin, sink), schedulerRelativeTo(origin, scheduler))
+// propagating events to sink.  In other words, upstream sources will see local times,
+// and downstream sinks will see non-local (original) times.
+export const withLocalTime = (origin, stream) =>
+  new WithLocalTime(origin, stream)
+
+class WithLocalTime {
+  constructor (origin, source) {
+    this.origin = origin
+    this.source = source
+  }
+
+  run (sink, scheduler) {
+    return this.source.run(relativeSink(this.origin, sink), schedulerRelativeTo(this.origin, scheduler))
+  }
+}
 
 // Accumulate offsets instead of nesting RelativeSinks, which can happen
 // with higher-order stream and combinators like continueWith when they're
diff --git a/packages/core/src/index.js b/packages/core/src/index.js
index fc4e0db..cc762c8 100644
--- a/packages/core/src/index.js
+++ b/packages/core/src/index.js
@@ -19,8 +19,16 @@ export const withArrayValues = curry2(_withArrayValues)
 // Observing
 
 import { runEffects as _runEffects } from './runEffects'
+import { run as _run } from './run'
 
 export const runEffects = curry2(_runEffects)
+export const run = curry3(_run)
+
+// -------------------------------------------------------
+
+import { withLocalTime as _withLocalTime } from './combinator/withLocalTime'
+
+export const withLocalTime = curry2(_withLocalTime)
 
 // -------------------------------------------------------
 
diff --git a/packages/core/src/index.js.flow b/packages/core/src/index.js.flow
index 6cccd5d..d4e23e0 100644
--- a/packages/core/src/index.js.flow
+++ b/packages/core/src/index.js.flow
@@ -7,6 +7,12 @@ export type SeedValue <S, A> = {
 }
 
 declare export function runEffects <A> (s: Stream<A>, scheduler: Scheduler): Promise<void>
+declare export function runEffects <A> (s: Stream<A>): (scheduler: Scheduler) => Promise<void>
+
+declare export function run <A> (sink: Sink<A>, scheduler: Scheduler, s: Stream<A>): Disposable
+declare export function run <A> (sink: Sink<A>): (scheduler: Scheduler, s: Stream<A>) => Disposable
+declare export function run <A> (sink: Sink<A>, scheduler: Scheduler): (s: Stream<A>) => Disposable
+declare export function run <A> (sink: Sink<A>): (scheduler: Scheduler) => (s: Stream<A>) => Disposable
 
 declare export function empty (): Stream<any>
 declare export function never (): Stream<any>
@@ -159,6 +165,9 @@ declare export function since <A> (since: Stream<any>): (s: Stream<A>) => Stream
 
 declare export function during <A> (since: Stream<Stream<any>>, s: Stream<A>): Stream<A>
 
+declare export function withLocalTime <A> (origin: Time, s: Stream<A>): Stream<A>
+declare export function withLocalTime <A> (origin: Time): (s: Stream<A>) => Stream<A>
+
 declare export function delay <A> (delay: Delay, s: Stream<A>): Stream<A>
 declare export function delay <A> (delay: Delay): (s: Stream<A>) => Stream<A>
 
diff --git a/packages/core/src/run.js b/packages/core/src/run.js
new file mode 100644
index 0000000..067cb6a
--- /dev/null
+++ b/packages/core/src/run.js
@@ -0,0 +1,6 @@
+/** @license MIT License (c) copyright 2010-2017 original author or authors */
+
+// Run a Stream, sending all its events to the
+// provided Sink.
+export const run = (sink, scheduler, stream) =>
+    stream.run(sink, scheduler)
diff --git a/packages/core/src/runEffects.js b/packages/core/src/runEffects.js
index 1c4ee30..82f4e87 100644
--- a/packages/core/src/runEffects.js
+++ b/packages/core/src/runEffects.js
@@ -1,10 +1,11 @@
 /** @license MIT License (c) copyright 2010-2017 original author or authors */
 
+import { curry2 } from '@most/prelude'
 import SettableDisposable from './disposable/SettableDisposable'
 
-export const runEffects = (stream, scheduler) =>
+export const runEffects = curry2((stream, scheduler) =>
   new Promise((resolve, reject) =>
-    runStream(stream, scheduler, resolve, reject))
+    runStream(stream, scheduler, resolve, reject)))
 
 function runStream (stream, scheduler, resolve, reject) {
   const disposable = new SettableDisposable()
diff --git a/packages/core/type-definitions/combinator/withLocalTime.d.ts b/packages/core/type-definitions/combinator/withLocalTime.d.ts
new file mode 100644
index 0000000..0ce2ed2
--- /dev/null
+++ b/packages/core/type-definitions/combinator/withLocalTime.d.ts
@@ -0,0 +1,4 @@
+import { Time, Stream } from '@most/types'
+
+export function withLocalTime <A> (origin: Time, s: Stream<A>): Stream<A>
+export function withLocalTime <A> (origin: Time): (s: Stream<A>) => Stream<A>
diff --git a/packages/core/type-definitions/most.d.ts b/packages/core/type-definitions/most.d.ts
index bff4e33..b89af77 100644
--- a/packages/core/type-definitions/most.d.ts
+++ b/packages/core/type-definitions/most.d.ts
@@ -1,4 +1,5 @@
 export * from './runEffects';
+export * from './run';
 export * from './PropagateTask';
 
 export * from './combinator/applicative';
@@ -24,6 +25,7 @@ export * from './combinator/timeslice';
 export * from './combinator/transform';
 export * from './combinator/zip';
 export * from './combinator/withArrayValues';
+export * from './combinator/withLocalTime';
 
 export * from './source/never';
 export * from './source/empty';
diff --git a/packages/core/type-definitions/run.d.ts b/packages/core/type-definitions/run.d.ts
new file mode 100644
index 0000000..04f6e39
--- /dev/null
+++ b/packages/core/type-definitions/run.d.ts
@@ -0,0 +1,6 @@
+import { Sink, Scheduler, Stream, Disposable } from '@most/types';
+
+export function run <A> (sink: Sink<A>, scheduler: Scheduler, s: Stream<A>): Disposable;
+export function run <A> (sink: Sink<A>): (scheduler: Scheduler, s: Stream<A>) => Disposable;
+export function run <A> (sink: Sink<A>, scheduler: Scheduler): (s: Stream<A>) => Disposable;
+export function run <A> (sink: Sink<A>): (scheduler: Scheduler) => (s: Stream<A>) => Disposable;


Test patch:
diff --git a/packages/core/test/combinator/withLocalTime-test.js b/packages/core/test/combinator/withLocalTime-test.js
new file mode 100644
index 0000000..b02f89c
--- /dev/null
+++ b/packages/core/test/combinator/withLocalTime-test.js
@@ -0,0 +1,18 @@
+// @flow
+import { describe, it } from 'mocha'
+import { eq } from '@briancavalier/assert'
+
+import { at } from '../../src/source/at'
+import { collectEventsFor } from '../helper/testEnv'
+import { timestamp } from '../helper/timestamp'
+import { withLocalTime } from '../../src/combinator/withLocalTime'
+
+describe('withLocalTime', () => {
+  it('should localize upstream times and unlocalize downstream times', () => {
+    const s = withLocalTime(1, timestamp(at(1, 1)))
+    return collectEventsFor(1, s)
+      .then(eq([{
+        time: 1, value: { time: 0, value: 1 }
+      }]))
+  })
+})
diff --git a/packages/core/test/helper/timestamp.js b/packages/core/test/helper/timestamp.js
new file mode 100644
index 0000000..818bf97
--- /dev/null
+++ b/packages/core/test/helper/timestamp.js
@@ -0,0 +1,20 @@
+import Pipe from '../../src/sink/Pipe'
+
+export const timestamp = stream =>
+  new Timestamp(stream)
+
+class Timestamp {
+  constructor (stream) {
+    this.stream = stream
+  }
+
+  run (sink, scheduler) {
+    return this.stream.run(new TimestampSink(sink), scheduler)
+  }
+}
+
+class TimestampSink extends Pipe {
+  event (time, value) {
+    this.sink.event(time, { time, value })
+  }
+}
diff --git a/packages/core/test/run-test.js b/packages/core/test/run-test.js
new file mode 100644
index 0000000..de24c96
--- /dev/null
+++ b/packages/core/test/run-test.js
@@ -0,0 +1,43 @@
+// @flow
+import { describe, it } from 'mocha'
+import { eq, fail } from '@briancavalier/assert'
+
+import { run } from '../src/run'
+import { at } from '../src/source/at'
+
+import { ticks } from './helper/testEnv'
+
+describe('run', () => {
+  it('should run a stream', done => {
+    // This could be tested with a spy that verifies
+    // stream.run was called with the same arguments
+    // and runStream doesn't modify the return value
+    // But, I think this is better: test the invariants
+    // rather than the implementation.  If we construct
+    // a stream with an expected outcome, then run
+    // the stream, it should, in fact, produce that
+    // outcome.
+    const events = []
+    const expectedTime = 10 * Math.random()
+    const expectedValue = Math.random()
+    const expected = [
+      { time: expectedTime, value: expectedValue }
+    ]
+    const sink = {
+      event (time, value) {
+        events.push({ time, value })
+      },
+      end (time) {
+        eq(expectedTime, time)
+        eq(expected, events)
+        done()
+      },
+      error (t, e) {
+        fail(`Stream failed: ${t} ${e}`)
+      }
+    }
+
+    const s = at(expectedTime, expectedValue)
+    run(sink, ticks(expectedTime), s)
+  })
+})