Repository: mostjs/core
Pull request #55 (2017-04-26 23:33:11)

feat(skipAfter): port skipAfter from cujojs/most
Port @davidchase's `skipAfter` implementation and tests from cujojs/most#429 and cujojs/most#431


Patch:
diff --git a/packages/core/src/combinator/combine.js b/packages/core/src/combinator/combine.js
index a0df217..2d516de 100644
--- a/packages/core/src/combinator/combine.js
+++ b/packages/core/src/combinator/combine.js
@@ -64,15 +64,16 @@ class CombineSink extends Pipe {
     const l = sinks.length
     this.awaiting = l
     this.values = new Array(l)
-    this.hasValue = new Array(l)
-    for (let i = 0; i < l; ++i) {
-      this.hasValue[i] = false
-    }
-
+    this.hasValue = new Array(l).fill(false)
     this.activeCount = sinks.length
   }
 
   event (t, indexedValue) {
+    if (!indexedValue.active) {
+      this._dispose(t, indexedValue.index)
+      return
+    }
+
     const i = indexedValue.index
     const awaiting = this._updateReady(i)
 
@@ -92,10 +93,10 @@ class CombineSink extends Pipe {
     return this.awaiting
   }
 
-  end (t, indexedValue) {
-    tryDispose(t, this.disposables[indexedValue.index], this.sink)
+  _dispose (t, index) {
+    tryDispose(t, this.disposables[index], this.sink)
     if (--this.activeCount === 0) {
-      this.sink.end(t, indexedValue.value)
+      this.sink.end(t)
     }
   }
 }
diff --git a/packages/core/src/combinator/continueWith.js b/packages/core/src/combinator/continueWith.js
index 6d77a5d..d1baab4 100644
--- a/packages/core/src/combinator/continueWith.js
+++ b/packages/core/src/combinator/continueWith.js
@@ -35,25 +35,25 @@ class ContinueWithSink extends Pipe {
     this.sink.event(t, x)
   }
 
-  end (t, x) {
+  end (t) {
     if (!this.active) {
       return
     }
 
     tryDispose(t, this.disposable, this.sink)
-    this._startNext(t, x, this.sink)
+    this._startNext(t, this.sink)
   }
 
-  _startNext (t, x, sink) {
+  _startNext (t, sink) {
     try {
-      this.disposable = this._continue(this.f, x, sink)
+      this.disposable = this._continue(this.f, sink)
     } catch (e) {
       sink.error(t, e)
     }
   }
 
-  _continue (f, x, sink) {
-    return f(x).run(sink, this.scheduler)
+  _continue (f, sink) {
+    return f().run(sink, this.scheduler)
   }
 
   dispose () {
diff --git a/packages/core/src/combinator/delay.js b/packages/core/src/combinator/delay.js
index dfff840..3e96766 100644
--- a/packages/core/src/combinator/delay.js
+++ b/packages/core/src/combinator/delay.js
@@ -41,7 +41,7 @@ class DelaySink extends Pipe {
     this.scheduler.delay(this.dt, propagateEventTask(x, this.sink))
   }
 
-  end (t, x) {
-    this.scheduler.delay(this.dt, propagateEndTask(x, this.sink))
+  end (t) {
+    this.scheduler.delay(this.dt, propagateEndTask(this.sink))
   }
 }
diff --git a/packages/core/src/combinator/errors.js b/packages/core/src/combinator/errors.js
index 3b29e78..5edb93c 100644
--- a/packages/core/src/combinator/errors.js
+++ b/packages/core/src/combinator/errors.js
@@ -58,8 +58,8 @@ class RecoverWithSink {
     tryEvent(t, x, this.sink)
   }
 
-  end (t, x) {
-    tryEnd(t, x, this.sink)
+  end (t) {
+    tryEnd(t, this.sink)
   }
 
   error (t, e) {
diff --git a/packages/core/src/combinator/limit.js b/packages/core/src/combinator/limit.js
index 9166acf..73c9bc0 100644
--- a/packages/core/src/combinator/limit.js
+++ b/packages/core/src/combinator/limit.js
@@ -88,12 +88,12 @@ class DebounceSink {
     this.timer = this.scheduler.delay(this.dt, propagateEventTask(x, this.sink))
   }
 
-  end (t, x) {
+  end (t) {
     if (this._clearTimer()) {
       this.sink.event(t, this.value)
-      this.value = void 0
+      this.value = undefined
     }
-    this.sink.end(t, x)
+    this.sink.end(t)
   }
 
   error (t, x) {
diff --git a/packages/core/src/combinator/loop.js b/packages/core/src/combinator/loop.js
index b469a0e..2ff2a22 100644
--- a/packages/core/src/combinator/loop.js
+++ b/packages/core/src/combinator/loop.js
@@ -42,9 +42,4 @@ class LoopSink extends Pipe {
     this.seed = result.seed
     this.sink.event(t, result.value)
   }
-
-  end (t) {
-    this.sink.end(t, this.seed)
-  }
 }
-
diff --git a/packages/core/src/combinator/merge.js b/packages/core/src/combinator/merge.js
index 28b401b..477ab25 100644
--- a/packages/core/src/combinator/merge.js
+++ b/packages/core/src/combinator/merge.js
@@ -72,13 +72,17 @@ class MergeSink extends Pipe {
   }
 
   event (t, indexValue) {
+    if (!indexValue.active) {
+      this._dispose(t, indexValue.index)
+      return
+    }
     this.sink.event(t, indexValue.value)
   }
 
-  end (t, indexedValue) {
-    tryDispose(t, this.disposables[indexedValue.index], this.sink)
+  _dispose (t, index) {
+    tryDispose(t, this.disposables[index], this.sink)
     if (--this.activeCount === 0) {
-      this.sink.end(t, indexedValue.value)
+      this.sink.end(t)
     }
   }
 }
diff --git a/packages/core/src/combinator/mergeConcurrently.js b/packages/core/src/combinator/mergeConcurrently.js
index 05d0ea2..a722916 100644
--- a/packages/core/src/combinator/mergeConcurrently.js
+++ b/packages/core/src/combinator/mergeConcurrently.js
@@ -62,10 +62,10 @@ class Outer {
     this.current.add(innerSink)
   }
 
-  end (t, x) {
+  end (t) {
     this.active = false
     tryDispose(t, this.disposable, this.sink)
-    this._checkEnd(t, x)
+    this._checkEnd(t)
   }
 
   error (t, e) {
@@ -80,20 +80,20 @@ class Outer {
     this.current.dispose()
   }
 
-  _endInner (t, x, inner) {
+  _endInner (t, inner) {
     this.current.remove(inner)
     tryDispose(t, inner, this)
 
     if (this.pending.length === 0) {
-      this._checkEnd(t, x)
+      this._checkEnd(t)
     } else {
       this._startInner(t, this.pending.shift())
     }
   }
 
-  _checkEnd (t, x) {
+  _checkEnd (t) {
     if (!this.active && this.current.isEmpty()) {
-      this.sink.end(t, x)
+      this.sink.end(t)
     }
   }
 }
@@ -114,8 +114,8 @@ class Inner {
     this.sink.event(Math.max(t, this.time), x)
   }
 
-  end (t, x) {
-    this.outer._endInner(Math.max(t, this.time), x, this)
+  end (t) {
+    this.outer._endInner(Math.max(t, this.time), this)
   }
 
   error (t, e) {
diff --git a/packages/core/src/combinator/promises.js b/packages/core/src/combinator/promises.js
index 54cc20d..1c73e07 100644
--- a/packages/core/src/combinator/promises.js
+++ b/packages/core/src/combinator/promises.js
@@ -39,7 +39,7 @@ class AwaitSink {
 
     // Pre-create closures, to avoid creating them per event
     this._eventBound = x => this.sink.event(this.scheduler.now(), x)
-    this._endBound = x => this.sink.end(this.scheduler.now(), x)
+    this._endBound = () => this.sink.end(this.scheduler.now())
     this._errorBound = e => this.sink.error(this.scheduler.now(), e)
   }
 
@@ -48,8 +48,8 @@ class AwaitSink {
       .catch(this._errorBound)
   }
 
-  end (t, x) {
-    this.queue = this.queue.then(() => this._end(x))
+  end (t) {
+    this.queue = this.queue.then(this._endBound)
       .catch(this._errorBound)
   }
 
@@ -62,8 +62,4 @@ class AwaitSink {
   _event (promise) {
     return promise.then(this._eventBound)
   }
-
-  _end (x) {
-    return Promise.resolve(x).then(this._endBound)
-  }
 }
diff --git a/packages/core/src/combinator/slice.js b/packages/core/src/combinator/slice.js
index 341ccac..2a34b56 100644
--- a/packages/core/src/combinator/slice.js
+++ b/packages/core/src/combinator/slice.js
@@ -79,7 +79,7 @@ class SliceSink extends Pipe {
     this.take -= 1
     this.sink.event(t, x)
     if (this.take === 0) {
-      this.sink.end(t, x)
+      this.sink.end(t)
     }
   }
 }
@@ -116,7 +116,7 @@ class TakeWhileSink extends Pipe {
     if (this.active) {
       this.sink.event(t, x)
     } else {
-      this.sink.end(t, x)
+      this.sink.end(t)
     }
   }
 }
@@ -154,3 +154,39 @@ class SkipWhileSink extends Pipe {
     this.sink.event(t, x)
   }
 }
+
+export const skipAfter = (p, stream) =>
+  new SkipAfter(p, stream)
+
+class SkipAfter {
+  constructor (p, source) {
+    this.p = p
+    this.source = source
+  }
+
+  run (sink, scheduler) {
+    return this.source.run(new SkipAfterSink(this.p, sink), scheduler)
+  }
+}
+
+class SkipAfterSink extends Pipe {
+  constructor (p, sink) {
+    super(sink)
+    this.p = p
+    this.skipping = false
+  }
+
+  event (t, x) {
+    if (this.skipping) {
+      return
+    }
+
+    const p = this.p
+    this.skipping = p(x)
+    this.sink.event(t, x)
+
+    if (this.skipping) {
+      this.sink.end(t)
+    }
+  }
+}
diff --git a/packages/core/src/combinator/switch.js b/packages/core/src/combinator/switch.js
index c4aa48c..a3c2f13 100644
--- a/packages/core/src/combinator/switch.js
+++ b/packages/core/src/combinator/switch.js
@@ -37,9 +37,9 @@ class SwitchSink {
     this.current.disposable = stream.run(this.current, this.scheduler)
   }
 
-  end (t, x) {
+  end (t) {
     this.ended = true
-    this._checkEnd(t, x)
+    this._checkEnd(t)
   }
 
   error (t, e) {
@@ -64,15 +64,15 @@ class SwitchSink {
     }
   }
 
-  _checkEnd (t, x) {
+  _checkEnd (t) {
     if (this.ended && this.current === null) {
-      this.sink.end(t, x)
+      this.sink.end(t)
     }
   }
 
-  _endInner (t, x, inner) {
+  _endInner (t, inner) {
     this._disposeInner(t, inner)
-    this._checkEnd(t, x)
+    this._checkEnd(t)
   }
 
   _errorInner (t, e, inner) {
@@ -96,8 +96,8 @@ class Segment {
     }
   }
 
-  end (t, x) {
-    this.outer._endInner(Math.max(t, this.min), x, this)
+  end (t) {
+    this.outer._endInner(Math.max(t, this.min), this)
   }
 
   error (t, e) {
diff --git a/packages/core/src/combinator/timeslice.js b/packages/core/src/combinator/timeslice.js
index de543b2..33005ad 100644
--- a/packages/core/src/combinator/timeslice.js
+++ b/packages/core/src/combinator/timeslice.js
@@ -101,7 +101,7 @@ class UpperBound extends Pipe {
   event (t, x) {
     if (t < this.value) {
       this.value = t
-      this.sink.end(t, x)
+      this.sink.end(t)
     }
   }
 
diff --git a/packages/core/src/combinator/zip.js b/packages/core/src/combinator/zip.js
index 111c96e..d669401 100644
--- a/packages/core/src/combinator/zip.js
+++ b/packages/core/src/combinator/zip.js
@@ -69,7 +69,13 @@ class ZipSink extends Pipe {
     this.buffers = buffers
   }
 
-  event (t, indexedValue) { // eslint-disable-line complexity
+  event (t, indexedValue) {
+    /* eslint complexity: [1, 5] */
+    if (!indexedValue.active) {
+      this._dispose(t, indexedValue.index)
+      return
+    }
+
     const buffers = this.buffers
     const buffer = buffers[indexedValue.index]
 
@@ -83,15 +89,15 @@ class ZipSink extends Pipe {
       emitZipped(this.f, t, buffers, this.sink)
 
       if (ended(this.buffers, this.sinks)) {
-        this.sink.end(t, void 0)
+        this.sink.end(t)
       }
     }
   }
 
-  end (t, indexedValue) {
-    const buffer = this.buffers[indexedValue.index]
+  _dispose (t, index) {
+    const buffer = this.buffers[index]
     if (buffer.isEmpty()) {
-      this.sink.end(t, indexedValue.value)
+      this.sink.end(t)
     }
   }
 }
diff --git a/packages/core/src/index.js b/packages/core/src/index.js
index 0d1db5d..21561ee 100644
--- a/packages/core/src/index.js
+++ b/packages/core/src/index.js
@@ -169,10 +169,10 @@ import {
   propagateTask as _propagateTask,
   propagateEventTask as _propagateEventTask,
   propagateErrorTask as _propagateErrorTask,
-  propagateEndTask as _propagateEndTask
+  propagateEndTask
 } from './scheduler/PropagateTask'
 
 export const propagateTask = curry3(_propagateTask)
 export const propagateEventTask = curry2(_propagateEventTask)
 export const propagateErrorTask = curry2(_propagateErrorTask)
-export const propagateEndTask = curry2(_propagateEndTask)
+export { propagateEndTask }
diff --git a/packages/core/src/index.js.flow b/packages/core/src/index.js.flow
index 8a45801..9eed0a7 100644
--- a/packages/core/src/index.js.flow
+++ b/packages/core/src/index.js.flow
@@ -6,7 +6,7 @@ export type SeedValue <S, A> = {
   value: A
 }
 
-declare export function runEffects <A> (s: Stream<A>, scheduler: Scheduler): Promise<any>
+declare export function runEffects <A> (s: Stream<A>, scheduler: Scheduler): Promise<void>
 
 declare export function empty (): Stream<any>
 declare export function never (): Stream<any>
@@ -151,6 +151,9 @@ declare export function takeWhile <A> (p: (A) => boolean): (s: Stream<A>) => Str
 declare export function skipWhile <A> (p: (A) => boolean, s: Stream<A>): Stream<A>
 declare export function skipWhile <A> (p: (A) => boolean): (s: Stream<A>) => Stream<A>
 
+declare export function skipAfter <A> (p: (A) => boolean, s: Stream<A>): Stream<A>
+declare export function skipAfter <A> (p: (A) => boolean): (s: Stream<A>) => Stream<A>
+
 declare export function until <A> (until: Stream<any>, s: Stream<A>): Stream<A>
 declare export function until <A> (until: Stream<any>): (s: Stream<A>) => Stream<A>
 
@@ -187,4 +190,4 @@ export type PropagateTaskRun<A> = (time: Time, value: A, sink: Sink<A>, task: Pr
 declare export function propagateTask<A> (run: PropagateTaskRun<A>, value: A, sink: Sink<A>): PropagateTask<A>
 declare export function propagateEventTask<A> (value: A, sink: Sink<A>): PropagateTask<A>
 declare export function propagateErrorTask<E: Error> (e: E, sink: Sink<any>): PropagateTask<any>
-declare export function propagateEndTask (value: any, sink: Sink<any>): PropagateTask<any>
+declare export function propagateEndTask (sink: Sink<any>): PropagateTask<void>
diff --git a/packages/core/src/runEffects.js b/packages/core/src/runEffects.js
index bd26bfe..1c4ee30 100644
--- a/packages/core/src/runEffects.js
+++ b/packages/core/src/runEffects.js
@@ -23,11 +23,11 @@ class RunEffectsSink {
 
   event (t, x) {}
 
-  end (t, x) {
+  end (t) {
     if (!this.active) {
       return
     }
-    this._dispose(this._error, this._end, x)
+    this._dispose(this._error, this._end, undefined)
   }
 
   error (t, e) {
diff --git a/packages/core/src/scheduler/PropagateTask.js b/packages/core/src/scheduler/PropagateTask.js
index bd1523e..5a144ee 100644
--- a/packages/core/src/scheduler/PropagateTask.js
+++ b/packages/core/src/scheduler/PropagateTask.js
@@ -8,7 +8,7 @@ export const propagateTask = (run, value, sink) => new PropagateTask(run, value,
 
 export const propagateEventTask = (value, sink) => propagateTask(runEvent, value, sink)
 
-export const propagateEndTask = (value, sink) => propagateTask(runEnd, value, sink)
+export const propagateEndTask = sink => propagateTask(runEnd, undefined, sink)
 
 export const propagateErrorTask = (value, sink) => propagateTask(runError, value, sink)
 
@@ -43,6 +43,6 @@ export class PropagateTask {
 
 const runEvent = (t, x, sink) => sink.event(t, x)
 
-const runEnd = (t, x, sink) => sink.end(t, x)
+const runEnd = (t, _, sink) => sink.end(t)
 
 const runError = (t, e, sink) => sink.error(t, e)
diff --git a/packages/core/src/sink/IndexSink.js b/packages/core/src/sink/IndexSink.js
index ada08fa..8dfb121 100644
--- a/packages/core/src/sink/IndexSink.js
+++ b/packages/core/src/sink/IndexSink.js
@@ -4,27 +4,27 @@
 
 import Sink from './Pipe'
 
-export default function IndexSink (i, sink) {
-  this.sink = sink
-  this.index = i
-  this.active = true
-  this.value = void 0
-}
+export default class IndexSink extends Sink {
+  constructor (i, sink) {
+    super(sink)
+    this.index = i
+    this.active = true
+    this.value = undefined
+  }
 
-IndexSink.prototype.event = function (t, x) {
-  if (!this.active) {
-    return
+  event (t, x) {
+    if (!this.active) {
+      return
+    }
+    this.value = x
+    this.sink.event(t, this)
   }
-  this.value = x
-  this.sink.event(t, this)
-}
 
-IndexSink.prototype.end = function (t, x) {
-  if (!this.active) {
-    return
+  end (t) {
+    if (!this.active) {
+      return
+    }
+    this.active = false
+    this.sink.event(t, this)
   }
-  this.active = false
-  this.sink.end(t, { index: this.index, value: x })
 }
-
-IndexSink.prototype.error = Sink.prototype.error
diff --git a/packages/core/src/sink/Pipe.js b/packages/core/src/sink/Pipe.js
index e7064ae..67aeb88 100644
--- a/packages/core/src/sink/Pipe.js
+++ b/packages/core/src/sink/Pipe.js
@@ -1,14 +1,7 @@
-/** @license MIT License (c) copyright 2010-2016 original author or authors */
+/** @license MIT License (c) copyright 2010-2017 original author or authors */
 /** @author Brian Cavalier */
-/** @author John Hann */
 
 export default class Pipe {
-  /**
-   * A sink mixin that simply forwards event, end, and error to
-   * another sink.
-   * @param sink
-   * @constructor
-   */
   constructor (sink) {
     this.sink = sink
   }
@@ -17,12 +10,11 @@ export default class Pipe {
     return this.sink.event(t, x)
   }
 
-  end (t, x) {
-    return this.sink.end(t, x)
+  end (t) {
+    return this.sink.end(t)
   }
 
   error (t, e) {
     return this.sink.error(t, e)
   }
 }
-
diff --git a/packages/core/src/source/core.js b/packages/core/src/source/core.js
index 63439fb..f9418c0 100644
--- a/packages/core/src/source/core.js
+++ b/packages/core/src/source/core.js
@@ -35,7 +35,7 @@ export const empty = () => EMPTY
 
 class Empty {
   run (sink, scheduler) {
-    return scheduler.asap(propagateEndTask(undefined, sink))
+    return scheduler.asap(propagateEndTask(sink))
   }
 }
 
diff --git a/packages/core/src/source/fromArray.js b/packages/core/src/source/fromArray.js
index b87870b..1145149 100644
--- a/packages/core/src/source/fromArray.js
+++ b/packages/core/src/source/fromArray.js
@@ -22,9 +22,5 @@ function runProducer (t, array, sink, task) {
     sink.event(t, array[i])
   }
 
-  task.active && end(t)
-
-  function end (t) {
-    sink.end(t)
-  }
+  task.active && sink.end(t)
 }
diff --git a/packages/core/src/source/fromIterable.js b/packages/core/src/source/fromIterable.js
index 202cf01..a3be31e 100644
--- a/packages/core/src/source/fromIterable.js
+++ b/packages/core/src/source/fromIterable.js
@@ -24,5 +24,5 @@ function runProducer (t, iterator, sink, task) {
     r = iterator.next()
   }
 
-  task.active && sink.end(t, r.value)
+  task.active && sink.end(t)
 }
diff --git a/packages/core/src/source/generate.js b/packages/core/src/source/generate.js
index 5456962..aa4c72e 100644
--- a/packages/core/src/source/generate.js
+++ b/packages/core/src/source/generate.js
@@ -47,7 +47,7 @@ const next = (generate, x) =>
 
 function handle (generate, result) {
   if (result.done) {
-    return generate.sink.end(generate.scheduler.now(), result.value)
+    return generate.sink.end(generate.scheduler.now())
   }
 
   return Promise.resolve(result.value).then(function (x) {
@@ -65,4 +65,3 @@ function emit (generate, x) {
 function error (generate, e) {
   return handle(generate, generate.iterator.throw(e))
 }
-
diff --git a/packages/core/src/source/tryEvent.js b/packages/core/src/source/tryEvent.js
index 5718574..20e6194 100644
--- a/packages/core/src/source/tryEvent.js
+++ b/packages/core/src/source/tryEvent.js
@@ -10,9 +10,9 @@ export function tryEvent (t, x, sink) {
   }
 }
 
-export function tryEnd (t, x, sink) {
+export function tryEnd (t, sink) {
   try {
-    sink.end(t, x)
+    sink.end(t)
   } catch (e) {
     sink.error(t, e)
   }
diff --git a/packages/core/src/source/unfold.js b/packages/core/src/source/unfold.js
index 9843549..d1abb3e 100644
--- a/packages/core/src/source/unfold.js
+++ b/packages/core/src/source/unfold.js
@@ -55,7 +55,7 @@ function stepUnfold (unfold, x) {
 
 function continueUnfold (unfold, tuple) {
   if (tuple.done) {
-    unfold.sink.end(unfold.scheduler.now(), tuple.value)
+    unfold.sink.end(unfold.scheduler.now())
     return tuple.value
   }
 
diff --git a/packages/core/type-definitions/PropagateTask.d.ts b/packages/core/type-definitions/PropagateTask.d.ts
index 4c2f538..4c4c230 100644
--- a/packages/core/type-definitions/PropagateTask.d.ts
+++ b/packages/core/type-definitions/PropagateTask.d.ts
@@ -8,8 +8,7 @@ export function propagateTask<T>(run: PropagateTaskRun<T>): (value: T) => (sink:
 export function propagateEventTask<T>(value: T, sink: Sink<T>): PropagateTask<T>;
 export function propagateEventTask<T>(value: T): (sink: Sink<T>) => PropagateTask<T>;
 
-export function propagateEndTask<T>(value: T | void, sink: Sink<T>): PropagateTask<T>;
-export function propagateEndTask<T>(value: T | void): (sink: Sink<T>) => PropagateTask<T>;
+export function propagateEndTask<T>(sink: Sink<T>): PropagateTask<void>;
 
 export function propagateErrorTask(error: Error, sink: Sink<any>): PropagateTask<any>;
 export function propagateErrorTask(error: Error): (sink: Sink<any>) => PropagateTask<any>;
diff --git a/packages/core/type-definitions/combinator/slice.d.ts b/packages/core/type-definitions/combinator/slice.d.ts
index 69648cb..60a22e3 100644
--- a/packages/core/type-definitions/combinator/slice.d.ts
+++ b/packages/core/type-definitions/combinator/slice.d.ts
@@ -12,6 +12,9 @@ export function takeWhile<A>(p: (a: A) => boolean): (s: Stream<A>) => Stream<A>;
 export function skipWhile<A>(p: (a: A) => boolean, s: Stream<A>): Stream<A>;
 export function skipWhile<A>(p: (a: A) => boolean): (s: Stream<A>) => Stream<A>;
 
+export function skipAfter<A>(p: (a: A) => boolean, s: Stream<A>): Stream<A>;
+export function skipAfter<A>(p: (a: A) => boolean): (s: Stream<A>) => Stream<A>;
+
 export function slice<A>(start: number, end: number, s: Stream<A>): Stream<A>;
 export function slice<A>(start: number): (end: number, s: Stream<A>) => Stream<A>;
 export function slice<A>(start: number, end: number): (s: Stream<A>) => Stream<A>;
diff --git a/packages/core/type-definitions/runEffects.d.ts b/packages/core/type-definitions/runEffects.d.ts
index 29e44de..ef43259 100644
--- a/packages/core/type-definitions/runEffects.d.ts
+++ b/packages/core/type-definitions/runEffects.d.ts
@@ -1,4 +1,4 @@
 import { Scheduler, Stream } from '@most/types';
 
-export function runEffects(stream: Stream<any>, scheduler: Scheduler): Promise<any>;
-export function runEffects(stream: Stream<any>): (scheduler: Scheduler) => Promise<any>;
+export function runEffects <T> (stream: Stream<T>, scheduler: Scheduler): Promise<void>;
+export function runEffects <T> (stream: Stream<T>): (scheduler: Scheduler) => Promise<void>;
diff --git a/packages/types/index.d.ts b/packages/types/index.d.ts
index e1cd54e..061c532 100644
--- a/packages/types/index.d.ts
+++ b/packages/types/index.d.ts
@@ -9,7 +9,7 @@ export interface Stream<A> {
 
 export interface Sink<A> {
   event(time: Time, value: A): void;
-  end(time: Time, value?: A): void;
+  end(time: Time): void;
   error(time: Time, err: Error): void;
 }
 
diff --git a/packages/types/index.js.flow b/packages/types/index.js.flow
index 0d4ae0d..f98d33d 100644
--- a/packages/types/index.js.flow
+++ b/packages/types/index.js.flow
@@ -11,7 +11,7 @@ export type Stream<A> = {
 
 export type Sink<A> = {
   event: (Time, A) => void,
-  end: (Time, any) => void,
+  end: (Time) => void,
   error: (Time, Error) => void,
 }
 


Test patch:
diff --git a/packages/core/test/combinator/scan-test.js b/packages/core/test/combinator/scan-test.js
index 857a50b..b068a93 100644
--- a/packages/core/test/combinator/scan-test.js
+++ b/packages/core/test/combinator/scan-test.js
@@ -7,7 +7,6 @@ import { scan } from '../../src/combinator/scan'
 import { runEffects } from '../../src/runEffects'
 import { atTime, makeEventsFromArray, collectEventsFor, ticks } from '../helper/testEnv'
 import FakeDisposeStream from '../helper/FakeDisposeStream'
-import { endWith } from '../helper/endWith'
 
 const sentinel = { value: 'sentinel' }
 
@@ -28,13 +27,6 @@ describe('scan', function () {
       .then(eq(expected))
   })
 
-  it('should preserve end value', function () {
-    const stream = endWith(sentinel, atTime(0, {}))
-    const s = scan((a, x) => x, {}, stream)
-
-    return runEffects(s, ticks(1)).then(eq(sentinel))
-  })
-
   it('should dispose', function () {
     const dispose = spy()
 
diff --git a/packages/core/test/helper/testEnv.js b/packages/core/test/helper/testEnv.js
index 8381175..ca54362 100644
--- a/packages/core/test/helper/testEnv.js
+++ b/packages/core/test/helper/testEnv.js
@@ -53,7 +53,7 @@ class AtTimes {
 
 const runEvents = (events, sink, scheduler) => {
   const s = events.reduce(appendEvent(sink, scheduler), { tasks: [], time: 0 })
-  const end = scheduler.delay(s.time, propagateEndTask(undefined, sink))
+  const end = scheduler.delay(s.time, propagateEndTask(sink))
   return disposeWith(cancelAll, s.tasks.concat(end))
 }
 
diff --git a/packages/core/test/slice-test.js b/packages/core/test/slice-test.js
index f042ed8..46e64dc 100644
--- a/packages/core/test/slice-test.js
+++ b/packages/core/test/slice-test.js
@@ -1,11 +1,13 @@
 import { describe, it } from 'mocha'
 import { eq, is, assert } from '@briancavalier/assert'
 
-import { slice, take, skip, takeWhile, skipWhile } from '../src/combinator/slice'
+import { slice, take, skip, takeWhile, skipWhile, skipAfter } from '../src/combinator/slice'
 import { map } from '../src/combinator/transform'
+import { empty } from '../src/source/core'
 import { default as Map } from '../src/fusion/Map'
 
 import { makeEventsFromArray, collectEventsFor, makeEvents } from './helper/testEnv'
+import { assertSame } from './helper/stream-helper'
 
 describe('slice', function () {
   describe('fusion', function () {
@@ -34,68 +36,97 @@ describe('slice', function () {
           { time: 2, value: 2 }
         ]))
     })
+
+    it('should retain only sliced range', function () {
+      const a = [0, 1, 2, 3, 4, 5, 6, 7, 8]
+      const s = slice(2, a.length - 2, makeEventsFromArray(1, a))
+      return collectEventsFor(a.length - 2, s)
+        .then(eq([
+          { time: 2, value: 2 },
+          { time: 3, value: 3 },
+          { time: 4, value: 4 },
+          { time: 5, value: 5 },
+          { time: 6, value: 6 }
+        ]))
+    })
   })
 
-  it('should retain only sliced range', function () {
-    const a = [0, 1, 2, 3, 4, 5, 6, 7, 8]
-    const s = slice(2, a.length - 2, makeEventsFromArray(1, a))
-    return collectEventsFor(a.length - 2, s)
-      .then(eq([
-        { time: 2, value: 2 },
-        { time: 3, value: 3 },
-        { time: 4, value: 4 },
-        { time: 5, value: 5 },
-        { time: 6, value: 6 }
-      ]))
+  describe('take', function () {
+    it('should take first n elements', function () {
+      const n = 2
+      const s = take(n, makeEvents(1, 10))
+      return collectEventsFor(n, s)
+        .then(eq([
+          { time: 0, value: 0 },
+          { time: 1, value: 1 }
+        ]))
+    })
   })
-})
 
-describe('take', function () {
-  it('should take first n elements', function () {
-    const n = 2
-    const s = take(n, makeEvents(1, 10))
-    return collectEventsFor(n, s)
-      .then(eq([
-        { time: 0, value: 0 },
-        { time: 1, value: 1 }
-      ]))
+  describe('skip', function () {
+    it('should skip first n elements', function () {
+      const n = 4
+      const s = skip(2, makeEvents(1, n))
+      return collectEventsFor(n, s)
+        .then(eq([
+          { time: 2, value: 2 },
+          { time: 3, value: 3 }
+        ]))
+    })
   })
-})
 
-describe('skip', function () {
-  it('should skip first n elements', function () {
-    const n = 4
-    const s = skip(2, makeEvents(1, n))
-    return collectEventsFor(n, s)
-      .then(eq([
-        { time: 2, value: 2 },
-        { time: 3, value: 3 }
-      ]))
+  describe('takeWhile', function () {
+    it('should take elements until condition becomes false', function () {
+      const n = 2
+      const p = x => x < n
+      const s = takeWhile(p, makeEvents(1, 10))
+      return collectEventsFor(n, s)
+        .then(eq([
+          { time: 0, value: 0 },
+          { time: 1, value: 1 }
+        ]))
+    })
   })
-})
 
-describe('takeWhile', function () {
-  it('should take elements until condition becomes false', function () {
-    const n = 2
-    const p = x => x < n
-    const s = takeWhile(p, makeEvents(1, 10))
-    return collectEventsFor(n, s)
-      .then(eq([
-        { time: 0, value: 0 },
-        { time: 1, value: 1 }
-      ]))
+  describe('skipWhile', function () {
+    it('should skip elements until condition becomes false', function () {
+      const n = 4
+      const p = x => x < 2
+      const s = skipWhile(p, makeEvents(1, n))
+      return collectEventsFor(n, s)
+        .then(eq([
+          { time: 2, value: 2 },
+          { time: 3, value: 3 }
+        ]))
+    })
   })
-})
 
-describe('skipWhile', function () {
-  it('should skip elements until condition becomes false', function () {
-    const n = 4
-    const p = x => x < 2
-    const s = skipWhile(p, makeEvents(1, n))
-    return collectEventsFor(n, s)
-      .then(eq([
-        { time: 2, value: 2 },
-        { time: 3, value: 3 }
-      ]))
+  describe('skipAfter', function () {
+    it('should be empty if source stream is empty', function () {
+      const s = skipAfter(_ => false, empty())
+
+      return collectEventsFor(10, s).then(eq([]))
+    })
+
+    it('should skip all elements after the first one for which the condition is true', function () {
+      const n = 5
+      const n2 = n * 2
+      const s = skipAfter(x => x === 5, makeEvents(1, n2))
+
+      return collectEventsFor(n2, s)
+        .then(eq([
+          { time: 0, value: 0 },
+          { time: 1, value: 1 },
+          { time: 2, value: 2 },
+          { time: 3, value: 3 },
+          { time: 4, value: 4 },
+          { time: 5, value: 5 }
+        ]))
+    })
+
+    it('should contain all elements when condition is false', function () {
+      const s = makeEvents(1, 10)
+      return assertSame(s, skipAfter(_ => false, s))
+    })
   })
 })
diff --git a/packages/core/test/source/generate-test.js b/packages/core/test/source/generate-test.js
index 31fb62e..b399ee4 100644
--- a/packages/core/test/source/generate-test.js
+++ b/packages/core/test/source/generate-test.js
@@ -2,23 +2,19 @@ import { describe, it } from 'mocha'
 import { eq } from '@briancavalier/assert'
 
 import { generate } from '../../src/source/generate'
-import { reduce } from '../helper/reduce'
+import { observe } from '../helper/observe'
 import { getIterator } from '../../src/iterable'
 import ArrayIterable from '../helper/ArrayIterable'
 import delayPromise from '../helper/delayPromise'
 
-function makeAsyncIterator (ms, n) {
-  const a = new Array(n)
-  for (let i = 0; i < n; ++i) {
-    a[i] = delayPromise(ms, i)
-  }
-
-  return getIterator(new ArrayIterable(a))
-}
+const makeAsyncIterator = (ms, events) =>
+  getIterator(new ArrayIterable(events.map(x => delayPromise(ms, x))))
 
 describe('generate', function () {
   it('should contain iterable items', function () {
-    const s = generate(makeAsyncIterator, 10, 5)
-    return reduce((a, x) => a.concat(x), [], s).then(eq([0, 1, 2, 3, 4]))
+    const expected = [0, 1, 2, 3]
+    const actual = []
+    const s = generate(makeAsyncIterator, 10, expected)
+    return observe(x => actual.push(x), s).then(() => eq(expected, actual))
   })
 })
diff --git a/packages/core/test/timeslice-test.js b/packages/core/test/timeslice-test.js
index 8976bf0..1e87016 100644
--- a/packages/core/test/timeslice-test.js
+++ b/packages/core/test/timeslice-test.js
@@ -1,5 +1,5 @@
 import { describe, it } from 'mocha'
-import { eq, is, assert } from '@briancavalier/assert'
+import { eq, assert } from '@briancavalier/assert'
 
 import { until, since, during } from '../src/combinator/timeslice'
 import { take } from '../src/combinator/slice'
@@ -7,15 +7,11 @@ import { periodic } from '../src/source/periodic'
 import { just, never } from '../src/source/core'
 import { delay } from '../src/combinator/delay'
 
-import { runEffects } from '../src/runEffects'
 import { ticks, collectEvents, collectEventsFor, makeEvents } from './helper/testEnv'
 import FakeDisposeStream from './helper/FakeDisposeStream'
-import { endWith } from './helper/endWith'
 
 import { spy } from 'sinon'
 
-const sentinel = { value: 'sentinel' }
-
 describe('during', function () {
   it('should contain events at or later than min and earlier than max', function () {
     const stream = take(10, periodic(1))
@@ -112,15 +108,6 @@ describe('until', function () {
         assert(dispose.calledOnce)
       })
   })
-
-  it('should use until value as end value', function () {
-    const stream = periodic(1)
-    const end = delay(3, just(sentinel))
-
-    const s = until(end, stream)
-    return runEffects(s, ticks(5))
-      .then(is(sentinel))
-  })
 })
 
 describe('since', function () {
@@ -146,23 +133,4 @@ describe('since', function () {
     return collectEventsFor(5, s)
       .then(() => assert(dispose.calledOnce))
   })
-
-  it('should preserve end value', function () {
-    const stream = endWith(sentinel, take(3, periodic(1)))
-    const start = delay(3, just())
-
-    const s = since(start, stream)
-    return runEffects(s, ticks(3))
-      .then(is(sentinel))
-  })
-
-  it('should preserve end value if end signal occurs before start signal', function () {
-    const stream = take(3, periodic(1))
-    const start = delay(3, just())
-    const end = delay(1, just(sentinel))
-
-    const s = since(start, until(end, stream))
-    return runEffects(s, ticks(3))
-      .then(is(sentinel))
-  })
 })