Repository: mostjs/core
Pull request #45 (2017-02-24 13:27:15)

feat(stream): Remove Stream constructor
This PR makes Stream a typeclass/interface, which removes a level of indirection across the entire library.  Streams directly implement run(sink, scheduler) rather than being a useless wrapper around a Source which implements run(sink, scheduler).

⚠️ Interop with cujojs/most requires cujojs/most#413 to be merged first.

### Todo

- [x] Update TS and flow type definitions


Patch:
diff --git a/packages/core/src/Stream.js b/packages/core/src/Stream.js
deleted file mode 100644
index 07b8230..0000000
--- a/packages/core/src/Stream.js
+++ /dev/null
@@ -1,7 +0,0 @@
-/** @license MIT License (c) copyright 2010-2016 original author or authors */
-/** @author Brian Cavalier */
-/** @author John Hann */
-
-export default function Stream (source) {
-  this.source = source
-}
diff --git a/packages/core/src/combinator/combine.js b/packages/core/src/combinator/combine.js
index 8b952bd..8f43da7 100644
--- a/packages/core/src/combinator/combine.js
+++ b/packages/core/src/combinator/combine.js
@@ -2,18 +2,14 @@
 /** @author Brian Cavalier */
 /** @author John Hann */
 
-import Stream from '../Stream'
-import * as transform from './transform'
-import * as core from '../source/core'
+import { map } from './transform'
+import { empty } from '../source/core'
 import Pipe from '../sink/Pipe'
 import IndexSink from '../sink/IndexSink'
-import * as dispose from '../disposable/dispose'
-import * as base from '@most/prelude'
+import { all, tryDispose } from '../disposable/dispose'
+import { tail } from '@most/prelude'
 import invoke from '../invoke'
 
