Repository: mostjs/core (ts)

Issue:
Return canonical empty() in filter operation
If the input stream is `empty()`, return the canonical `empty()`.

Reference #152

Pull request description:
Filter return canonical empty
Close #180
Reference #150, #152

Signed-off-by: Frederik Krautwald <fkrautwald@gmail.com>

Patch:
diff --git a/packages/core/package.json b/packages/core/package.json
index 0b32a9f..5b334c4 100644
--- a/packages/core/package.json
+++ b/packages/core/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@most/core",
-  "version": "1.0.1",
+  "version": "1.0.2",
   "description": "Reactive programming with lean, functions-only, curried, tree-shakeable API",
   "typings": "type-definitions/most.d.ts",
   "main": "dist/index.js",
diff --git a/packages/core/src/combinator/combine.js b/packages/core/src/combinator/combine.js
index aee954a..63254f6 100644
--- a/packages/core/src/combinator/combine.js
+++ b/packages/core/src/combinator/combine.js
@@ -1,9 +1,7 @@
-/** @license MIT License (c) copyright 2010-2016 original author or authors */
-/** @author Brian Cavalier */
-/** @author John Hann */
+/** @license MIT License (c) copyright 2010 original author or authors */
 
 import { map } from './transform'
-import { empty } from '../source/empty'
+import { empty, containsCanonicalEmpty } from '../source/empty'
 import Pipe from '../sink/Pipe'
 import IndexSink from '../sink/IndexSink'
 import { disposeAll, tryDispose } from '@most/disposable'
@@ -15,9 +13,8 @@ import invoke from '../invoke'
  * @returns {Stream} stream containing the result of applying f to the most recent
  *  event of each input stream, whenever a new event arrives on any stream.
  */
-export function combine (f, stream1, stream2) {
-  return combineArray(f, [stream1, stream2])
-}
+export const combine = (f, stream1, stream2) =>
+  combineArray(f, [stream1, stream2])
 
 /**
 * Combine latest events from all input streams
@@ -27,7 +24,7 @@ export function combine (f, stream1, stream2) {
 *  event of each input stream, whenever a new event arrives on any stream.
 */
 export const combineArray = (f, streams) =>
-  streams.length === 0 ? empty()
+  streams.length === 0 || containsCanonicalEmpty(streams) ? empty()
     : streams.length === 1 ? map(f, streams[0])
     : new Combine(f, streams)
 
diff --git a/packages/core/src/combinator/loop.js b/packages/core/src/combinator/loop.js
index 2ff2a22..3cc544c 100644
--- a/packages/core/src/combinator/loop.js
+++ b/packages/core/src/combinator/loop.js
@@ -1,8 +1,7 @@
-/** @license MIT License (c) copyright 2010-2016 original author or authors */
-/** @author Brian Cavalier */
-/** @author John Hann */
+/** @license MIT License (c) copyright 2010 original author or authors */
 
 import Pipe from '../sink/Pipe'
+import { empty, isCanonicalEmpty } from '../source/empty'
 
 /**
  * Generalized feedback loop. Call a stepper function for each event. The stepper
@@ -16,7 +15,8 @@ import Pipe from '../sink/Pipe'
  * returned by the stepper
  */
 export const loop = (stepper, seed, stream) =>
