Repository: mostjs/core
Pull request #46 (2017-03-05 19:54:47)

Add newStream
This adds a trivial constructor to make a stream from a `Sink -> Scheduler -> Disposable` function, which is all a stream is anyway now that #45 has been merged.  While someone *can* just make an object with a run() *method*, this feels a reasonable API to help folks.  It doesn't add any additional footgunnery beyond object-with-run-method, afaict, though it may encourage folks to try to write more imperative push streams ... maybe that's solvable via documentation?

### Todo

- [x] Add unit tests if this seems worthwhile.

TylorS: :shipit: 
briancavalier: Added tests, and moved reduce, observe, and drain to the test/helpers folder, so they're no longer even a part of the main codebase 🎉 

@TylorS @davidchase can you guys take another quick look?  Thanks!

Patch:
diff --git a/packages/core/src/index.js.flow b/packages/core/src/index.js.flow
index aad5454..8a45801 100644
--- a/packages/core/src/index.js.flow
+++ b/packages/core/src/index.js.flow
@@ -1,5 +1,5 @@
 // @flow
-import type { Stream, Sink, Scheduler, Task, Time, Period, Delay } from '@most/types'
+import type { Stream, Sink, Scheduler, Task, Disposable, Time, Period, Delay } from '@most/types'
 
 export type SeedValue <S, A> = {
   seed: S,
@@ -11,6 +11,10 @@ declare export function runEffects <A> (s: Stream<A>, scheduler: Scheduler): Pro
 declare export function empty (): Stream<any>
 declare export function never (): Stream<any>
 declare export function just <A> (a: A): Stream<A>
+
+export type RunStream<A> = (sink: Sink<A>, scheduler: Scheduler) => Disposable
+declare export function newStream <A> (run: RunStream<A>): Stream<A>
+
 declare export function fromArray <A> (a: Array<A>): Stream<A>
 declare export function fromIterable <A> (a: Iterable<A>): Stream<A>
 
diff --git a/packages/core/src/source/newStream.js b/packages/core/src/source/newStream.js
new file mode 100644
index 0000000..ae4a421
--- /dev/null
+++ b/packages/core/src/source/newStream.js
@@ -0,0 +1,9 @@
+/** @license MIT License (c) copyright 2010-2017 original author or authors */
+
+export const newStream = run => new Stream(run)
+
+class Stream {
+  constructor (run) {
+    this.run = run
+  }
+}
diff --git a/packages/core/src/combinator/observe.js b/packages/core/test/helper/observe.js
similarity index 84%
rename from packages/core/src/combinator/observe.js
rename to packages/core/test/helper/observe.js
index e92552d..6be7979 100644
--- a/packages/core/src/combinator/observe.js
+++ b/packages/core/test/helper/observe.js
@@ -2,9 +2,9 @@
 /** @author Brian Cavalier */
 /** @author John Hann */
 
-import { runEffects } from '../runEffects'
-import defaultScheduler from '../scheduler/defaultScheduler'
-import { tap } from './transform'
+import { runEffects } from '../../src/runEffects'
+import defaultScheduler from '../../src/scheduler/defaultScheduler'
+import { tap } from '../../src/combinator/transform'
 
 /**
  * Observe all the event values in the stream in time order. The
diff --git a/packages/core/src/combinator/reduce.js b/packages/core/test/helper/reduce.js
similarity index 87%
rename from packages/core/src/combinator/reduce.js
rename to packages/core/test/helper/reduce.js
index 34633c5..9380656 100644
--- a/packages/core/src/combinator/reduce.js
+++ b/packages/core/test/helper/reduce.js
@@ -2,9 +2,9 @@
 /** @author Brian Cavalier */
 /** @author John Hann */
 
-import Pipe from '../sink/Pipe'
-import { runEffects } from '../runEffects'
-import defaultScheduler from '../scheduler/defaultScheduler'
+import Pipe from '../../src/sink/Pipe'
+import { runEffects } from '../../src/runEffects'
+import defaultScheduler from '../../src/scheduler/defaultScheduler'
 
 /**
 * Reduce a stream to produce a single result.  Note that reducing an infinite
diff --git a/packages/core/type-definitions/most.d.ts b/packages/core/type-definitions/most.d.ts
index b277a73..7caf268 100644
--- a/packages/core/type-definitions/most.d.ts
+++ b/packages/core/type-definitions/most.d.ts
@@ -1,5 +1,4 @@
 export * from './runEffects';
-export * from './scheduler';
 export * from './PropagateTask';
 
 export * from './combinator/applicative';
@@ -25,6 +24,7 @@ export * from './combinator/transform';
 export * from './combinator/zip';
 
 export * from './source/core';
+export * from './source/newStream';
 export * from './source/fromArray';
 export * from './source/fromIterable';
 export * from './source/generate';
diff --git a/packages/core/type-definitions/scheduler.d.ts b/packages/core/type-definitions/scheduler.d.ts
deleted file mode 100644
index 8584b37..0000000
--- a/packages/core/type-definitions/scheduler.d.ts
+++ /dev/null
@@ -1,3 +0,0 @@
-import { Scheduler } from '@most/types';
-
-export function newDefaultScheduler(): Scheduler;
diff --git a/packages/core/type-definitions/source/newStream.d.ts b/packages/core/type-definitions/source/newStream.d.ts
new file mode 100644
index 0000000..b54cfe1
--- /dev/null
+++ b/packages/core/type-definitions/source/newStream.d.ts
@@ -0,0 +1,5 @@
+import { Stream, Sink, Scheduler, Disposable } from '@most/types';
+
+export type RunStream<A> = (sink: Sink<A>, scheduler: Scheduler) => Disposable
+
+export function newStream<A>(run: RunStream<A>): Stream<A>;


Test patch:
diff --git a/packages/core/test/combinator/chain-test.js b/packages/core/test/combinator/chain-test.js
index 063096d..2e577fc 100644
--- a/packages/core/test/combinator/chain-test.js
+++ b/packages/core/test/combinator/chain-test.js
@@ -6,7 +6,7 @@ import { chain, join } from '../../src/combinator/chain'
 import { delay } from '../../src/combinator/delay'
 import { concat } from '../../src/combinator/build'
 import { take } from '../../src/combinator/slice'
-import { drain } from '../../src/combinator/observe'
+import { drain } from '../helper/observe'
 import { just, never } from '../../src/source/core'
 import { fromArray } from '../../src/source/fromArray'
 
diff --git a/packages/core/test/combinator/concatMap-test.js b/packages/core/test/combinator/concatMap-test.js
index d0b13d8..06b350d 100644
--- a/packages/core/test/combinator/concatMap-test.js
+++ b/packages/core/test/combinator/concatMap-test.js
@@ -7,7 +7,7 @@ import { assertSame } from '../helper/stream-helper'
 import * as concatMap from '../../src/combinator/concatMap'
 import { concat } from '../../src/combinator/build'
 import { take } from '../../src/combinator/slice'
-import { drain } from '../../src/combinator/observe'
+import { drain } from '../helper/observe'
 import { just, never } from '../../src/source/core'
 import { fromArray } from '../../src/source/fromArray'
 
diff --git a/packages/core/test/combinator/continueWith-test.js b/packages/core/test/combinator/continueWith-test.js
index 44fb2db..10da3e4 100644
--- a/packages/core/test/combinator/continueWith-test.js
+++ b/packages/core/test/combinator/continueWith-test.js
@@ -2,7 +2,7 @@ import { describe, it } from 'mocha'
 import { is, fail } from '@briancavalier/assert'
 
 import { continueWith } from '../../src/combinator/continueWith'
-import { drain } from '../../src/combinator/observe'
+import { drain } from '../helper/observe'
 import { just } from '../../src/source/core'
 
 describe('continueWith', () => {
diff --git a/packages/core/test/combinator/errors-test.js b/packages/core/test/combinator/errors-test.js
index 507303c..c988aea 100644
--- a/packages/core/test/combinator/errors-test.js
+++ b/packages/core/test/combinator/errors-test.js
@@ -3,7 +3,7 @@ import { fail, is, eq } from '@briancavalier/assert'
 
 import { throwError, recoverWith } from '../../src/combinator/errors'
 import { map } from '../../src/combinator/transform'
-import { observe, drain } from '../../src/combinator/observe'
+import { observe, drain } from '../helper/observe'
 import { just } from '../../src/source/core'
 
 const sentinel = { value: 'sentinel' }
diff --git a/packages/core/test/combinator/mergeConcurrently-test.js b/packages/core/test/combinator/mergeConcurrently-test.js
index 07e6661..7bdc5e3 100644
--- a/packages/core/test/combinator/mergeConcurrently-test.js
+++ b/packages/core/test/combinator/mergeConcurrently-test.js
@@ -5,7 +5,7 @@ import { mergeMapConcurrently, mergeConcurrently } from '../../src/combinator/me
 import { periodic } from '../../src/source/periodic'
 import { take } from '../../src/combinator/slice'
 import { constant } from '../../src/combinator/transform'
-import { drain } from '../../src/combinator/observe'
+import { drain } from '../helper/observe'
 import { just as just } from '../../src/source/core'
 import { fromArray } from '../../src/source/fromArray'
 import { collectEventsFor } from '../helper/testEnv'
diff --git a/packages/core/test/helper/stream-helper.js b/packages/core/test/helper/stream-helper.js
index b87927e..7dc47dd 100644
--- a/packages/core/test/helper/stream-helper.js
+++ b/packages/core/test/helper/stream-helper.js
@@ -1,5 +1,5 @@
 import { eq } from '@briancavalier/assert'
-import { reduce } from '../../src/combinator/reduce'
+import { reduce } from './reduce'
 
 export function assertSame (s1, s2) {
   return Promise.all([toArray(s1), toArray(s2)]).then(arrayEquals)
diff --git a/packages/core/test/newStream-test.js b/packages/core/test/newStream-test.js
new file mode 100644
index 0000000..ddd839a
--- /dev/null
+++ b/packages/core/test/newStream-test.js
@@ -0,0 +1,13 @@
+import { describe, it } from 'mocha'
+import { is } from '@briancavalier/assert'
+
+import { newStream } from '../src/source/newStream'
+
+describe('newStream', () => {
+  it('should create new stream from RunStream function', () => {
+    const run = (sink, scheduler) => ({ dispose: () => Promise.resolve() })
+    const s = newStream(run)
+
+    is(run, s.run)
+  })
+})
diff --git a/packages/core/test/observe-test.js b/packages/core/test/observe-test.js
index 323b90f..797e26a 100644
--- a/packages/core/test/observe-test.js
+++ b/packages/core/test/observe-test.js
@@ -1,7 +1,7 @@
 import { describe, it } from 'mocha'
 import { eq, assert } from '@briancavalier/assert'
 
-import { observe, drain } from '../src/combinator/observe'
+import { observe, drain } from './helper/observe'
 import { iterate } from '../src/source/iterate'
 import { take } from '../src/combinator/slice'
 import { just as streamOf } from '../src/source/core'
diff --git a/packages/core/test/source/generate-test.js b/packages/core/test/source/generate-test.js
index 5e92606..31fb62e 100644
--- a/packages/core/test/source/generate-test.js
+++ b/packages/core/test/source/generate-test.js
@@ -2,7 +2,7 @@ import { describe, it } from 'mocha'
 import { eq } from '@briancavalier/assert'
 
 import { generate } from '../../src/source/generate'
-import { reduce } from '../../src/combinator/reduce'
+import { reduce } from '../helper/reduce'
 import { getIterator } from '../../src/iterable'
 import ArrayIterable from '../helper/ArrayIterable'
 import delayPromise from '../helper/delayPromise'
diff --git a/packages/core/test/source/iterate-test.js b/packages/core/test/source/iterate-test.js
index 180fb06..f75df9e 100644
--- a/packages/core/test/source/iterate-test.js
+++ b/packages/core/test/source/iterate-test.js
@@ -3,7 +3,7 @@ import { eq, is, fail } from '@briancavalier/assert'
 
 import { iterate } from '../../src/source/iterate'
 import { take } from '../../src/combinator/slice'
-import { observe } from '../../src/combinator/observe'
+import { observe } from '../helper/observe'
 
 import { collectEventsFor } from '../helper/testEnv'
 
diff --git a/packages/core/test/source/unfold-test.js b/packages/core/test/source/unfold-test.js
index 893fc21..5b94b3b 100644
--- a/packages/core/test/source/unfold-test.js
+++ b/packages/core/test/source/unfold-test.js
@@ -3,7 +3,7 @@ import { eq, is, fail } from '@briancavalier/assert'
 
 import { unfold } from '../../src/source/unfold'
 import { take } from '../../src/combinator/slice'
-import { observe } from '../../src/combinator/observe'
+import { observe } from '../helper/observe'
 
 import { collectEventsFor } from '../helper/testEnv'