-var map = base.map
-var tail = base.tail
-
 /**
  * Combine latest events from all input streams
  * @param {function(...events):*} f function to combine most recent events
@@ -31,83 +27,76 @@ export function combine (f /*, ...streams */) {
 * @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 combineArray (f, streams) {
-  var l = streams.length
-  return l === 0 ? core.empty()
-  : l === 1 ? transform.map(f, streams[0])
-  : new Stream(combineSources(f, streams))
-}
-
-function combineSources (f, streams) {
-  return new Combine(f, map(getSource, streams))
-}
-
-function getSource (stream) {
-  return stream.source
-}
+export const combineArray = (f, streams) =>
+  streams.length === 0 ? empty()
+    : streams.length === 1 ? map(f, streams[0])
+    : new Combine(f, streams)
+
+class Combine {
+  constructor (f, sources) {
+    this.f = f
+    this.sources = sources
+  }
 
-function Combine (f, sources) {
-  this.f = f
-  this.sources = sources
-}
+  run (sink, scheduler) {
+    const l = this.sources.length
+    const disposables = new Array(l)
+    const sinks = new Array(l)
 
-Combine.prototype.run = function (sink, scheduler) {
-  var l = this.sources.length
-  var disposables = new Array(l)
-  var sinks = new Array(l)
+    const mergeSink = new CombineSink(disposables, sinks, sink, this.f)
 
-  var mergeSink = new CombineSink(disposables, sinks, sink, this.f)
+    for (let indexSink, i = 0; i < l; ++i) {
+      indexSink = sinks[i] = new IndexSink(i, mergeSink)
+      disposables[i] = this.sources[i].run(indexSink, scheduler)
+    }
 
-  for (var indexSink, i = 0; i < l; ++i) {
-    indexSink = sinks[i] = new IndexSink(i, mergeSink)
-    disposables[i] = this.sources[i].run(indexSink, scheduler)
+    return all(disposables)
   }
-
-  return dispose.all(disposables)
 }
 
-function CombineSink (disposables, sinks, sink, f) {
-  this.sink = sink
-  this.disposables = disposables
-  this.sinks = sinks
-  this.f = f
-
-  var l = sinks.length
-  this.awaiting = l
-  this.values = new Array(l)
-  this.hasValue = new Array(l)
-  for (var i = 0; i < l; ++i) {
-    this.hasValue[i] = false
-  }
-
-  this.activeCount = sinks.length
-}
+class CombineSink extends Pipe {
+  constructor (disposables, sinks, sink, f) {
+    super(sink)
+    this.disposables = disposables
+    this.sinks = sinks
+    this.f = f
+
+    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
+    }
 
-CombineSink.prototype.error = Pipe.prototype.error
+    this.activeCount = sinks.length
+  }
 
-CombineSink.prototype.event = function (t, indexedValue) {
-  var i = indexedValue.index
-  var awaiting = this._updateReady(i)
+  event (t, indexedValue) {
+    const i = indexedValue.index
+    const awaiting = this._updateReady(i)
 
-  this.values[i] = indexedValue.value
-  if (awaiting === 0) {
-    this.sink.event(t, invoke(this.f, this.values))
+    this.values[i] = indexedValue.value
+    if (awaiting === 0) {
+      this.sink.event(t, invoke(this.f, this.values))
+    }
   }
-}
 
-CombineSink.prototype._updateReady = function (index) {
-  if (this.awaiting > 0) {
-    if (!this.hasValue[index]) {
-      this.hasValue[index] = true
-      this.awaiting -= 1
+  _updateReady (index) {
+    if (this.awaiting > 0) {
+      if (!this.hasValue[index]) {
+        this.hasValue[index] = true
+        this.awaiting -= 1
+      }
     }
+    return this.awaiting
   }
-  return this.awaiting
-}
 
-CombineSink.prototype.end = function (t, indexedValue) {
-  dispose.tryDispose(t, this.disposables[indexedValue.index], this.sink)
-  if (--this.activeCount === 0) {
-    this.sink.end(t, indexedValue.value)
+  end (t, indexedValue) {
+    tryDispose(t, this.disposables[indexedValue.index], this.sink)
+    if (--this.activeCount === 0) {
+      this.sink.end(t, indexedValue.value)
+    }
   }
 }
+
diff --git a/packages/core/src/combinator/concatMap.js b/packages/core/src/combinator/concatMap.js
index 2640bea..5ed05cc 100644
--- a/packages/core/src/combinator/concatMap.js
+++ b/packages/core/src/combinator/concatMap.js
@@ -15,6 +15,5 @@ import { mergeMapConcurrently } from './mergeConcurrently'
  * @param {Stream} stream
  * @returns {Stream} new stream containing all events from each stream returned by f
  */
-export function concatMap (f, stream) {
-  return mergeMapConcurrently(f, 1, stream)
-}
+export const concatMap = (f, stream) =>
+  mergeMapConcurrently(f, 1, stream)
diff --git a/packages/core/src/combinator/continueWith.js b/packages/core/src/combinator/continueWith.js
index d3242bf..cb11793 100644
--- a/packages/core/src/combinator/continueWith.js
+++ b/packages/core/src/combinator/continueWith.js
@@ -2,62 +2,63 @@
 /** @author Brian Cavalier */
 /** @author John Hann */
 
-import Stream from '../Stream'
 import Pipe from '../sink/Pipe'
-import * as dispose from '../disposable/dispose'
+import { once, tryDispose } from '../disposable/dispose'
 
-export function continueWith (f, stream) {
-  return new Stream(new ContinueWith(f, stream.source))
-}
-
-function ContinueWith (f, source) {
-  this.f = f
-  this.source = source
-}
+export const continueWith = (f, stream) =>
+  new ContinueWith(f, stream)
 
-ContinueWith.prototype.run = function (sink, scheduler) {
-  return new ContinueWithSink(this.f, this.source, sink, scheduler)
-}
+class ContinueWith {
+  constructor (f, source) {
+    this.f = f
+    this.source = source
+  }
 
-function ContinueWithSink (f, source, sink, scheduler) {
-  this.f = f
-  this.sink = sink
-  this.scheduler = scheduler
-  this.active = true
-  this.disposable = dispose.once(source.run(this, scheduler))
+  run (sink, scheduler) {
+    return new ContinueWithSink(this.f, this.source, sink, scheduler)
+  }
 }
 
-ContinueWithSink.prototype.error = Pipe.prototype.error
+class ContinueWithSink extends Pipe {
+  constructor (f, source, sink, scheduler) {
+    super(sink)
+    this.f = f
+    this.scheduler = scheduler
+    this.active = true
+    this.disposable = once(source.run(this, scheduler))
+  }
 
-ContinueWithSink.prototype.event = function (t, x) {
-  if (!this.active) {
-    return
+  event (t, x) {
+    if (!this.active) {
+      return
+    }
+    this.sink.event(t, x)
   }
-  this.sink.event(t, x)
-}
 
-ContinueWithSink.prototype.end = function (t, x) {
-  if (!this.active) {
-    return
+  end (t, x) {
+    if (!this.active) {
+      return
+    }
+
+    tryDispose(t, this.disposable, this.sink)
+    this._startNext(t, x, this.sink)
   }
 
-  dispose.tryDispose(t, this.disposable, this.sink)
-  this._startNext(t, x, this.sink)
-}
+  _startNext (t, x, sink) {
+    try {
+      this.disposable = this._continue(this.f, x, sink)
+    } catch (e) {
+      sink.error(t, e)
+    }
+  }
 
-ContinueWithSink.prototype._startNext = function (t, x, sink) {
-  try {
-    this.disposable = this._continue(this.f, x, sink)
-  } catch (e) {
-    sink.error(t, e)
+  _continue (f, x, sink) {
+    return f(x).run(sink, this.scheduler)
   }
-}
 
-ContinueWithSink.prototype._continue = function (f, x, sink) {
-  return f(x).source.run(sink, this.scheduler)
+  dispose () {
+    this.active = false
+    return this.disposable.dispose()
+  }
 }
 
-ContinueWithSink.prototype.dispose = function () {
-  this.active = false
-  return this.disposable.dispose()
-}
diff --git a/packages/core/src/combinator/delay.js b/packages/core/src/combinator/delay.js
index 2ceafe1..15a2291 100644
--- a/packages/core/src/combinator/delay.js
+++ b/packages/core/src/combinator/delay.js
@@ -2,9 +2,8 @@
 /** @author Brian Cavalier */
 /** @author John Hann */
 
-import Stream from '../Stream'
 import Pipe from '../sink/Pipe'
-import * as dispose from '../disposable/dispose'
+import { all } from '../disposable/dispose'
 import { propagateEventTask, propagateEndTask } from '../scheduler/PropagateTask'
 
 /**
@@ -12,40 +11,38 @@ import { propagateEventTask, propagateEndTask } from '../scheduler/PropagateTask
  * @param {Stream} stream
  * @returns {Stream} new stream containing the same items, but delayed by ms
  */
-export function delay (delayTime, stream) {
-  return delayTime <= 0 ? stream
-    : new Stream(new Delay(delayTime, stream.source))
+export const delay = (delayTime, stream) =>
+  delayTime <= 0 ? stream : new Delay(delayTime, stream)
+
+class Delay {
+  constructor (dt, source) {
+    this.dt = dt
+    this.source = source
+  }
+
+  run (sink, scheduler) {
+    const delaySink = new DelaySink(this.dt, sink, scheduler)
+    return all([delaySink, this.source.run(delaySink, scheduler)])
+  }
 }
 
-function Delay (dt, source) {
-  this.dt = dt
-  this.source = source
-}
-
-Delay.prototype.run = function (sink, scheduler) {
-  var delaySink = new DelaySink(this.dt, sink, scheduler)
-  return dispose.all([delaySink, this.source.run(delaySink, scheduler)])
-}
+class DelaySink extends Pipe {
+  constructor (dt, sink, scheduler) {
+    super(sink)
+    this.dt = dt
+    this.scheduler = scheduler
+  }
 
-function DelaySink (dt, sink, scheduler) {
-  this.dt = dt
-  this.sink = sink
-  this.scheduler = scheduler
-}
+  dispose () {
+    this.scheduler.cancelAll(task => task.sink === this.sink)
+  }
 
-DelaySink.prototype.dispose = function () {
-  var self = this
-  this.scheduler.cancelAll(function (task) {
-    return task.sink === self.sink
-  })
-}
-
-DelaySink.prototype.event = function (t, x) {
-  this.scheduler.delay(this.dt, propagateEventTask(x, this.sink))
-}
+  event (t, x) {
+    this.scheduler.delay(this.dt, propagateEventTask(x, this.sink))
+  }
 
-DelaySink.prototype.end = function (t, x) {
-  this.scheduler.delay(this.dt, propagateEndTask(x, this.sink))
+  end (t, x) {
+    this.scheduler.delay(this.dt, propagateEndTask(x, this.sink))
+  }
 }
 
-DelaySink.prototype.error = Pipe.prototype.error
diff --git a/packages/core/src/combinator/errors.js b/packages/core/src/combinator/errors.js
index 5860e9a..95995ed 100644
--- a/packages/core/src/combinator/errors.js
+++ b/packages/core/src/combinator/errors.js
@@ -2,10 +2,9 @@
 /** @author Brian Cavalier */
 /** @author John Hann */
 
-import Stream from '../Stream'
 import SafeSink from '../sink/SafeSink'
-import * as dispose from '../disposable/dispose'
-import * as tryEvent from '../source/tryEvent'
+import { tryDispose } from '../disposable/dispose'
+import { tryEvent, tryEnd } from '../source/tryEvent'
 import { propagateErrorTask } from '../scheduler/PropagateTask'
 
 /**
@@ -16,7 +15,7 @@ import { propagateErrorTask } from '../scheduler/PropagateTask'
  * @returns {Stream} new stream which will recover from an error by calling f
  */
 export const recoverWith = (f, stream) =>
-  new Stream(new RecoverWith(f, stream.source))
+  new RecoverWith(f, stream)
 
 /**
  * Create a stream containing only an error
@@ -24,9 +23,9 @@ export const recoverWith = (f, stream) =>
  * @returns {Stream} new stream containing only an error
  */
 export const throwError = e =>
-  new Stream(new ErrorSource(e))
+  new ErrorStream(e)
 
-class ErrorSource {
+class ErrorStream {
   constructor (e) {
     this.value = e
   }
@@ -56,17 +55,17 @@ class RecoverWithSink {
   }
 
   event (t, x) {
-    tryEvent.tryEvent(t, x, this.sink)
+    tryEvent(t, x, this.sink)
   }
 
   end (t, x) {
-    tryEvent.tryEnd(t, x, this.sink)
+    tryEnd(t, x, this.sink)
   }
 
   error (t, e) {
     const nextSink = this.sink.disable()
 
-    dispose.tryDispose(t, this.disposable, this.sink)
+    tryDispose(t, this.disposable, this.sink)
     this._startNext(t, e, nextSink)
   }
 
@@ -80,7 +79,7 @@ class RecoverWithSink {
 
   _continue (f, x, sink) {
     const stream = f(x)
-    return stream.source.run(sink, this.scheduler)
+    return stream.run(sink, this.scheduler)
   }
 
   dispose () {
diff --git a/packages/core/src/combinator/filter.js b/packages/core/src/combinator/filter.js
index d948323..88c6e1b 100644
--- a/packages/core/src/combinator/filter.js
+++ b/packages/core/src/combinator/filter.js
@@ -2,7 +2,6 @@
 /** @author Brian Cavalier */
 /** @author John Hann */
 
-import Stream from '../Stream'
 import Pipe from '../sink/Pipe'
 import Filter from '../fusion/Filter'
 
@@ -12,18 +11,16 @@ import Filter from '../fusion/Filter'
  * @param {Stream} stream stream to filter
  * @returns {Stream} stream containing only items for which predicate returns truthy
  */
-export function filter (p, stream) {
-  return new Stream(Filter.create(p, stream.source))
-}
+export const filter = (p, stream) =>
+  Filter.create(p, stream)
 
 /**
  * Skip repeated events, using === to detect duplicates
  * @param {Stream} stream stream from which to omit repeated events
  * @returns {Stream} stream without repeated events
  */
-export function skipRepeats (stream) {
-  return skipRepeatsWith(same, stream)
-}
+export const skipRepeats = stream =>
+  skipRepeatsWith(same, stream)
 
 /**
  * Skip repeated events using the provided equals function to detect duplicates
@@ -31,37 +28,37 @@ export function skipRepeats (stream) {
  * @param {Stream} stream stream from which to omit repeated events
  * @returns {Stream} stream without repeated events
  */
-export function skipRepeatsWith (equals, stream) {
-  return new Stream(new SkipRepeats(equals, stream.source))
-}
-
-function SkipRepeats (equals, source) {
-  this.equals = equals
-  this.source = source
-}
+export const skipRepeatsWith = (equals, stream) =>
+  new SkipRepeats(equals, stream)
 
-SkipRepeats.prototype.run = function (sink, scheduler) {
-  return this.source.run(new SkipRepeatsSink(this.equals, sink), scheduler)
-}
+class SkipRepeats {
+  constructor (equals, source) {
+    this.equals = equals
+    this.source = source
+  }
 
-function SkipRepeatsSink (equals, sink) {
-  this.equals = equals
-  this.sink = sink
-  this.value = void 0
-  this.init = true
+  run (sink, scheduler) {
+    return this.source.run(new SkipRepeatsSink(this.equals, sink), scheduler)
+  }
 }
 
-SkipRepeatsSink.prototype.end = Pipe.prototype.end
-SkipRepeatsSink.prototype.error = Pipe.prototype.error
+class SkipRepeatsSink extends Pipe {
+  constructor (equals, sink) {
+    super(sink)
+    this.equals = equals
+    this.value = void 0
+    this.init = true
+  }
 
-SkipRepeatsSink.prototype.event = function (t, x) {
-  if (this.init) {
-    this.init = false
-    this.value = x
-    this.sink.event(t, x)
-  } else if (!this.equals(this.value, x)) {
-    this.value = x
-    this.sink.event(t, x)
+  event (t, x) {
+    if (this.init) {
+      this.init = false
+      this.value = x
+      this.sink.event(t, x)
+    } else if (!this.equals(this.value, x)) {
+      this.value = x
+      this.sink.event(t, x)
+    }
   }
 }
 
diff --git a/packages/core/src/combinator/limit.js b/packages/core/src/combinator/limit.js
index f36a81e..4e2ba9c 100644
--- a/packages/core/src/combinator/limit.js
+++ b/packages/core/src/combinator/limit.js
@@ -2,9 +2,8 @@
 /** @author Brian Cavalier */
 /** @author John Hann */
 
-import Stream from '../Stream'
 import Pipe from '../sink/Pipe'
-import * as dispose from '../disposable/dispose'
+import { all } from '../disposable/dispose'
 import { propagateEventTask } from '../scheduler/PropagateTask'
 import Map from '../fusion/Map'
 
@@ -15,18 +14,15 @@ import Map from '../fusion/Map'
  * @returns {Stream}
  */
 export const throttle = (period, stream) =>
-  new Stream(throttleSource(period, stream.source))
+  stream instanceof Map ? commuteMapThrottle(period, stream)
+    : stream instanceof Throttle ? fuseThrottle(period, stream)
+    : new Throttle(period, stream)
 
-const throttleSource = (period, source) =>
-  source instanceof Map ? commuteMapThrottle(period, source)
-    : source instanceof Throttle ? fuseThrottle(period, source)
-    : new Throttle(period, source)
+const commuteMapThrottle = (period, mapStream) =>
+  Map.create(mapStream.f, throttle(period, mapStream.source))
 
-const commuteMapThrottle = (period, source) =>
-  Map.create(source.f, throttleSource(period, source.source))
-
-const fuseThrottle = (period, source) =>
-  new Throttle(Math.max(period, source.period), source.source)
+const fuseThrottle = (period, throttleStream) =>
+  new Throttle(Math.max(period, throttleStream.period), throttleStream.source)
 
 class Throttle {
   constructor (period, source) {
@@ -61,7 +57,7 @@ class ThrottleSink extends Pipe {
  * @returns {Stream} new debounced stream
  */
 export const debounce = (period, stream) =>
-  new Stream(new Debounce(period, stream.source))
+  new Debounce(period, stream)
 
 class Debounce {
   constructor (dt, source) {
@@ -83,7 +79,7 @@ class DebounceSink {
     this.timer = null
 
     const sourceDisposable = source.run(this, scheduler)
-    this.disposable = dispose.all([this, sourceDisposable])
+    this.disposable = all([this, sourceDisposable])
   }
 
   event (t, x) {
diff --git a/packages/core/src/combinator/loop.js b/packages/core/src/combinator/loop.js
index c7b7f3d..b469a0e 100644
--- a/packages/core/src/combinator/loop.js
+++ b/packages/core/src/combinator/loop.js
@@ -2,7 +2,6 @@
 /** @author Brian Cavalier */
 /** @author John Hann */
 
-import Stream from '../Stream'
 import Pipe from '../sink/Pipe'
 
 /**
@@ -16,34 +15,36 @@ import Pipe from '../sink/Pipe'
  * @returns {Stream} new stream whose values are the `value` field of the objects
  * returned by the stepper
  */
-export function loop (stepper, seed, stream) {
-  return new Stream(new Loop(stepper, seed, stream.source))
+export const loop = (stepper, seed, stream) =>
+  new Loop(stepper, seed, stream)
+
+class Loop {
+  constructor (stepper, seed, source) {
+    this.step = stepper
+    this.seed = seed
+    this.source = source
+  }
+
+  run (sink, scheduler) {
+    return this.source.run(new LoopSink(this.step, this.seed, sink), scheduler)
+  }
 }
 
-function Loop (stepper, seed, source) {
-  this.step = stepper
-  this.seed = seed
-  this.source = source
+class LoopSink extends Pipe {
+  constructor (stepper, seed, sink) {
+    super(sink)
+    this.step = stepper
+    this.seed = seed
+  }
+
+  event (t, x) {
+    const result = this.step(this.seed, x)
+    this.seed = result.seed
+    this.sink.event(t, result.value)
+  }
+
+  end (t) {
+    this.sink.end(t, this.seed)
+  }
 }
 
-Loop.prototype.run = function (sink, scheduler) {
-  return this.source.run(new LoopSink(this.step, this.seed, sink), scheduler)
-}
-
-function LoopSink (stepper, seed, sink) {
-  this.step = stepper
-  this.seed = seed
-  this.sink = sink
-}
-
-LoopSink.prototype.error = Pipe.prototype.error
-
-LoopSink.prototype.event = function (t, x) {
-  var result = this.step(this.seed, x)
-  this.seed = result.seed
-  this.sink.event(t, result.value)
-}
-
-LoopSink.prototype.end = function (t) {
-  this.sink.end(t, this.seed)
-}
diff --git a/packages/core/src/combinator/merge.js b/packages/core/src/combinator/merge.js
index 5a20d01..389a66f 100644
--- a/packages/core/src/combinator/merge.js
+++ b/packages/core/src/combinator/merge.js
@@ -2,15 +2,11 @@
 /** @author Brian Cavalier */
 /** @author John Hann */
 
-import Stream from '../Stream'
 import Pipe from '../sink/Pipe'
 import IndexSink from '../sink/IndexSink'
 import { empty } from '../source/core'
-import * as dispose from '../disposable/dispose'
-import * as base from '@most/prelude'
-
-var copy = base.copy
-var reduce = base.reduce
+import { all, tryDispose } from '../disposable/dispose'
+import { copy, reduce } from '@most/prelude'
 
 /**
  * @returns {Stream} stream containing events from all streams in the argument
@@ -27,12 +23,10 @@ export function merge (/* ...streams*/) {
  * in time order.  If two events are simultaneous they will be merged in
  * arbitrary order.
  */
-export function mergeArray (streams) {
-  var l = streams.length
-  return l === 0 ? empty()
-    : l === 1 ? streams[0]
-    : new Stream(mergeSources(streams))
-}
+export const mergeArray = streams =>
+  streams.length === 0 ? empty()
+    : streams.length === 1 ? streams[0]
+    : mergeStreams(streams)
 
 /**
  * This implements fusion/flattening for merge.  It will
@@ -43,51 +37,49 @@ export function mergeArray (streams) {
  * any nested Merge sources, in effect "flattening" nested
  * merge operations into a single merge.
  */
-function mergeSources (streams) {
-  return new Merge(reduce(appendSources, [], streams))
-}
+const mergeStreams = streams =>
+  new Merge(reduce(appendSources, [], streams))
 
-function appendSources (sources, stream) {
-  var source = stream.source
-  return source instanceof Merge
-    ? sources.concat(source.sources)
-    : sources.concat(source)
-}
+const appendSources = (sources, stream) =>
+  sources.concat(stream instanceof Merge ? stream.sources : stream)
 
-function Merge (sources) {
-  this.sources = sources
-}
-
-Merge.prototype.run = function (sink, scheduler) {
-  var l = this.sources.length
-  var disposables = new Array(l)
-  var sinks = new Array(l)
+class Merge {
+  constructor (sources) {
+    this.sources = sources
+  }
 
-  var mergeSink = new MergeSink(disposables, sinks, sink)
+  run (sink, scheduler) {
+    const l = this.sources.length
+    const disposables = new Array(l)
+    const sinks = new Array(l)
 
-  for (var indexSink, i = 0; i < l; ++i) {
-    indexSink = sinks[i] = new IndexSink(i, mergeSink)
-    disposables[i] = this.sources[i].run(indexSink, scheduler)
-  }
+    const mergeSink = new MergeSink(disposables, sinks, sink)
 
-  return dispose.all(disposables)
-}
+    for (let indexSink, i = 0; i < l; ++i) {
+      indexSink = sinks[i] = new IndexSink(i, mergeSink)
+      disposables[i] = this.sources[i].run(indexSink, scheduler)
+    }
 
-function MergeSink (disposables, sinks, sink) {
-  this.sink = sink
-  this.disposables = disposables
-  this.activeCount = sinks.length
+    return all(disposables)
+  }
 }
 
-MergeSink.prototype.error = Pipe.prototype.error
+class MergeSink extends Pipe {
+  constructor (disposables, sinks, sink) {
+    super(sink)
+    this.disposables = disposables
+    this.activeCount = sinks.length
+  }
 
-MergeSink.prototype.event = function (t, indexValue) {
-  this.sink.event(t, indexValue.value)
-}
+  event (t, indexValue) {
+    this.sink.event(t, indexValue.value)
+  }
 
-MergeSink.prototype.end = function (t, indexedValue) {
-  dispose.tryDispose(t, this.disposables[indexedValue.index], this.sink)
-  if (--this.activeCount === 0) {
-    this.sink.end(t, indexedValue.value)
+  end (t, indexedValue) {
+    tryDispose(t, this.disposables[indexedValue.index], this.sink)
+    if (--this.activeCount === 0) {
+      this.sink.end(t, indexedValue.value)
+    }
   }
 }
+
diff --git a/packages/core/src/combinator/mergeConcurrently.js b/packages/core/src/combinator/mergeConcurrently.js
index ee08198..0919672 100644
--- a/packages/core/src/combinator/mergeConcurrently.js
+++ b/packages/core/src/combinator/mergeConcurrently.js
@@ -2,124 +2,127 @@
 /** @author Brian Cavalier */
 /** @author John Hann */
 
-import Stream from '../Stream'
-import * as dispose from '../disposable/dispose'
+import { once, tryDispose } from '../disposable/dispose'
 import LinkedList from '../LinkedList'
 import { id as identity } from '@most/prelude'
 
-export function mergeConcurrently (concurrency, stream) {
-  return mergeMapConcurrently(identity, concurrency, stream)
-}
+export const mergeConcurrently = (concurrency, stream) =>
+  mergeMapConcurrently(identity, concurrency, stream)
 
-export function mergeMapConcurrently (f, concurrency, stream) {
-  return new Stream(new MergeConcurrently(f, concurrency, stream.source))
-}
+export const mergeMapConcurrently = (f, concurrency, stream) =>
+  new MergeConcurrently(f, concurrency, stream)
 
-function MergeConcurrently (f, concurrency, source) {
-  this.f = f
-  this.concurrency = concurrency
-  this.source = source
-}
+class MergeConcurrently {
+  constructor (f, concurrency, source) {
+    this.f = f
+    this.concurrency = concurrency
+    this.source = source
+  }
 
-MergeConcurrently.prototype.run = function (sink, scheduler) {
-  return new Outer(this.f, this.concurrency, this.source, sink, scheduler)
+  run (sink, scheduler) {
+    return new Outer(this.f, this.concurrency, this.source, sink, scheduler)
+  }
 }
 
-function Outer (f, concurrency, source, sink, scheduler) {
-  this.f = f
-  this.concurrency = concurrency
-  this.sink = sink
-  this.scheduler = scheduler
-  this.pending = []
-  this.current = new LinkedList()
-  this.disposable = dispose.once(source.run(this, scheduler))
-  this.active = true
-}
+class Outer {
+  constructor (f, concurrency, source, sink, scheduler) {
+    this.f = f
+    this.concurrency = concurrency
+    this.sink = sink
+    this.scheduler = scheduler
+    this.pending = []
+    this.current = new LinkedList()
+    this.disposable = once(source.run(this, scheduler))
+    this.active = true
+  }
 
-Outer.prototype.event = function (t, x) {
-  this._addInner(t, x)
-}
+  event (t, x) {
+    this._addInner(t, x)
+  }
 
-Outer.prototype._addInner = function (t, x) {
-  if (this.current.length < this.concurrency) {
-    this._startInner(t, x)
-  } else {
-    this.pending.push(x)
+  _addInner (t, x) {
+    if (this.current.length < this.concurrency) {
+      this._startInner(t, x)
+    } else {
+      this.pending.push(x)
+    }
   }
-}
 
-Outer.prototype._startInner = function (t, x) {
-  try {
-    this._initInner(t, x)
-  } catch (e) {
-    this.error(t, e)
+  _startInner (t, x) {
+    try {
+      this._initInner(t, x)
+    } catch (e) {
+      this.error(t, e)
+    }
   }
-}
 
-Outer.prototype._initInner = function (t, x) {
-  var innerSink = new Inner(t, this, this.sink)
-  innerSink.disposable = mapAndRun(this.f, x, innerSink, this.scheduler)
-  this.current.add(innerSink)
-}
+  _initInner (t, x) {
+    const innerSink = new Inner(t, this, this.sink)
+    innerSink.disposable = mapAndRun(this.f, x, innerSink, this.scheduler)
+    this.current.add(innerSink)
+  }
 
-function mapAndRun (f, x, sink, scheduler) {
-  return f(x).source.run(sink, scheduler)
-}
+  end (t, x) {
+    this.active = false
+    tryDispose(t, this.disposable, this.sink)
+    this._checkEnd(t, x)
+  }
 
-Outer.prototype.end = function (t, x) {
-  this.active = false
-  dispose.tryDispose(t, this.disposable, this.sink)
-  this._checkEnd(t, x)
-}
+  error (t, e) {
+    this.active = false
+    this.sink.error(t, e)
+  }
 
-Outer.prototype.error = function (t, e) {
-  this.active = false
-  this.sink.error(t, e)
-}
+  dispose () {
+    this.active = false
+    this.pending.length = 0
+    return Promise.all([this.disposable.dispose(), this.current.dispose()])
+  }
 
-Outer.prototype.dispose = function () {
-  this.active = false
-  this.pending.length = 0
-  return Promise.all([this.disposable.dispose(), this.current.dispose()])
-}
+  _endInner (t, x, inner) {
+    this.current.remove(inner)
+    tryDispose(t, inner, this)
 
-Outer.prototype._endInner = function (t, x, inner) {
-  this.current.remove(inner)
-  dispose.tryDispose(t, inner, this)
+    if (this.pending.length === 0) {
+      this._checkEnd(t, x)
+    } else {
+      this._startInner(t, this.pending.shift())
+    }
+  }
 
-  if (this.pending.length === 0) {
-    this._checkEnd(t, x)
-  } else {
-    this._startInner(t, this.pending.shift())
+  _checkEnd (t, x) {
+    if (!this.active && this.current.isEmpty()) {
+      this.sink.end(t, x)
+    }
   }
 }
 
-Outer.prototype._checkEnd = function (t, x) {
-  if (!this.active && this.current.isEmpty()) {
-    this.sink.end(t, x)
+const mapAndRun = (f, x, sink, scheduler) =>
+  f(x).run(sink, scheduler)
+
+class Inner {
+  constructor (time, outer, sink) {
+    this.prev = this.next = null
+    this.time = time
+    this.outer = outer
+    this.sink = sink
+    this.disposable = void 0
   }
-}
 
-function Inner (time, outer, sink) {
-  this.prev = this.next = null
-  this.time = time
-  this.outer = outer
-  this.sink = sink
-  this.disposable = void 0
-}
+  event (t, x) {
+    this.sink.event(Math.max(t, this.time), x)
+  }
 
-Inner.prototype.event = function (t, x) {
-  this.sink.event(Math.max(t, this.time), x)
-}
+  end (t, x) {
+    this.outer._endInner(Math.max(t, this.time), x, this)
+  }
 
-Inner.prototype.end = function (t, x) {
-  this.outer._endInner(Math.max(t, this.time), x, this)
-}
+  error (t, e) {
+    this.outer.error(Math.max(t, this.time), e)
+  }
 
-Inner.prototype.error = function (t, e) {
-  this.outer.error(Math.max(t, this.time), e)
+  dispose () {
+    return this.disposable.dispose()
+  }
 }
 
-Inner.prototype.dispose = function () {
-  return this.disposable.dispose()
-}
diff --git a/packages/core/src/combinator/observe.js b/packages/core/src/combinator/observe.js
index 29cfb69..e92552d 100644
--- a/packages/core/src/combinator/observe.js
+++ b/packages/core/src/combinator/observe.js
@@ -14,9 +14,8 @@ import { tap } from './transform'
  * @return {Promise} promise that fulfills after the stream ends without
  *  an error, or rejects if the stream ends with an error.
  */
-export function observe (f, stream) {
-  return drain(tap(f, stream))
-}
+export const observe = (f, stream) =>
+  drain(tap(f, stream))
 
 /**
  * "Run" a stream by creating demand and consuming all events
@@ -24,6 +23,5 @@ export function observe (f, stream) {
  * @return {Promise} promise that fulfills after the stream ends without
  *  an error, or rejects if the stream ends with an error.
  */
-export function drain (stream) {
-  return runEffects(stream, defaultScheduler)
-}
+export const drain = stream =>
+  runEffects(stream, defaultScheduler)
diff --git a/packages/core/src/combinator/promises.js b/packages/core/src/combinator/promises.js
index 22bcca2..54cc20d 100644
--- a/packages/core/src/combinator/promises.js
+++ b/packages/core/src/combinator/promises.js
@@ -2,7 +2,6 @@
 /** @author Brian Cavalier */
 /** @author John Hann */
 
-import Stream from '../Stream'
 import fatal from '../fatalError'
 import { just } from '../source/core'
 import { compose } from '@most/prelude'
@@ -11,7 +10,7 @@ import { compose } from '@most/prelude'
  * Turn a Stream<Promise<T>> into Stream<T> by awaiting each promise.
  * Event order is preserved. The stream will fail if any promise rejects.
  */
-export const awaitPromises = stream => new Stream(new Await(stream.source))
+export const awaitPromises = stream => new Await(stream)
 
 /**
  * Create a stream containing only the promise's fulfillment
diff --git a/packages/core/src/combinator/reduce.js b/packages/core/src/combinator/reduce.js
index 5b26198..424e664 100644
--- a/packages/core/src/combinator/reduce.js
+++ b/packages/core/src/combinator/reduce.js
@@ -3,7 +3,7 @@
 /** @author John Hann */
 
 import Pipe from '../sink/Pipe'
-import { runSourceEffects } from '../runEffects'
+import { runEffects } from '../runEffects'
 import defaultScheduler from '../scheduler/defaultScheduler'
 
 /**
@@ -16,7 +16,7 @@ import defaultScheduler from '../scheduler/defaultScheduler'
 * @returns {Promise} promise for the file result of the reduce
 */
 export const reduce = (f, initial, stream) =>
-  runSourceEffects(new Reduce(f, initial, stream.source), defaultScheduler)
+  runEffects(new Reduce(f, initial, stream), defaultScheduler)
 
 class Reduce {
   constructor (f, z, source) {
@@ -32,10 +32,9 @@ class Reduce {
 
 class ReduceSink extends Pipe {
   constructor (f, z, sink) {
-    super()
+    super(sink)
     this.f = f
     this.value = z
-    this.sink = sink
   }
 
   event (t, x) {
diff --git a/packages/core/src/combinator/sample.js b/packages/core/src/combinator/sample.js
index e7803bc..0a82b02 100644
--- a/packages/core/src/combinator/sample.js
+++ b/packages/core/src/combinator/sample.js
@@ -1,16 +1,15 @@
 /** @license MIT License (c) copyright 2010-2016 original author or authors */
 
-import Stream from '../Stream'
 import Pipe from '../sink/Pipe'
-import * as dispose from '../disposable/dispose'
+import { all } from '../disposable/dispose'
 
 export const sample = (f, sampler, stream) =>
-  new Stream(new SampleSource(f, sampler, stream))
+  new Sample(f, sampler, stream)
 
-export class SampleSource {
+export class Sample {
   constructor (f, sampler, stream) {
-    this.source = stream.source
-    this.sampler = sampler.source
+    this.source = stream
+    this.sampler = sampler
     this.f = f
   }
 
@@ -19,14 +18,13 @@ export class SampleSource {
     const sourceDisposable = this.source.run(sampleSink.hold, scheduler)
     const samplerDisposable = this.sampler.run(sampleSink, scheduler)
 
-    return dispose.all([samplerDisposable, sourceDisposable])
+    return all([samplerDisposable, sourceDisposable])
   }
 }
 
 export class SampleSink extends Pipe {
   constructor (f, source, sink) {
     super(sink)
-    this.sink = sink
     this.source = source
     this.f = f
     this.hold = new SampleHold(this)
diff --git a/packages/core/src/combinator/scan.js b/packages/core/src/combinator/scan.js
index 2f53f94..d6552aa 100644
--- a/packages/core/src/combinator/scan.js
+++ b/packages/core/src/combinator/scan.js
@@ -2,9 +2,8 @@
 /** @author Brian Cavalier */
 /** @author John Hann */
 
-import Stream from '../Stream'
 import Pipe from '../sink/Pipe'
-import * as dispose from '../disposable/dispose'
+import { all } from '../disposable/dispose'
 import { propagateEventTask } from '../scheduler/PropagateTask'
 
 /**
@@ -16,7 +15,7 @@ import { propagateEventTask } from '../scheduler/PropagateTask'
  * @returns {Stream} new stream containing successive reduce results
  */
 export const scan = (f, initial, stream) =>
-  new Stream(new Scan(f, initial, stream.source))
+  new Scan(f, initial, stream)
 
 class Scan {
   constructor (f, z, source) {
@@ -28,16 +27,15 @@ class Scan {
   run (sink, scheduler) {
     const d1 = scheduler.asap(propagateEventTask(this.value, sink))
     const d2 = this.source.run(new ScanSink(this.f, this.value, sink), scheduler)
-    return dispose.all([d1, d2])
+    return all([d1, d2])
   }
 }
 
 class ScanSink extends Pipe {
   constructor (f, z, sink) {
-    super()
+    super(sink)
     this.f = f
     this.value = z
-    this.sink = sink
   }
 
   event (t, x) {
diff --git a/packages/core/src/combinator/slice.js b/packages/core/src/combinator/slice.js
index c8a6a00..b6fe07a 100644
--- a/packages/core/src/combinator/slice.js
+++ b/packages/core/src/combinator/slice.js
@@ -2,10 +2,9 @@
 /** @author Brian Cavalier */
 /** @author John Hann */
 
-import Stream from '../Stream'
 import Pipe from '../sink/Pipe'
-import * as core from '../source/core'
-import * as dispose from '../disposable/dispose'
+import { empty } from '../source/core'
+import { once } from '../disposable/dispose'
 import Map from '../fusion/Map'
 
 /**
@@ -13,18 +12,16 @@ import Map from '../fusion/Map'
  * @param {Stream} stream
  * @returns {Stream} new stream containing only up to the first n items from stream
  */
-export function take (n, stream) {
-  return slice(0, n, stream)
-}
+export const take = (n, stream) =>
+  slice(0, n, stream)
 
 /**
  * @param {number} n
  * @param {Stream} stream
  * @returns {Stream} new stream with the first n items removed
  */
-export function skip (n, stream) {
-  return slice(n, Infinity, stream)
-}
+export const skip = (n, stream) =>
+  slice(n, Infinity, stream)
 
 /**
  * Slice a stream by index. Negative start/end indexes are not supported
@@ -33,141 +30,140 @@ export function skip (n, stream) {
  * @param {Stream} stream
  * @returns {Stream} stream containing items where start <= index < end
  */
-export function slice (start, end, stream) {
-  return end <= start ? core.empty()
-    : new Stream(sliceSource(start, end, stream.source))
-}
+export const slice = (start, end, stream) =>
+  end <= start ? empty() : sliceSource(start, end, stream)
 
-function sliceSource (start, end, source) {
-  return source instanceof Map ? commuteMapSlice(start, end, source)
-    : source instanceof Slice ? fuseSlice(start, end, source)
-    : new Slice(start, end, source)
-}
+const sliceSource = (start, end, stream) =>
+  stream instanceof Map ? commuteMapSlice(start, end, stream)
+    : stream instanceof Slice ? fuseSlice(start, end, stream)
+    : new Slice(start, end, stream)
 
-function commuteMapSlice (start, end, source) {
-  return Map.create(source.f, sliceSource(start, end, source.source))
-}
+const commuteMapSlice = (start, end, mapStream) =>
+  Map.create(mapStream.f, sliceSource(start, end, mapStream.source))
 
-function fuseSlice (start, end, source) {
-  start += source.min
-  end = Math.min(end + source.min, source.max)
-  return new Slice(start, end, source.source)
+function fuseSlice (start, end, sliceStream) {
+  const fusedStart = start + sliceStream.min
+  const fusedEnd = Math.min(end + sliceStream.min, sliceStream.max)
+  return new Slice(fusedStart, fusedEnd, sliceStream.source)
 }
 
-function Slice (min, max, source) {
-  this.source = source
-  this.min = min
-  this.max = max
-}
+class Slice {
+  constructor (min, max, source) {
+    this.source = source
+    this.min = min
+    this.max = max
+  }
 
-Slice.prototype.run = function (sink, scheduler) {
-  return new SliceSink(this.min, this.max - this.min, this.source, sink, scheduler)
+  run (sink, scheduler) {
+    return new SliceSink(this.min, this.max - this.min, this.source, sink, scheduler)
+  }
 }
 
-function SliceSink (skip, take, source, sink, scheduler) {
-  this.sink = sink
-  this.skip = skip
-  this.take = take
-  this.disposable = dispose.once(source.run(this, scheduler))
-}
+class SliceSink extends Pipe {
+  constructor (skip, take, source, sink, scheduler) {
+    super(sink)
+    this.skip = skip
+    this.take = take
+    this.disposable = once(source.run(this, scheduler))
+  }
 
-SliceSink.prototype.end = Pipe.prototype.end
-SliceSink.prototype.error = Pipe.prototype.error
+  event (t, x) { // eslint-disable-line complexity
+    if (this.skip > 0) {
+      this.skip -= 1
+      return
+    }
 
-SliceSink.prototype.event = function (t, x) { // eslint-disable-line complexity
-  if (this.skip > 0) {
-    this.skip -= 1
-    return
-  }
+    if (this.take === 0) {
+      return
+    }
 
-  if (this.take === 0) {
-    return
+    this.take -= 1
+    this.sink.event(t, x)
+    if (this.take === 0) {
+      this.dispose()
+      this.sink.end(t, x)
+    }
   }
 
-  this.take -= 1
-  this.sink.event(t, x)
-  if (this.take === 0) {
-    this.dispose()
-    this.sink.end(t, x)
+  dispose () {
+    return this.disposable.dispose()
   }
 }
 
-SliceSink.prototype.dispose = function () {
-  return this.disposable.dispose()
-}
+export const takeWhile = (p, stream) =>
+  new TakeWhile(p, stream)
 
-export function takeWhile (p, stream) {
-  return new Stream(new TakeWhile(p, stream.source))
-}
+class TakeWhile {
+  constructor (p, source) {
+    this.p = p
+    this.source = source
+  }
 
-function TakeWhile (p, source) {
-  this.p = p
-  this.source = source
+  run (sink, scheduler) {
+    return new TakeWhileSink(this.p, this.source, sink, scheduler)
+  }
 }
 
-TakeWhile.prototype.run = function (sink, scheduler) {
-  return new TakeWhileSink(this.p, this.source, sink, scheduler)
-}
+class TakeWhileSink extends Pipe {
+  constructor (p, source, sink, scheduler) {
+    super(sink)
+    this.p = p
+    this.active = true
+    this.disposable = once(source.run(this, scheduler))
+  }
 
-function TakeWhileSink (p, source, sink, scheduler) {
-  this.p = p
-  this.sink = sink
-  this.active = true
-  this.disposable = dispose.once(source.run(this, scheduler))
-}
+  event (t, x) {
+    if (!this.active) {
+      return
+    }
 
-TakeWhileSink.prototype.end = Pipe.prototype.end
-TakeWhileSink.prototype.error = Pipe.prototype.error
+    const p = this.p
+    this.active = p(x)
 
-TakeWhileSink.prototype.event = function (t, x) {
-  if (!this.active) {
-    return
+    if (this.active) {
+      this.sink.event(t, x)
+    } else {
+      this.dispose()
+      this.sink.end(t, x)
+    }
   }
 
-  var p = this.p
-  this.active = p(x)
-  if (this.active) {
-    this.sink.event(t, x)
-  } else {
-    this.dispose()
-    this.sink.end(t, x)
+  dispose () {
+    return this.disposable.dispose()
   }
 }
 
-TakeWhileSink.prototype.dispose = function () {
-  return this.disposable.dispose()
-}
+export const skipWhile = (p, stream) =>
+  new SkipWhile(p, stream)
 
-export function skipWhile (p, stream) {
-  return new Stream(new SkipWhile(p, stream.source))
-}
-
-function SkipWhile (p, source) {
-  this.p = p
-  this.source = source
-}
-
-SkipWhile.prototype.run = function (sink, scheduler) {
-  return this.source.run(new SkipWhileSink(this.p, sink), scheduler)
-}
+class SkipWhile {
+  constructor (p, source) {
+    this.p = p
+    this.source = source
+  }
 
-function SkipWhileSink (p, sink) {
-  this.p = p
-  this.sink = sink
-  this.skipping = true
+  run (sink, scheduler) {
+    return this.source.run(new SkipWhileSink(this.p, sink), scheduler)
+  }
 }
 
-SkipWhileSink.prototype.end = Pipe.prototype.end
-SkipWhileSink.prototype.error = Pipe.prototype.error
+class SkipWhileSink extends Pipe {
+  constructor (p, sink) {
+    super(sink)
+    this.p = p
+    this.skipping = true
+  }
 
-SkipWhileSink.prototype.event = function (t, x) {
-  if (this.skipping) {
-    var p = this.p
-    this.skipping = p(x)
+  event (t, x) {
     if (this.skipping) {
-      return
+      const p = this.p
+      this.skipping = p(x)
+      if (this.skipping) {
+        return
+      }
     }
-  }
 
-  this.sink.event(t, x)
+    this.sink.event(t, x)
+  }
 }
+
diff --git a/packages/core/src/combinator/switch.js b/packages/core/src/combinator/switch.js
index c9a7711..56807bb 100644
--- a/packages/core/src/combinator/switch.js
+++ b/packages/core/src/combinator/switch.js
@@ -2,8 +2,7 @@
 /** @author Brian Cavalier */
 /** @author John Hann */
 
-import Stream from '../Stream'
-import * as dispose from '../disposable/dispose'
+import { all, empty, tryDispose } from '../disposable/dispose'
 
 /**
  * Given a stream of streams, return a new stream that adopts the behavior
@@ -11,7 +10,7 @@ import * as dispose from '../disposable/dispose'
  * @param {Stream} stream of streams on which to switch
  * @returns {Stream} switching stream
  */
-export const switchLatest = stream => new Stream(new Switch(stream.source))
+export const switchLatest = stream => new Switch(stream)
 
 class Switch {
   constructor (source) {
@@ -20,7 +19,7 @@ class Switch {
 
   run (sink, scheduler) {
     const switchSink = new SwitchSink(sink, scheduler)
-    return dispose.all([switchSink, this.source.run(switchSink, scheduler)])
+    return all([switchSink, this.source.run(switchSink, scheduler)])
   }
 }
 
@@ -35,7 +34,7 @@ class SwitchSink {
   event (t, stream) {
     this._disposeCurrent(t) // TODO: capture the result of this dispose
     this.current = new Segment(t, Infinity, this, this.sink)
-    this.current.disposable = stream.source.run(this.current, this.scheduler)
+    this.current.disposable = stream.run(this.current, this.scheduler)
   }
 
   end (t, x) {
@@ -88,7 +87,7 @@ class Segment {
     this.max = max
     this.outer = outer
     this.sink = sink
-    this.disposable = dispose.empty()
+    this.disposable = empty()
   }
 
   event (t, x) {
@@ -107,7 +106,7 @@ class Segment {
 
   _dispose (t) {
     this.max = t
-    dispose.tryDispose(t, this.disposable, this.sink)
+    tryDispose(t, this.disposable, this.sink)
   }
 }
 
diff --git a/packages/core/src/combinator/timeslice.js b/packages/core/src/combinator/timeslice.js
index 2a07462..07700b4 100644
--- a/packages/core/src/combinator/timeslice.js
+++ b/packages/core/src/combinator/timeslice.js
@@ -2,111 +2,112 @@
 /** @author Brian Cavalier */
 /** @author John Hann */
 
-import Stream from '../Stream'
 import Pipe from '../sink/Pipe'
-import * as dispose from '../disposable/dispose'
+import { all } from '../disposable/dispose'
 import { join } from './chain'
 
-export function until (signal, stream) {
-  return new Stream(new Until(signal.source, stream.source))
-}
+export const until = (signal, stream) =>
+  new Until(signal, stream)
 
-export function since (signal, stream) {
-  return new Stream(new Since(signal.source, stream.source))
-}
+export const since = (signal, stream) =>
+  new Since(signal, stream)
 
-export function during (timeWindow, stream) {
-  return until(join(timeWindow), since(timeWindow, stream))
-}
+export const during = (timeWindow, stream) =>
+  until(join(timeWindow), since(timeWindow, stream))
 
-function Until (maxSignal, source) {
-  this.maxSignal = maxSignal
-  this.source = source
-}
+class Until {
+  constructor (maxSignal, source) {
+    this.maxSignal = maxSignal
+    this.source = source
+  }
 
-Until.prototype.run = function (sink, scheduler) {
-  var min = new Bound(-Infinity, sink)
-  var max = new UpperBound(this.maxSignal, sink, scheduler)
-  var disposable = this.source.run(new TimeWindowSink(min, max, sink), scheduler)
+  run (sink, scheduler) {
+    const min = new Bound(-Infinity, sink)
+    const max = new UpperBound(this.maxSignal, sink, scheduler)
+    const disposable = this.source.run(new TimeWindowSink(min, max, sink), scheduler)
 
-  return dispose.all([min, max, disposable])
+    return all([min, max, disposable])
+  }
 }
 
-function Since (minSignal, source) {
-  this.minSignal = minSignal
-  this.source = source
-}
+class Since {
+  constructor (minSignal, source) {
+    this.minSignal = minSignal
+    this.source = source
+  }
 
-Since.prototype.run = function (sink, scheduler) {
-  var min = new LowerBound(this.minSignal, sink, scheduler)
-  var max = new Bound(Infinity, sink)
-  var disposable = this.source.run(new TimeWindowSink(min, max, sink), scheduler)
+  run (sink, scheduler) {
+    const min = new LowerBound(this.minSignal, sink, scheduler)
+    const max = new Bound(Infinity, sink)
+    const disposable = this.source.run(new TimeWindowSink(min, max, sink), scheduler)
 
-  return dispose.all([min, max, disposable])
+    return all([min, max, disposable])
+  }
 }
 
-function Bound (value, sink) {
-  this.value = value
-  this.sink = sink
-}
+class Bound extends Pipe {
+  constructor (value, sink) {
+    super(sink)
+    this.value = value
+  }
 
-Bound.prototype.error = Pipe.prototype.error
-Bound.prototype.event = noop
-Bound.prototype.end = noop
-Bound.prototype.dispose = noop
+  event () {}
+  end () {}
 
-function TimeWindowSink (min, max, sink) {
-  this.min = min
-  this.max = max
-  this.sink = sink
+  dispose () {}
 }
 
-TimeWindowSink.prototype.event = function (t, x) {
-  if (t >= this.min.value && t < this.max.value) {
-    this.sink.event(t, x)
+class TimeWindowSink extends Pipe {
+  constructor (min, max, sink) {
+    super(sink)
+    this.min = min
+    this.max = max
   }
-}
 
-TimeWindowSink.prototype.error = Pipe.prototype.error
-TimeWindowSink.prototype.end = Pipe.prototype.end
-
-function LowerBound (signal, sink, scheduler) {
-  this.value = Infinity
-  this.sink = sink
-  this.disposable = signal.run(this, scheduler)
+  event (t, x) {
+    if (t >= this.min.value && t < this.max.value) {
+      this.sink.event(t, x)
+    }
+  }
 }
 
-LowerBound.prototype.event = function (t /*, x */) {
-  if (t < this.value) {
-    this.value = t
+class LowerBound extends Pipe {
+  constructor (signal, sink, scheduler) {
+    super(sink)
+    this.value = Infinity
+    this.disposable = signal.run(this, scheduler)
   }
-}
 
-LowerBound.prototype.end = noop
-LowerBound.prototype.error = Pipe.prototype.error
+  event (t /*, x */) {
+    if (t < this.value) {
+      this.value = t
+    }
+  }
 
-LowerBound.prototype.dispose = function () {
-  return this.disposable.dispose()
-}
+  end () {}
 
-function UpperBound (signal, sink, scheduler) {
-  this.value = Infinity
-  this.sink = sink
-  this.disposable = signal.run(this, scheduler)
+  dispose () {
+    return this.disposable.dispose()
+  }
 }
 
-UpperBound.prototype.event = function (t, x) {
-  if (t < this.value) {
-    this.value = t
-    this.sink.end(t, x)
+class UpperBound extends Pipe {
+  constructor (signal, sink, scheduler) {
+    super(sink)
+    this.value = Infinity
+    this.disposable = signal.run(this, scheduler)
   }
-}
 
-UpperBound.prototype.end = noop
-UpperBound.prototype.error = Pipe.prototype.error
+  event (t, x) {
+    if (t < this.value) {
+      this.value = t
+      this.sink.end(t, x)
+    }
+  }
 
-UpperBound.prototype.dispose = function () {
-  return this.disposable.dispose()
-}
+  end () {}
 
-function noop () {}
+  dispose () {
+    return this.disposable.dispose()
+  }
+}
diff --git a/packages/core/src/combinator/transform.js b/packages/core/src/combinator/transform.js
index f56aefa..a0d8a06 100644
--- a/packages/core/src/combinator/transform.js
+++ b/packages/core/src/combinator/transform.js
@@ -2,7 +2,6 @@
 /** @author Brian Cavalier */
 /** @author John Hann */
 
-import Stream from '../Stream'
 import Map from '../fusion/Map'
 import Pipe from '../sink/Pipe'
 
@@ -12,9 +11,8 @@ import Pipe from '../sink/Pipe'
  * @param {Stream} stream stream to map
  * @returns {Stream} stream containing items transformed by f
  */
-export function map (f, stream) {
-  return new Stream(Map.create(f, stream.source))
-}
+export const map = (f, stream) =>
+  Map.create(f, stream)
 
 /**
 * Replace each value in the stream with x
@@ -22,11 +20,8 @@ export function map (f, stream) {
 * @param {Stream} stream
 * @returns {Stream} stream containing items replaced with x
 */
-export function constant (x, stream) {
-  return map(function () {
-    return x
-  }, stream)
-}
+export const constant = (x, stream) =>
+  map(() => x, stream)
 
 /**
 * Perform a side effect for each item in the stream
@@ -35,29 +30,30 @@ export function constant (x, stream) {
 * @param {Stream} stream stream to tap
 * @returns {Stream} new stream containing the same items as this stream
 */
-export function tap (f, stream) {
-  return new Stream(new Tap(f, stream.source))
-}
-
-function Tap (f, source) {
-  this.source = source
-  this.f = f
+export const tap = (f, stream) =>
+  new Tap(f, stream)
+
+class Tap {
+  constructor (f, source) {
+    this.source = source
+    this.f = f
+  }
+
+  run (sink, scheduler) {
+    return this.source.run(new TapSink(this.f, sink), scheduler)
+  }
 }
 
-Tap.prototype.run = function (sink, scheduler) {
-  return this.source.run(new TapSink(this.f, sink), scheduler)
+class TapSink extends Pipe {
+  constructor (f, sink) {
+    super(sink)
+    this.f = f
+  }
+
+  event (t, x) {
+    const f = this.f
+    f(x)
+    this.sink.event(t, x)
+  }
 }
 
-function TapSink (f, sink) {
-  this.sink = sink
-  this.f = f
-}
-
-TapSink.prototype.end = Pipe.prototype.end
-TapSink.prototype.error = Pipe.prototype.error
-
-TapSink.prototype.event = function (t, x) {
-  var f = this.f
-  f(x)
-  this.sink.event(t, x)
-}
diff --git a/packages/core/src/combinator/zip.js b/packages/core/src/combinator/zip.js
index dd08b81..748fa85 100644
--- a/packages/core/src/combinator/zip.js
+++ b/packages/core/src/combinator/zip.js
@@ -2,19 +2,15 @@
 /** @author Brian Cavalier */
 /** @author John Hann */
 
-import Stream from '../Stream'
-import * as transform from './transform'
-import * as core from '../source/core'
+import { map } from './transform'
+import { empty } from '../source/core'
 import Pipe from '../sink/Pipe'
 import IndexSink from '../sink/IndexSink'
-import * as dispose from '../disposable/dispose'
-import * as base from '@most/prelude'
+import { all } from '../disposable/dispose'
+import { map as mapArray, tail } from '@most/prelude'
 import invoke from '../invoke'
 import Queue from '../Queue'
 
-var map = base.map
-var tail = base.tail
-
 /**
  * Combine streams pairwise (or tuple-wise) by index by applying f to values
  * at corresponding indices.  The returned stream ends when any of the input
@@ -36,83 +32,77 @@ export function zip (f /*, ...streams */) {
 * @returns {Stream} new stream with items at corresponding indices combined
 *  using f
 */
-export function zipArray (f, streams) {
-  return streams.length === 0 ? core.empty()
-: streams.length === 1 ? transform.map(f, streams[0])
-: new Stream(new Zip(f, map(getSource, streams)))
-}
-
-function getSource (stream) {
-  return stream.source
-}
+export const zipArray = (f, streams) =>
+  streams.length === 0 ? empty()
+    : streams.length === 1 ? map(f, streams[0])
+    : new Zip(f, streams)
+
+class Zip {
+  constructor (f, sources) {
+    this.f = f
+    this.sources = sources
+  }
 
-function Zip (f, sources) {
-  this.f = f
-  this.sources = sources
-}
+  run (sink, scheduler) {
+    const l = this.sources.length
+    const disposables = new Array(l)
+    const sinks = new Array(l)
+    const buffers = new Array(l)
 
-Zip.prototype.run = function (sink, scheduler) {
-  var l = this.sources.length
-  var disposables = new Array(l)
-  var sinks = new Array(l)
-  var buffers = new Array(l)
+    const zipSink = new ZipSink(this.f, buffers, sinks, sink)
 
-  var zipSink = new ZipSink(this.f, buffers, sinks, sink)
+    for (let indexSink, i = 0; i < l; ++i) {
+      buffers[i] = new Queue()
+      indexSink = sinks[i] = new IndexSink(i, zipSink)
+      disposables[i] = this.sources[i].run(indexSink, scheduler)
+    }
 
-  for (var indexSink, i = 0; i < l; ++i) {
-    buffers[i] = new Queue()
-    indexSink = sinks[i] = new IndexSink(i, zipSink)
-    disposables[i] = this.sources[i].run(indexSink, scheduler)
+    return all(disposables)
   }
-
-  return dispose.all(disposables)
 }
 
-function ZipSink (f, buffers, sinks, sink) {
-  this.f = f
-  this.sinks = sinks
-  this.sink = sink
-  this.buffers = buffers
-}
+class ZipSink extends Pipe {
+  constructor (f, buffers, sinks, sink) {
+    super(sink)
+    this.f = f
+    this.sinks = sinks
+    this.buffers = buffers
+  }
 
-ZipSink.prototype.event = function (t, indexedValue) { // eslint-disable-line complexity
-  var buffers = this.buffers
-  var buffer = buffers[indexedValue.index]
+  event (t, indexedValue) { // eslint-disable-line complexity
+    const buffers = this.buffers
+    const buffer = buffers[indexedValue.index]
 
-  buffer.push(indexedValue.value)
+    buffer.push(indexedValue.value)
 
-  if (buffer.length() === 1) {
-    if (!ready(this.buffers)) {
-      return
-    }
+    if (buffer.length() === 1) {
+      if (!ready(this.buffers)) {
+        return
+      }
 
-    emitZipped(this.f, t, buffers, this.sink)
+      emitZipped(this.f, t, buffers, this.sink)
 
-    if (ended(this.buffers, this.sinks)) {
-      this.sink.end(t, void 0)
+      if (ended(this.buffers, this.sinks)) {
+        this.sink.end(t, void 0)
+      }
     }
   }
-}
 
-ZipSink.prototype.end = function (t, indexedValue) {
-  var buffer = this.buffers[indexedValue.index]
-  if (buffer.isEmpty()) {
-    this.sink.end(t, indexedValue.value)
+  end (t, indexedValue) {
+    const buffer = this.buffers[indexedValue.index]
+    if (buffer.isEmpty()) {
+      this.sink.end(t, indexedValue.value)
+    }
   }
 }
 
-ZipSink.prototype.error = Pipe.prototype.error
-
-function emitZipped (f, t, buffers, sink) {
-  sink.event(t, invoke(f, map(head, buffers)))
-}
+const emitZipped = (f, t, buffers, sink) =>
+  sink.event(t, invoke(f, mapArray(head, buffers)))
 
-function head (buffer) {
-  return buffer.shift()
-}
+const head = buffer => buffer.shift()
 
 function ended (buffers, sinks) {
-  for (var i = 0, l = buffers.length; i < l; ++i) {
+  for (let i = 0, l = buffers.length; i < l; ++i) {
     if (buffers[i].isEmpty() && !sinks[i].active) {
       return true
     }
@@ -121,7 +111,7 @@ function ended (buffers, sinks) {
 }
 
 function ready (buffers) {
-  for (var i = 0, l = buffers.length; i < l; ++i) {
+  for (let i = 0, l = buffers.length; i < l; ++i) {
     if (buffers[i].isEmpty()) {
       return false
     }
diff --git a/packages/core/src/fatalError.js b/packages/core/src/fatalError.js
index 31fe596..bd02bc6 100644
--- a/packages/core/src/fatalError.js
+++ b/packages/core/src/fatalError.js
@@ -3,7 +3,9 @@
 /** @author John Hann */
 
 export default function fatalError (e) {
-  setTimeout(function () {
-    throw e
-  }, 0)
+  setTimeout(rethrow, 0, e)
+}
+
+function rethrow (e) {
+  throw e
 }
diff --git a/packages/core/src/fusion/Filter.js b/packages/core/src/fusion/Filter.js
index c97b7d0..5fba29b 100644
--- a/packages/core/src/fusion/Filter.js
+++ b/packages/core/src/fusion/Filter.js
@@ -4,44 +4,41 @@
 
 import Pipe from '../sink/Pipe'
 
-export default function Filter (p, source) {
-  this.p = p
-  this.source = source
-}
-
-/**
- * Create a filtered source, fusing adjacent filter.filter if possible
- * @param {function(x:*):boolean} p filtering predicate
- * @param {{run:function}} source source to filter
- * @returns {Filter} filtered source
- */
-Filter.create = function createFilter (p, source) {
-  if (source instanceof Filter) {
-    return new Filter(and(source.p, p), source.source)
+export default class Filter {
+  constructor (p, source) {
+    this.p = p
+    this.source = source
   }
 
-  return new Filter(p, source)
-}
-
-Filter.prototype.run = function (sink, scheduler) {
-  return this.source.run(new FilterSink(this.p, sink), scheduler)
-}
+  run (sink, scheduler) {
+    return this.source.run(new FilterSink(this.p, sink), scheduler)
+  }
 
-function FilterSink (p, sink) {
-  this.p = p
-  this.sink = sink
+  /**
+   * Create a filtered source, fusing adjacent filter.filter if possible
+   * @param {function(x:*):boolean} p filtering predicate
+   * @param {{run:function}} source source to filter
+   * @returns {Filter} filtered source
+   */
+  static create (p, source) {
+    if (source instanceof Filter) {
+      return new Filter(and(source.p, p), source.source)
+    }
+
+    return new Filter(p, source)
+  }
 }
 
-FilterSink.prototype.end = Pipe.prototype.end
-FilterSink.prototype.error = Pipe.prototype.error
-
-FilterSink.prototype.event = function (t, x) {
-  var p = this.p
-  p(x) && this.sink.event(t, x)
-}
+class FilterSink extends Pipe {
+  constructor (p, sink) {
+    super(sink)
+    this.p = p
+  }
 
-function and (p, q) {
-  return function (x) {
-    return p(x) && q(x)
+  event (t, x) {
+    const p = this.p
+    p(x) && this.sink.event(t, x)
   }
 }
+
+const and = (p, q) => x => p(x) && q(x)
diff --git a/packages/core/src/fusion/FilterMap.js b/packages/core/src/fusion/FilterMap.js
index 0576837..1f331b8 100644
--- a/packages/core/src/fusion/FilterMap.js
+++ b/packages/core/src/fusion/FilterMap.js
@@ -4,27 +4,29 @@
 
 import Pipe from '../sink/Pipe'
 
-export default function FilterMap (p, f, source) {
-  this.p = p
-  this.f = f
-  this.source = source
-}
+export default class FilterMap {
+  constructor (p, f, source) {
+    this.p = p
+    this.f = f
+    this.source = source
+  }
 
-FilterMap.prototype.run = function (sink, scheduler) {
-  return this.source.run(new FilterMapSink(this.p, this.f, sink), scheduler)
+  run (sink, scheduler) {
+    return this.source.run(new FilterMapSink(this.p, this.f, sink), scheduler)
+  }
 }
 
-function FilterMapSink (p, f, sink) {
-  this.p = p
-  this.f = f
-  this.sink = sink
-}
+class FilterMapSink extends Pipe {
+  constructor (p, f, sink) {
+    super(sink)
+    this.p = p
+    this.f = f
+  }
 
-FilterMapSink.prototype.event = function (t, x) {
-  var f = this.f
-  var p = this.p
-  p(x) && this.sink.event(t, f(x))
+  event (t, x) {
+    const f = this.f
+    const p = this.p
+    p(x) && this.sink.event(t, f(x))
+  }
 }
 
-FilterMapSink.prototype.end = Pipe.prototype.end
-FilterMapSink.prototype.error = Pipe.prototype.error
diff --git a/packages/core/src/fusion/Map.js b/packages/core/src/fusion/Map.js
index af03df1..157628a 100644
--- a/packages/core/src/fusion/Map.js
+++ b/packages/core/src/fusion/Map.js
@@ -5,45 +5,47 @@
 import Pipe from '../sink/Pipe'
 import Filter from './Filter'
 import FilterMap from './FilterMap'
-import * as base from '@most/prelude'
+import { compose } from '@most/prelude'
 
-export default function Map (f, source) {
-  this.f = f
-  this.source = source
-}
-
-/**
- * Create a mapped source, fusing adjacent map.map, filter.map,
- * and filter.map.map if possible
- * @param {function(*):*} f mapping function
- * @param {{run:function}} source source to map
- * @returns {Map|FilterMap} mapped source, possibly fused
- */
-Map.create = function createMap (f, source) {
-  if (source instanceof Map) {
-    return new Map(base.compose(f, source.f), source.source)
+export default class Map {
+  constructor (f, source) {
+    this.f = f
+    this.source = source
   }
 
-  if (source instanceof Filter) {
-    return new FilterMap(source.p, f, source.source)
+  run (sink, scheduler) { // eslint-disable-line no-extend-native
+    return this.source.run(new MapSink(this.f, sink), scheduler)
   }
 
-  return new Map(f, source)
+  /**
+   * Create a mapped source, fusing adjacent map.map, filter.map,
+   * and filter.map.map if possible
+   * @param {function(*):*} f mapping function
+   * @param {{run:function}} source source to map
+   * @returns {Map|FilterMap} mapped source, possibly fused
+   */
+  static create (f, source) {
+    if (source instanceof Map) {
+      return new Map(compose(f, source.f), source.source)
+    }
+
+    if (source instanceof Filter) {
+      return new FilterMap(source.p, f, source.source)
+    }
+
+    return new Map(f, source)
+  }
 }
 
-Map.prototype.run = function (sink, scheduler) { // eslint-disable-line no-extend-native
-  return this.source.run(new MapSink(this.f, sink), scheduler)
-}
+class MapSink extends Pipe {
+  constructor (f, sink) {
+    super(sink)
+    this.f = f
+  }
 
-function MapSink (f, sink) {
-  this.f = f
-  this.sink = sink
+  event (t, x) {
+    const f = this.f
+    this.sink.event(t, f(x))
+  }
 }
 
-MapSink.prototype.end = Pipe.prototype.end
-MapSink.prototype.error = Pipe.prototype.error
-
-MapSink.prototype.event = function (t, x) {
-  var f = this.f
-  this.sink.event(t, f(x))
-}
diff --git a/packages/core/src/index.js b/packages/core/src/index.js
index bbe6a93..e7d3dca 100644
--- a/packages/core/src/index.js
+++ b/packages/core/src/index.js
@@ -1,8 +1,6 @@
 /** @license MIT License (c) copyright 2016 original author or authors */
 import { curry2, curry3 } from '@most/prelude'
 
-export { default as Stream } from './Stream'
-
 export { just, empty, never } from './source/core'
 
 export { periodic } from './source/periodic'
diff --git a/packages/core/src/index.js.flow b/packages/core/src/index.js.flow
index 63b8d52..3e5dd3e 100644
--- a/packages/core/src/index.js.flow
+++ b/packages/core/src/index.js.flow
@@ -68,12 +68,8 @@ export type Sink<A> = {
   error: (t: Time, e: Error) => void,
 }
 
-export type Source<A> = {
-  run: (sink: Sink<A>, scheduler: Scheduler) => Disposable
-}
-
 export type Stream<A> = {
-  source: Source<A>
+  run: (sink: Sink<A>, scheduler: Scheduler) => Disposable
 }
 
 export type SeedValue <S, A> = {
diff --git a/packages/core/src/runEffects.js b/packages/core/src/runEffects.js
index a315f14..66a1308 100644
--- a/packages/core/src/runEffects.js
+++ b/packages/core/src/runEffects.js
@@ -4,17 +4,15 @@
 
 import { settable } from './disposable/dispose'
 
-export const runEffects = ({ source }, scheduler) => runSourceEffects(source, scheduler)
-
-export const runSourceEffects = (source, scheduler) =>
+export const runEffects = (stream, scheduler) =>
   new Promise((resolve, reject) =>
-    runSource(source, scheduler, resolve, reject))
+    runStream(stream, scheduler, resolve, reject))
 
-function runSource (source, scheduler, resolve, reject) {
+function runStream (stream, scheduler, resolve, reject) {
   const disposable = settable()
   const observer = new RunEffectsSink(resolve, reject, disposable)
 
-  disposable.setDisposable(source.run(observer, scheduler))
+  disposable.setDisposable(stream.run(observer, scheduler))
 }
 
 class RunEffectsSink {
diff --git a/packages/core/src/sink/Pipe.js b/packages/core/src/sink/Pipe.js
index 3a945e7..e7064ae 100644
--- a/packages/core/src/sink/Pipe.js
+++ b/packages/core/src/sink/Pipe.js
@@ -2,24 +2,27 @@
 /** @author Brian Cavalier */
 /** @author John Hann */
 
-/**
- * A sink mixin that simply forwards event, end, and error to
- * another sink.
- * @param sink
- * @constructor
- */
-export default function Pipe (sink) {
-  this.sink = sink
-}
+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
+  }
 
-Pipe.prototype.event = function (t, x) {
-  return this.sink.event(t, x)
-}
+  event (t, x) {
+    return this.sink.event(t, x)
+  }
 
-Pipe.prototype.end = function (t, x) {
-  return this.sink.end(t, x)
-}
+  end (t, x) {
+    return this.sink.end(t, x)
+  }
 
-Pipe.prototype.error = function (t, e) {
-  return this.sink.error(t, e)
+  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 057106c..f2cf8f7 100644
--- a/packages/core/src/source/core.js
+++ b/packages/core/src/source/core.js
@@ -2,7 +2,6 @@
 /** @author Brian Cavalier */
 /** @author John Hann */
 
-import Stream from '../Stream'
 import * as dispose from '../disposable/dispose'
 import { propagateTask, propagateEndTask } from '../scheduler/PropagateTask'
 
@@ -11,7 +10,7 @@ import { propagateTask, propagateEndTask } from '../scheduler/PropagateTask'
  * @param {*} x
  * @returns {Stream}
  */
-export const just = x => new Stream(new Just(x))
+export const just = x => new Just(x)
 
 class Just {
   constructor (x) {
@@ -34,7 +33,7 @@ function runJust (t, x, sink) {
  */
 export const empty = () => EMPTY
 
-class EmptySource {
+class Empty {
   run (sink, scheduler) {
     const task = propagateEndTask(undefined, sink)
     scheduler.asap(task)
@@ -45,7 +44,7 @@ class EmptySource {
 
 const disposeEmpty = task => task.dispose()
 
-const EMPTY = new Stream(new EmptySource())
+const EMPTY = new Empty()
 
 /**
  * Stream containing no events and never ends
@@ -53,10 +52,10 @@ const EMPTY = new Stream(new EmptySource())
  */
 export const never = () => NEVER
 
-class NeverSource {
+class Never {
   run () {
     return dispose.empty()
   }
 }
 
-const NEVER = new Stream(new NeverSource())
+const NEVER = new Never()
diff --git a/packages/core/src/source/from.js b/packages/core/src/source/from.js
deleted file mode 100644
index c1cd5cb..0000000
--- a/packages/core/src/source/from.js
+++ /dev/null
@@ -1,27 +0,0 @@
-/** @license MIT License (c) copyright 2010-2016 original author or authors */
-/** @author Brian Cavalier */
-/** @author John Hann */
-
-import Stream from '../Stream'
-import { fromArray } from './fromArray'
-import { isIterable } from '../iterable'
-import { fromIterable } from './fromIterable'
-import { isArrayLike } from '@most/prelude'
-
-function coerce (a) { // eslint-disable-line complexity
-  if (a instanceof Stream) {
-    return a
-  }
-
-  if (Array.isArray(a) || isArrayLike(a)) {
-    return fromArray(a)
-  }
-
-  if (isIterable(a)) {
-    return fromIterable(a)
-  }
-
-  throw new TypeError('from(x) must be observable, iterable, or array-like: ' + a)
-}
-
-export { coerce as from }
diff --git a/packages/core/src/source/fromArray.js b/packages/core/src/source/fromArray.js
index 51a49e6..b87870b 100644
--- a/packages/core/src/source/fromArray.js
+++ b/packages/core/src/source/fromArray.js
@@ -2,13 +2,12 @@
 /** @author Brian Cavalier */
 /** @author John Hann */
 
-import Stream from '../Stream'
 import { propagateTask } from '../scheduler/PropagateTask'
 
 export const fromArray = a =>
-  new Stream(new ArraySource(a))
+  new ArrayStream(a)
 
-class ArraySource {
+class ArrayStream {
   constructor (a) {
     this.array = a
   }
diff --git a/packages/core/src/source/fromIterable.js b/packages/core/src/source/fromIterable.js
index 0c9b21a..202cf01 100644
--- a/packages/core/src/source/fromIterable.js
+++ b/packages/core/src/source/fromIterable.js
@@ -1,13 +1,12 @@
 /** @license MIT License (c) copyright 2010-2016 original author or authors */
 
-import Stream from '../Stream'
 import { getIterator } from '../iterable'
 import { propagateTask } from '../scheduler/PropagateTask'
 
 export const fromIterable = iterable =>
-  new Stream(new IterableSource(iterable))
+  new IterableStream(iterable)
 
-class IterableSource {
+class IterableStream {
   constructor (iterable) {
     this.iterable = iterable
   }
diff --git a/packages/core/src/source/generate.js b/packages/core/src/source/generate.js
index 6b94fac..5456962 100644
--- a/packages/core/src/source/generate.js
+++ b/packages/core/src/source/generate.js
@@ -2,8 +2,7 @@
 /** @author Brian Cavalier */
 /** @author John Hann */
 
-import Stream from '../Stream'
-import * as base from '@most/prelude'
+import { tail } from '@most/prelude'
 
 /**
  * Compute a stream using an *async* generator, which yields promises
@@ -12,35 +11,39 @@ import * as base from '@most/prelude'
  * @returns {Stream}
  */
 export function generate (f /*, ...args */) {
-  return new Stream(new GenerateSource(f, base.tail(arguments)))
+  return new GeneratorStream(f, tail(arguments))
 }
 
-function GenerateSource (f, args) {
-  this.f = f
-  this.args = args
-}
+class GeneratorStream {
+  constructor (f, args) {
+    this.f = f
+    this.args = args
+  }
 
-GenerateSource.prototype.run = function (sink, scheduler) {
-  return new Generate(this.f.apply(void 0, this.args), sink, scheduler)
+  run (sink, scheduler) {
+    return new Generate(this.f.apply(void 0, this.args), sink, scheduler)
+  }
 }
 
-function Generate (iterator, sink, scheduler) {
-  this.iterator = iterator
-  this.sink = sink
-  this.scheduler = scheduler
-  this.active = true
+class Generate {
+  constructor (iterator, sink, scheduler) {
+    this.iterator = iterator
+    this.sink = sink
+    this.scheduler = scheduler
+    this.active = true
+
+    const err = e => this.sink.error(this.scheduler.now(), e)
 
-  var self = this
-  function err (e) {
-    self.sink.error(self.scheduler.now(), e)
+    Promise.resolve(this).then(next).catch(err)
   }
 
-  Promise.resolve(this).then(next).catch(err)
+  dispose () {
+    this.active = false
+  }
 }
 
-function next (generate, x) {
-  return generate.active ? handle(generate, generate.iterator.next(x)) : x
-}
+const next = (generate, x) =>
+  generate.active ? handle(generate, generate.iterator.next(x)) : x
 
 function handle (generate, result) {
   if (result.done) {
@@ -63,6 +66,3 @@ function error (generate, e) {
   return handle(generate, generate.iterator.throw(e))
 }
 
-Generate.prototype.dispose = function () {
-  this.active = false
-}
diff --git a/packages/core/src/source/iterate.js b/packages/core/src/source/iterate.js
index 58bd7d9..36b6be0 100644
--- a/packages/core/src/source/iterate.js
+++ b/packages/core/src/source/iterate.js
@@ -2,8 +2,6 @@
 /** @author Brian Cavalier */
 /** @author John Hann */
 
-import Stream from '../Stream'
-
 /**
  * Compute a stream by iteratively calling f to produce values
  * Event times may be controlled by returning a Promise from f
@@ -11,41 +9,39 @@ import Stream from '../Stream'
  * @param {*} x initial value
  * @returns {Stream}
  */
-export function iterate (f, x) {
-  return new Stream(new IterateSource(f, x))
-}
+export const iterate = (f, x) =>
+  new IterateStream(f, x)
 
-function IterateSource (f, x) {
-  this.f = f
-  this.value = x
-}
+class IterateStream {
+  constructor (f, x) {
+    this.f = f
+    this.value = x
+  }
 
-IterateSource.prototype.run = function (sink, scheduler) {
-  return new Iterate(this.f, this.value, sink, scheduler)
+  run (sink, scheduler) {
+    return new Iterate(this.f, this.value, sink, scheduler)
+  }
 }
 
-function Iterate (f, initial, sink, scheduler) {
-  this.f = f
-  this.sink = sink
-  this.scheduler = scheduler
-  this.active = true
+class Iterate {
+  constructor (f, initial, sink, scheduler) {
+    this.f = f
+    this.sink = sink
+    this.scheduler = scheduler
+    this.active = true
 
-  var x = initial
+    const err = e =>
+      this.sink.error(this.scheduler.now(), e)
 
-  var self = this
-  function err (e) {
-    self.sink.error(self.scheduler.now(), e)
-  }
+    const start = iterate =>
+      stepIterate(iterate, initial)
 
-  function start (iterate) {
-    return stepIterate(iterate, x)
+    Promise.resolve(this).then(start).catch(err)
   }
 
-  Promise.resolve(this).then(start).catch(err)
-}
-
-Iterate.prototype.dispose = function () {
-  this.active = false
+  dispose () {
+    this.active = false
+  }
 }
 
 function stepIterate (iterate, x) {
@@ -55,12 +51,10 @@ function stepIterate (iterate, x) {
     return x
   }
 
-  var f = iterate.f
-  return Promise.resolve(f(x)).then(function (y) {
-    return continueIterate(iterate, y)
-  })
+  const f = iterate.f
+  return Promise.resolve(f(x))
+    .then(y => continueIterate(iterate, y))
 }
 
-function continueIterate (iterate, x) {
-  return !iterate.active ? iterate.value : stepIterate(iterate, x)
-}
+const continueIterate = (iterate, x) =>
+  iterate.active ? stepIterate(iterate, x) : iterate.value
diff --git a/packages/core/src/source/periodic.js b/packages/core/src/source/periodic.js
index d3d9939..cc49017 100644
--- a/packages/core/src/source/periodic.js
+++ b/packages/core/src/source/periodic.js
@@ -2,7 +2,6 @@
 /** @author Brian Cavalier */
 /** @author John Hann */
 
-import Stream from '../Stream'
 import { propagateEventTask } from '../scheduler/PropagateTask'
 
 /**
@@ -10,9 +9,8 @@ import { propagateEventTask } from '../scheduler/PropagateTask'
  * @param {Number} period periodicity of events in millis
  * @returns {Stream} new stream of periodic events, the event value is undefined
  */
-export function periodic (period) {
-  return new Stream(new Periodic(period))
-}
+export const periodic = period =>
+  new Periodic(period)
 
 class Periodic {
   constructor (period) {
diff --git a/packages/core/src/source/unfold.js b/packages/core/src/source/unfold.js
index 3dc29c8..9843549 100644
--- a/packages/core/src/source/unfold.js
+++ b/packages/core/src/source/unfold.js
@@ -2,8 +2,6 @@
 /** @author Brian Cavalier */
 /** @author John Hann */
 
-import Stream from '../Stream'
-
 /**
  * Compute a stream by unfolding tuples of future values from a seed value
  * Event times may be controlled by returning a Promise from f
@@ -14,46 +12,45 @@ import Stream from '../Stream'
  * @returns {Stream} stream containing all value of all tuples produced by the
  *  unfolding function.
  */
-export function unfold (f, seed) {
-  return new Stream(new UnfoldSource(f, seed))
-}
+export const unfold = (f, seed) =>
+  new UnfoldStream(f, seed)
 
-function UnfoldSource (f, seed) {
-  this.f = f
-  this.value = seed
-}
+class UnfoldStream {
+  constructor (f, seed) {
+    this.f = f
+    this.value = seed
+  }
 
-UnfoldSource.prototype.run = function (sink, scheduler) {
-  return new Unfold(this.f, this.value, sink, scheduler)
+  run (sink, scheduler) {
+    return new Unfold(this.f, this.value, sink, scheduler)
+  }
 }
 
-function Unfold (f, x, sink, scheduler) {
-  this.f = f
-  this.sink = sink
-  this.scheduler = scheduler
-  this.active = true
+class Unfold {
+  constructor (f, x, sink, scheduler) {
+    this.f = f
+    this.sink = sink
+    this.scheduler = scheduler
+    this.active = true
 
-  var self = this
-  function err (e) {
-    self.sink.error(self.scheduler.now(), e)
-  }
+    const err = e =>
+      this.sink.error(this.scheduler.now(), e)
 
-  function start (unfold) {
-    return stepUnfold(unfold, x)
-  }
+    const start = unfold =>
+      stepUnfold(unfold, x)
 
-  Promise.resolve(this).then(start).catch(err)
-}
+    Promise.resolve(this).then(start).catch(err)
+  }
 
-Unfold.prototype.dispose = function () {
-  this.active = false
+  dispose () {
+    this.active = false
+  }
 }
 
 function stepUnfold (unfold, x) {
-  var f = unfold.f
-  return Promise.resolve(f(x)).then(function (tuple) {
-    return continueUnfold(unfold, tuple)
-  })
+  const f = unfold.f
+  return Promise.resolve(f(x))
+    .then(tuple => continueUnfold(unfold, tuple))
 }
 
 function continueUnfold (unfold, tuple) {
@@ -64,8 +61,5 @@ function continueUnfold (unfold, tuple) {
 
   unfold.sink.event(unfold.scheduler.now(), tuple.value)
 
-  if (!unfold.active) {
-    return tuple.value
-  }
-  return stepUnfold(unfold, tuple.seed)
+  return unfold.active ? stepUnfold(unfold, tuple.seed) : tuple.value
 }
diff --git a/packages/core/type-definitions/types.d.ts b/packages/core/type-definitions/types.d.ts
index 5472540..147aa31 100644
--- a/packages/core/type-definitions/types.d.ts
+++ b/packages/core/type-definitions/types.d.ts
@@ -1,13 +1,4 @@
 export interface Stream<A> {
-  source: Source<A>;
-}
-
-export class Stream<A> {
-  public source: Source<A>;
-  constructor (source: Source<A>);
-}
-
-export interface Source<A> {
   run (sink: Sink<A>, scheduler: Scheduler): Disposable;
 }
 


Test patch:
diff --git a/packages/core/test/Stream-test.js b/packages/core/test/Stream-test.js
deleted file mode 100644
index 55980ea..0000000
--- a/packages/core/test/Stream-test.js
+++ /dev/null
@@ -1,12 +0,0 @@
-import { describe, it } from 'mocha'
-import { is } from '@briancavalier/assert'
-
-import { default as Stream } from '../src/Stream'
-
-describe('Stream', function () {
-  it('should have expected source', function () {
-    const source = { run: () => ({ dispose: () => {} }) }
-    const s = new Stream(source)
-    is(source, s.source)
-  })
-})
diff --git a/packages/core/test/combinator/chain-test.js b/packages/core/test/combinator/chain-test.js
index f6d1c61..063096d 100644
--- a/packages/core/test/combinator/chain-test.js
+++ b/packages/core/test/combinator/chain-test.js
@@ -9,11 +9,10 @@ import { take } from '../../src/combinator/slice'
 import { drain } from '../../src/combinator/observe'
 import { just, never } from '../../src/source/core'
 import { fromArray } from '../../src/source/fromArray'
-import Stream from '../../src/Stream'
 
 import { assertSame } from '../helper/stream-helper'
 import { collectEventsFor } from '../helper/testEnv'
-import FakeDisposeSource from './../helper/FakeDisposeSource'
+import FakeDisposeStream from '../helper/FakeDisposeStream'
 
 const sentinel = { value: 'sentinel' }
 
@@ -67,14 +66,14 @@ describe('join', function () {
     const inner = just(sentinel)
     const outer = just(inner)
 
-    const s = join(new Stream(new FakeDisposeSource(dispose, outer.source)))
+    const s = join(new FakeDisposeStream(dispose, outer))
 
     return drain(s).then(() => assert(dispose.calledOnce))
   })
 
   it('should dispose inner stream', function () {
     const dispose = spy()
-    const inner = new Stream(new FakeDisposeSource(dispose, just(sentinel).source))
+    const inner = new FakeDisposeStream(dispose, just(sentinel))
 
     const s = join(just(inner))
 
@@ -90,7 +89,7 @@ describe('join', function () {
     const values = [1, 2, 3]
     const spies = values.map(() => spy())
 
-    const inners = values.map((x, i) => new Stream(new FakeDisposeSource(spies[i], just(x).source)))
+    const inners = values.map((x, i) => new FakeDisposeStream(spies[i], just(x)))
 
     const s = join(fromArray(inners))
 
diff --git a/packages/core/test/combinator/concatMap-test.js b/packages/core/test/combinator/concatMap-test.js
index d3e0024..d0b13d8 100644
--- a/packages/core/test/combinator/concatMap-test.js
+++ b/packages/core/test/combinator/concatMap-test.js
@@ -10,10 +10,9 @@ import { take } from '../../src/combinator/slice'
 import { drain } from '../../src/combinator/observe'
 import { just, never } from '../../src/source/core'
 import { fromArray } from '../../src/source/fromArray'
-import { default as Stream } from '../../src/Stream'
 
 import { ticks, atTimes, collectEventsFor, collectEvents } from '../helper/testEnv'
-import FakeDisposeSource from '../helper/FakeDisposeSource'
+import FakeDisposeSource from '../helper/FakeDisposeStream'
 
 const sentinel = { value: 'sentinel' }
 
@@ -70,14 +69,14 @@ describe('concatMap', function () {
     const inner = just(sentinel)
     const outer = just(inner)
 
-    const s = concatMap.concatMap(identity, new Stream(new FakeDisposeSource(dispose, outer.source)))
+    const s = concatMap.concatMap(identity, new FakeDisposeSource(dispose, outer))
 
     return drain(s).then(() => assert(dispose.called))
   })
 
   it('should dispose inner stream', function () {
     const dispose = spy()
-    const inner = new Stream(new FakeDisposeSource(dispose, just(sentinel).source))
+    const inner = new FakeDisposeSource(dispose, just(sentinel))
 
     const s = concatMap.concatMap(identity, just(inner))
 
@@ -94,7 +93,7 @@ describe('concatMap', function () {
     const values = [1, 2, 3]
     const spies = values.map(() => spy())
 
-    const inners = values.map((x, i) => new Stream(new FakeDisposeSource(spies[i], just(x).source)))
+    const inners = values.map((x, i) => new FakeDisposeSource(spies[i], just(x)))
 
     const s = concatMap.concatMap(identity, fromArray(inners))
 
diff --git a/packages/core/test/combinator/loop-test.js b/packages/core/test/combinator/loop-test.js
index 3e66100..295f1a4 100644
--- a/packages/core/test/combinator/loop-test.js
+++ b/packages/core/test/combinator/loop-test.js
@@ -2,12 +2,11 @@ import { describe, it } from 'mocha'
 import { is, eq, assert } from '@briancavalier/assert'
 import { spy } from 'sinon'
 
-import Stream from '../../src/Stream'
 import { loop } from '../../src/combinator/loop'
 import { throwError } from '../../src/combinator/errors'
 import { just } from '../../src/source/core'
 
-import FakeDisposeSource from '../helper/FakeDisposeSource'
+import FakeDisposeSource from '../helper/FakeDisposeStream'
 
 import { makeEventsFromArray, collectEventsFor } from '../helper/testEnv'
 
@@ -44,7 +43,7 @@ describe('loop', function () {
   it('should dispose', function () {
     const dispose = spy()
 
-    const stream = new Stream(new FakeDisposeSource(dispose, just(sentinel).source))
+    const stream = new FakeDisposeSource(dispose, just(sentinel))
     const s = loop(toPair, 0, stream)
 
     return collectEventsFor(1, s).then(() => assert(dispose.calledOnce))
diff --git a/packages/core/test/combinator/scan-test.js b/packages/core/test/combinator/scan-test.js
index 32590e4..857a50b 100644
--- a/packages/core/test/combinator/scan-test.js
+++ b/packages/core/test/combinator/scan-test.js
@@ -2,33 +2,15 @@ import { describe, it } from 'mocha'
 import { assert, eq } from '@briancavalier/assert'
 import { spy } from 'sinon'
 
-import Stream from '../../src/Stream'
 import { scan } from '../../src/combinator/scan'
 
 import { runEffects } from '../../src/runEffects'
 import { atTime, makeEventsFromArray, collectEventsFor, ticks } from '../helper/testEnv'
-import FakeDisposeSource from '../helper/FakeDisposeSource'
+import FakeDisposeStream from '../helper/FakeDisposeStream'
+import { endWith } from '../helper/endWith'
 
 const sentinel = { value: 'sentinel' }
 
-function endWith (endValue, { source }) {
-  return new Stream({
-    run: (sink, scheduler) => {
-      return source.run({
-        end (t, _) {
-          sink.end(t, endValue)
-        },
-        event (t, x) {
-          sink.event(t, x)
-        },
-        error (t, e) {
-          sink.error(t, e)
-        }
-      }, scheduler)
-    }
-  })
-}
-
 describe('scan', function () {
   it('should yield incremental accumulated values', function () {
     let a = ['a', 'b', 'c', 'd']
@@ -56,7 +38,7 @@ describe('scan', function () {
   it('should dispose', function () {
     const dispose = spy()
 
-    const stream = new Stream(new FakeDisposeSource(dispose, atTime(0, sentinel).source))
+    const stream = new FakeDisposeStream(dispose, atTime(0, sentinel))
     const s = scan((z, x) => x, 0, stream)
 
     return runEffects(s, ticks(1))
diff --git a/packages/core/test/helper/FakeDisposeSource.js b/packages/core/test/helper/FakeDisposeSource.js
deleted file mode 100644
index c934678..0000000
--- a/packages/core/test/helper/FakeDisposeSource.js
+++ /dev/null
@@ -1,16 +0,0 @@
-import * as dispose from '../../src/disposable/dispose'
-
-export default FakeDisposeSource
-
-FakeDisposeSource.from = function (disposer, stream) {
-  return new FakeDisposeSource(disposer, stream.source)
-}
-
-function FakeDisposeSource (disposer, source) {
-  this.source = source
-  this.disposable = dispose.create(disposer)
-}
-
-FakeDisposeSource.prototype.run = function (sink, scheduler) {
-  return dispose.all([this.source.run(sink, scheduler), this.disposable])
-}
diff --git a/packages/core/test/helper/FakeDisposeStream.js b/packages/core/test/helper/FakeDisposeStream.js
new file mode 100644
index 0000000..e166e95
--- /dev/null
+++ b/packages/core/test/helper/FakeDisposeStream.js
@@ -0,0 +1,16 @@
+import * as dispose from '../../src/disposable/dispose'
+
+export default FakeDisposeStream
+
+FakeDisposeStream.from = function (disposer, stream) {
+  return new FakeDisposeStream(disposer, stream)
+}
+
+function FakeDisposeStream (disposer, source) {
+  this.source = source
+  this.disposable = dispose.create(disposer)
+}
+
+FakeDisposeStream.prototype.run = function (sink, scheduler) {
+  return dispose.all([this.source.run(sink, scheduler), this.disposable])
+}
diff --git a/packages/core/test/helper/endWith.js b/packages/core/test/helper/endWith.js
index 7745680..d4a732e 100644
--- a/packages/core/test/helper/endWith.js
+++ b/packages/core/test/helper/endWith.js
@@ -1,9 +1,9 @@
 import Pipe from '../../src/sink/Pipe'
 
 export const endWith = (endValue, stream) =>
-  new stream.constructor({
+  ({
     run: (sink, scheduler) =>
-      stream.source.run(new EndWithSink(endValue, sink), scheduler)
+      stream.run(new EndWithSink(endValue, sink), scheduler)
   })
 
 class EndWithSink extends Pipe {
diff --git a/packages/core/test/helper/testEnv.js b/packages/core/test/helper/testEnv.js
index 0652d5b..32b3bda 100644
--- a/packages/core/test/helper/testEnv.js
+++ b/packages/core/test/helper/testEnv.js
@@ -2,7 +2,6 @@
 /** @author Brian Cavalier */
 /** @author John Hann */
 
-import Stream from '../../src/Stream'
 import { propagateEventTask, propagateEndTask } from '../../src/scheduler/PropagateTask'
 import Scheduler from '../../src/scheduler/Scheduler'
 import Timeline from '../../src/scheduler/Timeline'
@@ -39,7 +38,7 @@ export const makeEvents = (dt, n) =>
 
 export const atTime = (time, value) => atTimes([{ time, value }])
 
-export const atTimes = array => new Stream(new AtTimes(array))
+export const atTimes = array => new AtTimes(array)
 
 class AtTimes {
   constructor (array) {
diff --git a/packages/core/test/limit-test.js b/packages/core/test/limit-test.js
index d182241..4c04a51 100644
--- a/packages/core/test/limit-test.js
+++ b/packages/core/test/limit-test.js
@@ -84,16 +84,16 @@ describe('throttle', function () {
     it('should use max', function () {
       const s1 = throttle(2, throttle(1, atTimes([])))
       const s2 = throttle(1, throttle(2, atTimes([])))
-      eq(2, eq(s1.source.period, s2.source.period))
+      eq(2, eq(s1.period, s2.period))
     })
 
     it('should commute map', function () {
       const id = x => x
       const s = throttle(1, map(id, just()))
 
-      assert(s.source instanceof Map)
-      is(id, s.source.f)
-      eq(1, s.source.source.period)
+      assert(s instanceof Map)
+      is(id, s.f)
+      eq(1, s.source.period)
     })
   })
 
diff --git a/packages/core/test/slice-test.js b/packages/core/test/slice-test.js
index c14d690..f042ed8 100644
--- a/packages/core/test/slice-test.js
+++ b/packages/core/test/slice-test.js
@@ -11,22 +11,22 @@ describe('slice', function () {
   describe('fusion', function () {
     it('should narrow when second slice is smaller', function () {
       const s = slice(1, 5, slice(1, 10, makeEvents(1, 1)))
-      eq(2, s.source.min)
-      eq(6, s.source.max)
+      eq(2, s.min)
+      eq(6, s.max)
     })
 
     it('should narrow when second slice is larger', function () {
       const s = slice(1, 10, slice(1, 5, makeEvents(1, 1)))
-      eq(2, s.source.min)
-      eq(5, s.source.max)
+      eq(2, s.min)
+      eq(5, s.max)
     })
 
     it('should commute map', function () {
       const id = x => x
       const s = slice(0, 3, map(id, makeEventsFromArray(1, [0, 1, 2, 3])))
 
-      assert(s.source instanceof Map)
-      is(id, s.source.f)
+      assert(s instanceof Map)
+      is(id, s.f)
       return collectEventsFor(3, s)
         .then(eq([
           { time: 0, value: 0 },
diff --git a/packages/core/test/timeslice-test.js b/packages/core/test/timeslice-test.js
index e213a56..8976bf0 100644
--- a/packages/core/test/timeslice-test.js
+++ b/packages/core/test/timeslice-test.js
@@ -6,11 +6,10 @@ import { take } from '../src/combinator/slice'
 import { periodic } from '../src/source/periodic'
 import { just, never } from '../src/source/core'
 import { delay } from '../src/combinator/delay'
-import { default as Stream } from '../src/Stream'
 
 import { runEffects } from '../src/runEffects'
 import { ticks, collectEvents, collectEventsFor, makeEvents } from './helper/testEnv'
-import FakeDisposeSource from './helper/FakeDisposeSource'
+import FakeDisposeStream from './helper/FakeDisposeStream'
 import { endWith } from './helper/endWith'
 
 import { spy } from 'sinon'
@@ -35,7 +34,7 @@ describe('during', function () {
 
   it('should dispose source stream', function () {
     const dispose = spy()
-    const stream = new Stream(FakeDisposeSource.from(dispose, periodic(1)))
+    const stream = FakeDisposeStream.from(dispose, periodic(1))
     const timespan = delay(1, just(delay(5, just())))
 
     const s = during(timespan, stream)
@@ -48,7 +47,7 @@ describe('during', function () {
 
     const stream = periodic(1)
     const timespan = delay(1, just(delay(5, just())))
-    const dt = new Stream(FakeDisposeSource.from(dispose, timespan))
+    const dt = FakeDisposeStream.from(dispose, timespan)
 
     const s = during(dt, stream)
     return collectEvents(s, ticks(6))
@@ -76,7 +75,7 @@ describe('until', function () {
 
   it('should dispose source stream', function () {
     const dispose = spy()
-    const stream = new Stream(FakeDisposeSource.from(dispose, periodic(1)))
+    const stream = FakeDisposeStream.from(dispose, periodic(1))
     const signal = delay(3, just())
 
     const s = until(signal, stream)
@@ -86,7 +85,7 @@ describe('until', function () {
 
   it('should end immediately on signal', function () {
     const dispose = spy()
-    const stream = new Stream(FakeDisposeSource.from(dispose, never()))
+    const stream = FakeDisposeStream.from(dispose, never())
     const signal = just()
 
     const s = until(signal, stream)
@@ -100,7 +99,7 @@ describe('until', function () {
   it('should dispose signal', function () {
     const dispose = spy()
     const stream = periodic(1)
-    const signal = new Stream(FakeDisposeSource.from(dispose, delay(3, just())))
+    const signal = FakeDisposeStream.from(dispose, delay(3, just()))
 
     const s = until(signal, stream)
     return collectEventsFor(5, s)
@@ -141,7 +140,7 @@ describe('since', function () {
   it('should dispose signal', function () {
     const dispose = spy()
     const stream = take(5, periodic(1))
-    const signal = new Stream(FakeDisposeSource.from(dispose, delay(3, just())))
+    const signal = FakeDisposeStream.from(dispose, delay(3, just()))
 
     const s = since(signal, stream)
     return collectEventsFor(5, s)