-  new Loop(stepper, seed, stream)
+  isCanonicalEmpty(stream) ? empty()
+    : new Loop(stepper, seed, stream)
 
 class Loop {
   constructor (stepper, seed, source) {
diff --git a/packages/core/src/combinator/merge.js b/packages/core/src/combinator/merge.js
index 5dd46d1..f6b4c64 100644
--- a/packages/core/src/combinator/merge.js
+++ b/packages/core/src/combinator/merge.js
@@ -4,7 +4,7 @@
 
 import Pipe from '../sink/Pipe'
 import IndexSink from '../sink/IndexSink'
-import { empty } from '../source/empty'
+import { empty, isCanonicalEmpty } from '../source/empty'
 import { disposeAll, tryDispose } from '@most/disposable'
 import { reduce } from '@most/prelude'
 
@@ -23,9 +23,7 @@ export function merge (stream1, stream2) {
  * arbitrary order.
  */
 export const mergeArray = streams =>
-  streams.length === 0 ? empty()
-    : streams.length === 1 ? streams[0]
-    : mergeStreams(streams)
+  mergeStreams(withoutCanonicalEmpty(streams))
 
 /**
  * This implements fusion/flattening for merge.  It will
@@ -37,7 +35,15 @@ export const mergeArray = streams =>
  * merge operations into a single merge.
  */
 const mergeStreams = streams =>
-  new Merge(reduce(appendSources, [], streams))
+  streams.length === 0 ? empty()
+    : streams.length === 1 ? streams[0]
+    : new Merge(reduce(appendSources, [], streams))
+
+const withoutCanonicalEmpty = streams =>
+  streams.filter(isNotCanonicalEmpty)
+
+const isNotCanonicalEmpty = stream =>
+  !isCanonicalEmpty(stream)
 
 const appendSources = (sources, stream) =>
   sources.concat(stream instanceof Merge ? stream.sources : stream)
diff --git a/packages/core/src/combinator/snapshot.js b/packages/core/src/combinator/snapshot.js
index 1b133b0..d36a018 100644
--- a/packages/core/src/combinator/snapshot.js
+++ b/packages/core/src/combinator/snapshot.js
@@ -2,12 +2,15 @@
 
 import Pipe from '../sink/Pipe'
 import { disposeBoth } from '@most/disposable'
+import { empty, isCanonicalEmpty } from '../source/empty'
 
 export const sample = (values, sampler) =>
-  snapshot((_, x) => x, values, sampler)
+  snapshot((x, _) => x, values, sampler)
 
 export const snapshot = (f, values, sampler) =>
-  new Snapshot(f, values, sampler)
+  isCanonicalEmpty(sampler) || isCanonicalEmpty(values)
+    ? empty()
+    : new Snapshot(f, values, sampler)
 
 export class Snapshot {
   constructor (f, values, sampler) {
diff --git a/packages/core/src/combinator/zip.js b/packages/core/src/combinator/zip.js
index 972adbd..5400126 100644
--- a/packages/core/src/combinator/zip.js
+++ b/packages/core/src/combinator/zip.js
@@ -1,9 +1,7 @@
-/** @license MIT License (c) copyright 2010-2016 original author or authors */
-/** @author Brian Cavalier */
-/** @author John Hann */
+/** @license MIT License (c) copyright 2010 original author or authors */
 
 import { map } from './transform'
-import { empty } from '../source/empty'
+import { empty, containsCanonicalEmpty } from '../source/empty'
 import Pipe from '../sink/Pipe'
 import IndexSink from '../sink/IndexSink'
 import { disposeAll } from '@most/disposable'
@@ -32,7 +30,7 @@ export function zip (f, stream1, stream2) {
 *  using f
 */
 export const zipArray = (f, streams) =>
-  streams.length === 0 ? empty()
+  streams.length === 0 || containsCanonicalEmpty(streams) ? empty()
     : streams.length === 1 ? map(f, streams[0])
     : new Zip(f, streams)
 
diff --git a/packages/core/src/fusion/Filter.js b/packages/core/src/fusion/Filter.js
index 5fba29b..e1bf652 100644
--- a/packages/core/src/fusion/Filter.js
+++ b/packages/core/src/fusion/Filter.js
@@ -3,6 +3,7 @@
 /** @author John Hann */
 
 import Pipe from '../sink/Pipe'
+import { isCanonicalEmpty } from '../source/empty'
 
 export default class Filter {
   constructor (p, source) {
@@ -21,6 +22,10 @@ export default class Filter {
    * @returns {Filter} filtered source
    */
   static create (p, source) {
+    if (isCanonicalEmpty(source)) {
+      return source
+    }
+
     if (source instanceof Filter) {
       return new Filter(and(source.p, p), source.source)
     }
diff --git a/packages/core/src/fusion/Map.js b/packages/core/src/fusion/Map.js
index 7d53101..710e36e 100644
--- a/packages/core/src/fusion/Map.js
+++ b/packages/core/src/fusion/Map.js
@@ -6,6 +6,7 @@ import Pipe from '../sink/Pipe'
 import Filter from './Filter'
 import FilterMap from './FilterMap'
 import { compose } from '@most/prelude'
+import { isCanonicalEmpty, empty } from '../source/empty'
 
 export default class Map {
   constructor (f, source) {
@@ -25,6 +26,10 @@ export default class Map {
    * @returns {Map|FilterMap} mapped source, possibly fused
    */
   static create (f, source) {
+    if (isCanonicalEmpty(source)) {
+      return empty()
+    }
+
     if (source instanceof Map) {
       return new Map(compose(f, source.f), source.source)
     }
diff --git a/packages/core/src/source/empty.js b/packages/core/src/source/empty.js
index a4c0801..c7c0391 100644
--- a/packages/core/src/source/empty.js
+++ b/packages/core/src/source/empty.js
@@ -8,6 +8,9 @@ export const empty = () => EMPTY
 export const isCanonicalEmpty = stream =>
   stream === EMPTY
 
+export const containsCanonicalEmpty = streams =>
+  streams.some(isCanonicalEmpty)
+
 class Empty {
   run (sink, scheduler) {
     return asap(propagateEndTask(sink), scheduler)


Test patch:
diff --git a/packages/core/test/combinator/loop-test.js b/packages/core/test/combinator/loop-test.js
index 1652144..921dfee 100644
--- a/packages/core/test/combinator/loop-test.js
+++ b/packages/core/test/combinator/loop-test.js
@@ -5,6 +5,7 @@ import { spy } from 'sinon'
 import { loop } from '../../src/combinator/loop'
 import { throwError } from '../../src/combinator/errors'
 import { now } from '../../src/source/now'
+import { empty, isCanonicalEmpty } from '../../src/source/empty'
 
 import FakeDisposeSource from '../helper/FakeDisposeStream'
 
@@ -16,6 +17,11 @@ const other = { value: 'other' }
 const toPair = (z, x) => ({ value: x, seed: z })
 
 describe('loop', function () {
+  it('given canonical empty, should return canonical empty', () => {
+    const s = loop((seed, value) => ({ seed, value }), 1, empty())
+    assert(isCanonicalEmpty(s))
+  })
+
   it('should call stepper with seed, value', function () {
     const a = ['a', 'b', 'c', 'd']
 
diff --git a/packages/core/test/combinator/merge-test.js b/packages/core/test/combinator/merge-test.js
index 8faa0ed..f7452c2 100644
--- a/packages/core/test/combinator/merge-test.js
+++ b/packages/core/test/combinator/merge-test.js
@@ -1,13 +1,33 @@
 import { describe, it } from 'mocha'
-import { eq } from '@briancavalier/assert'
+import { assert, is, eq } from '@briancavalier/assert'
 
 import { now } from '../../src/source/now'
+import { empty, isCanonicalEmpty } from '../../src/source/empty'
 import { delay } from '../../src/combinator/delay'
 import { merge, mergeArray } from '../../src/combinator/merge'
 
 import { makeEventsFromArray, collectEventsFor } from '../helper/testEnv'
 
 describe('merge', function () {
+  it('given one canonical empty, should return other stream', () => {
+    const s = now(0)
+    is(s, merge(s, empty()))
+  })
+
+  it('given one canonical empty, should return other stream', () => {
+    const s = now(1)
+    is(s, merge(empty(), s))
+  })
+
+  it('given both canonical empty, should return canonical empty', () => {
+    assert(isCanonicalEmpty(merge(empty(), empty())))
+  })
+
+  it('should recursively return canonical empty', () => {
+    const s = merge(empty(), merge(empty(), empty()))
+    assert(isCanonicalEmpty(s))
+  })
+
   it('should include items from all inputs', function () {
     return testMerge(merge)
   })
@@ -27,6 +47,24 @@ describe('merge', function () {
 })
 
 describe('mergeArray', function () {
+  it('given empty array, should return canonical empty', () => {
+    assert(isCanonicalEmpty(mergeArray([])))
+  })
+
+  it('given array containing only canonical empty, should return canonical empty', () => {
+    assert(isCanonicalEmpty(mergeArray([empty(), empty(), empty()])))
+  })
+
+  it('should recursively return canonical empty', () => {
+    const s = mergeArray([empty(), mergeArray([empty(), empty()])])
+    assert(isCanonicalEmpty(s))
+  })
+
+  it('given array containing 1 non-canonical empty stream, should return that stream', () => {
+    const s = now(1)
+    is(s, mergeArray([empty(), s, empty()]))
+  })
+
   it('should include items from all inputs', function () {
     return testMerge((s1, s2) => mergeArray([s1, s2]))
   })
diff --git a/packages/core/test/combinator/snapshot-test.js b/packages/core/test/combinator/snapshot-test.js
index 254ae68..c9ff2be 100644
--- a/packages/core/test/combinator/snapshot-test.js
+++ b/packages/core/test/combinator/snapshot-test.js
@@ -1,18 +1,29 @@
 // @flow
 import { describe, it } from 'mocha'
-import { is, eq } from '@briancavalier/assert'
+import { assert, is, eq } from '@briancavalier/assert'
 
 import { throwError } from '../../src/combinator/errors'
 import { at } from '../../src/source/at'
 import { now } from '../../src/source/now'
+import { empty, isCanonicalEmpty } from '../../src/source/empty'
 import { never } from '../../src/source/never'
-import { snapshot } from '../../src/combinator/snapshot'
+import { snapshot, sample } from '../../src/combinator/snapshot'
 
 import { makeEvents, ticks, collectEvents } from '../helper/testEnv'
 
 const rint = n => Math.ceil(n * Math.random())
 
 describe('snapshot', () => {
+  it('given canonical empty sampler, should return canonical empty', () => {
+    const s = snapshot(Array, makeEvents(1, 3), empty())
+    assert(isCanonicalEmpty(s))
+  })
+
+  it('given canonical empty values, should return canonical empty', () => {
+    const s = snapshot(Array, empty(), makeEvents(1, 3))
+    assert(isCanonicalEmpty(s))
+  })
+
   it('should pass in the sampler value', () => {
     const streamValue = Math.random()
     const referenceValue = Math.random()
@@ -61,3 +72,28 @@ describe('snapshot', () => {
       .catch(is(error))
   })
 })
+
+describe('sample', () => {
+  it('given canonical empty sampler, should return canonical empty', () => {
+    const s = sample(makeEvents(1, 3), empty())
+    assert(isCanonicalEmpty(s))
+  })
+
+  it('given canonical empty values, should return canonical empty', () => {
+    const s = sample(empty(), makeEvents(1, 3))
+    assert(isCanonicalEmpty(s))
+  })
+
+  it('use event values from values and event times from sampler', () => {
+    const s = sample(makeEvents(2, 5), makeEvents(1, 5))
+
+    return collectEvents(s, ticks(5))
+      .then(eq([
+        { time: 0, value: 0 },
+        { time: 1, value: 0 },
+        { time: 2, value: 1 },
+        { time: 3, value: 1 },
+        { time: 4, value: 2 }
+      ]))
+  })
+})
diff --git a/packages/core/test/combine-test.js b/packages/core/test/combine-test.js
index 3868f54..72b1d59 100644
--- a/packages/core/test/combine-test.js
+++ b/packages/core/test/combine-test.js
@@ -1,15 +1,31 @@
 import { describe, it } from 'mocha'
-import { eq } from '@briancavalier/assert'
+import { assert, eq } from '@briancavalier/assert'
 
 import { combine, combineArray } from '../src/combinator/combine'
 import { delay } from '../src/combinator/delay'
 import { now } from '../src/source/now'
+import { empty, isCanonicalEmpty } from '../src/source/empty'
 
 import { collectEventsFor, makeEvents, makeEventsFromArray } from './helper/testEnv'
 
 const sentinel = { value: 'sentinel' }
 
 describe('combine', function () {
+  it('given one canonical empty stream, should return canonical empty', () => {
+    const s = combine(Array, now(0), empty())
+    assert(isCanonicalEmpty(s))
+  })
+
+  it('given one canonical empty stream, should return canonical empty', () => {
+    const s = combine(Array, empty(), now(1))
+    assert(isCanonicalEmpty(s))
+  })
+
+  it('given two canonical empty streams, should return canonical empty', () => {
+    const s = combine(Array, empty(), empty())
+    assert(isCanonicalEmpty(s))
+  })
+
   it('should yield initial only after all inputs yield', function () {
     const s1 = now(1)
     const s2 = now(sentinel)
@@ -40,6 +56,16 @@ describe('combine', function () {
 })
 
 describe('combineArray', function () {
+  it('given empty array, should return canonical empty', () => {
+    const s = combineArray(Array, [])
+    assert(isCanonicalEmpty(s))
+  })
+
+  it('given array containing canonical empty, should return canonical empty', () => {
+    const s = combineArray(Array, [now(0), empty(), now(1)])
+    assert(isCanonicalEmpty(s))
+  })
+
   it('should yield initial only after all inputs yield', function () {
     const s1 = now(1)
     const s2 = now(sentinel)
diff --git a/packages/core/test/filter-test.js b/packages/core/test/filter-test.js
index 20b4b9b..ab87fd0 100644
--- a/packages/core/test/filter-test.js
+++ b/packages/core/test/filter-test.js
@@ -1,9 +1,10 @@
 import { describe, it } from 'mocha'
-import { eq } from '@briancavalier/assert'
+import { eq, assert } from '@briancavalier/assert'
 
 import { filter, skipRepeats, skipRepeatsWith } from '../src/combinator/filter'
 
 import { collectEventsFor, makeEventsFromArray } from './helper/testEnv'
+import { empty, isCanonicalEmpty } from '../src/source/empty'
 
 const sentinel = { value: 'sentinel' }
 const other = { value: 'other' }
@@ -20,6 +21,17 @@ describe('filter', function () {
         { time: 2, value: sentinel }
       ]))
   })
+
+  describe('given a canonical empty stream', function () {
+    it('should return a canonical empty stream', function () {
+      // Fixture setup
+      const emptyStream = empty()
+      // Exercise system
+      const sut = filter(_ => _, emptyStream)
+      // Verify outcome
+      assert(isCanonicalEmpty(sut))
+    })
+  })
 })
 
 describe('skipRepeats', function () {
diff --git a/packages/core/test/source/empty-test.js b/packages/core/test/source/empty-test.js
index 6bf0c74..fca2f93 100644
--- a/packages/core/test/source/empty-test.js
+++ b/packages/core/test/source/empty-test.js
@@ -2,21 +2,36 @@
 import { describe, it } from 'mocha'
 import { eq, assert } from '@briancavalier/assert'
 
-import { empty, isCanonicalEmpty } from '../../src/source/empty'
+import { empty, isCanonicalEmpty, containsCanonicalEmpty } from '../../src/source/empty'
+import { now } from '../../src/source/now'
 
 import { collectEventsFor, makeEvents } from '../helper/testEnv'
 
 describe('empty', () => {
-  describe('isEmpty', () => {
-    it('should return true given empty()', () => {
+  describe('isCanonicalEmpty', () => {
+    it('given empty(), should return true', () => {
       assert(isCanonicalEmpty(empty()))
     })
 
-    it('should return false given non-empty()', () => {
+    it('given non-empty(), should return false', () => {
       assert(!isCanonicalEmpty(makeEvents(1, 0)))
     })
   })
 
+  describe('containsCanonicalEmpty', () => {
+    it('given empty array, should return false', () => {
+      assert(!containsCanonicalEmpty([]))
+    })
+
+    it('given array containing  empty(), should return true', () => {
+      assert(containsCanonicalEmpty([now(0), empty(), now(1)]))
+    })
+
+    it('given array containing only non-empty(), should return false', () => {
+      assert(!containsCanonicalEmpty([now(0), now(1)]))
+    })
+  })
+
   describe('empty', () => {
     it('should yield no items before end', () => {
       return collectEventsFor(1, empty()).then(eq([]))
diff --git a/packages/core/test/transform-test.js b/packages/core/test/transform-test.js
index de637c5..8ce9bd8 100644
--- a/packages/core/test/transform-test.js
+++ b/packages/core/test/transform-test.js
@@ -1,11 +1,13 @@
 import { describe, it } from 'mocha'
-import { eq } from '@briancavalier/assert'
+import { eq, assert } from '@briancavalier/assert'
 
 import { assertSame } from './helper/stream-helper'
 import { map, tap, constant } from '../src/combinator/transform'
 import { now } from '../src/source/now'
+import { empty, isCanonicalEmpty } from '../src/source/empty'
 
 import { collectEventsFor } from './helper/testEnv'
+import { id } from '@most/prelude'
 
 describe('map', function () {
   it('should satisfy identity', function () {
@@ -26,6 +28,17 @@ describe('map', function () {
       map(f, map(g, u))
     )
   })
+
+  describe('given a canonical empty stream', function () {
+    it('should return a canonical empty stream', function () {
+      // Fixture setup
+      const emptyStream = empty()
+      // Exercise system
+      const sut = map(id, emptyStream)
+      // Verify outcome
+      assert(isCanonicalEmpty(sut))
+    })
+  })
 })
 
 describe('constant', function () {
diff --git a/packages/core/test/zip-test.js b/packages/core/test/zip-test.js
index f3d601e..435e88b 100644
--- a/packages/core/test/zip-test.js
+++ b/packages/core/test/zip-test.js
@@ -1,11 +1,66 @@
 import { describe, it } from 'mocha'
-import { eq } from '@briancavalier/assert'
+import { assert, eq } from '@briancavalier/assert'
 
-import { zip } from '../src/combinator/zip'
+import { zip, zipArray } from '../src/combinator/zip'
+import { now } from '../src/source/now'
+import { empty, isCanonicalEmpty } from '../src/source/empty'
 
 import { makeEvents, collectEventsFor } from './helper/testEnv'
 
+describe('zipArray', function () {
+  it('given empty array, should return canonical empty', () => {
+    const s = zipArray(Array, [])
+    assert(isCanonicalEmpty(s))
+  })
+
+  it('given array containing canonical empty, should return canonical empty', () => {
+    const s = zipArray(Array, [now(0), empty(), now(1)])
+    assert(isCanonicalEmpty(s))
+  })
+
+  it('should invoke f for each tuple', function () {
+    const n = 3
+    const s = zipArray(Array, [makeEvents(1, n), makeEvents(2, n), makeEvents(3, n)])
+
+    return collectEventsFor(n * 3, s)
+      .then(eq([
+        { time: 0, value: [0, 0, 0] },
+        { time: 3, value: [1, 1, 1] },
+        { time: 6, value: [2, 2, 2] }
+      ]))
+  })
+
+  it('should end when shortest stream ends', function () {
+    const n = 3
+    const a = makeEvents(1, n + 1)
+    const b = makeEvents(1, n)
+    const c = makeEvents(1, n + 1)
+
+    return collectEventsFor(n, zipArray(Array, [a, b, c]))
+      .then(eq([
+        { time: 0, value: [0, 0, 0] },
+        { time: 1, value: [1, 1, 1] },
+        { time: 2, value: [2, 2, 2] }
+      ]))
+  })
+})
+
 describe('zip', function () {
+  it('given one canonical empty stream, should return canonical empty', () => {
+    const s = zip(Array, now(0), empty())
+    assert(isCanonicalEmpty(s))
+  })
+
+  it('given one canonical empty stream, should return canonical empty', () => {
+    const s = zip(Array, empty(), now(1))
+    assert(isCanonicalEmpty(s))
+  })
+
+  it('given two canonical empty streams, should return canonical empty', () => {
+    const s = zip(Array, empty(), empty())
+    assert(isCanonicalEmpty(s))
+  })
+
   it('should invoke f for each tuple', function () {
     const n = 3
     const s = zip(Array, makeEvents(1, n), makeEvents(2, n))