Repository: vigetlabs/microcosm
Pull request #512 (2018-04-25 20:52:32)

[microcosm] Async observables
*Review notes:* It's probably best to view this in split-view

### What is it

This commit updates the Observable abstraction in Microcosm such that it is asynchronous.

Updates to observables are queued with a universal scheduler. Until this scheduler "flushes" an observable, updates are buffered within an internal list of updates.

### Why 

The scheduler dispatches updates as fast as the browser will allow it to, using requestIdleCallback. When requestIdleCallback is not provided, it uses setTimeout.

This means that Microcosm's scheduler does not block the browser. Work might be slow, but the browser never suffers.

nhunzaker: Thanks, @zporter !

Patch:
diff --git a/packages/microcosm/src/index.js b/packages/microcosm/src/index.js
index c170328f..7f12c668 100644
--- a/packages/microcosm/src/index.js
+++ b/packages/microcosm/src/index.js
@@ -15,3 +15,5 @@ export { tag } from './tag'
 export { Domain } from './domain'
 export { Effect } from './effect'
 export { Agent } from './agent'
+
+export { scheduler } from './scheduler'
diff --git a/packages/microcosm/src/observable.js b/packages/microcosm/src/observable.js
index bc3d66af..c9fe9ffa 100644
--- a/packages/microcosm/src/observable.js
+++ b/packages/microcosm/src/observable.js
@@ -5,149 +5,122 @@
 
 import assert from 'assert'
 import { observable } from './symbols'
-import { noop, EMPTY_OBJECT } from './empty'
-
-interface Unsubscribable {
+import { noop } from './empty'
+import { scheduler } from './scheduler'
+import { isObject } from './type-checks'
+
+// A subscription has no work in progress
+const IDLE = 0
+// A subscription will send, but the scheduler is not ready for it.
+const BUFFERING = 1
+// A subscription is processing
+const WORKING = 2
+// A subscription is done
+const CLOSED = 3
+
+const NEXT = 'next'
+const ERROR = 'error'
+const COMPLETE = 'complete'
+const CANCEL = 'cancel'
+
+export interface Unsubscribable {
   unsubscribe(): any;
 }
 
-type Cleanup = Unsubscribable | (() => any)
-type Subscriber = (observer: *) => ?Cleanup
-
-export class Observer {
-  _config: Object
-
-  constructor(next: *, error?: *, complete?: *) {
-    if (typeof next === 'function') {
-      this._config = { next, error, complete }
+export type Cleanup = Unsubscribable | (() => any)
+export type Subscriber = (observer: *) => ?Cleanup
+export type Queue = { type: string, value: * }[]
+
+class Observer {
+  next: any => void
+  error: any => void
+  complete: () => void
+  cancel: () => void
+
+  constructor(next: *, error?: *, complete?: *, cancel?: *) {
+    if (isObject(next)) {
+      this.next = next.next || noop
+      this.error = next.error || noop
+      this.complete = next.complete || noop
+      this.cancel = next.cancel || noop
     } else {
-      assert(next, 'Unable to subscribe to ' + String(next))
-      this._config = next || EMPTY_OBJECT
+      this.next = next || noop
+      this.error = error || noop
+      this.complete = complete || noop
+      this.cancel = cancel || noop
     }
   }
-
-  get next(): (payload?: *) => any {
-    return this._config.next || noop
-  }
-
-  get error(): (error?: *) => any {
-    return this._config.error || noop
-  }
-
-  get complete(): () => any {
-    return this._config.complete || noop
-  }
-
-  get cancel(): (reason?: *) => any {
-    return this._config.cancel || noop
-  }
 }
 
-const ClosedObserver = new Observer({})
-
 export class SubscriptionObserver {
   _subscription: Subscription
 
   constructor(subscription: Subscription) {
     this._subscription = subscription
   }
-
-  get next(): (value?: *) => void {
-    return handleNext.bind(null, this._subscription)
+  get next() {
+    return onNotify.bind(null, this._subscription, NEXT)
   }
-
-  get complete(): () => void {
-    return handleComplete.bind(null, this._subscription)
+  get error() {
+    return onNotify.bind(null, this._subscription, ERROR)
   }
-
-  get error(): (error?: *) => void {
-    return handleError.bind(null, this._subscription)
+  get complete() {
+    return onNotify.bind(null, this._subscription, COMPLETE)
   }
-
-  get cancel(): (reason?: *) => void {
-    return handleCancel.bind(null, this._subscription)
+  get cancel() {
+    return onNotify.bind(null, this._subscription, CANCEL)
+  }
+  get unsubscribe() {
+    return this._subscription.unsubscribe
   }
 }
 
 export class Subscription implements Unsubscribable {
-  _cleanup: ?() => void
+  _cleanup: Cleanup
   _observer: Observer
-  _origin: *
+  _state: number
+  _queue: Queue
 
-  constructor(observer: Observer, subscriber: Subscriber, origin: any) {
-    this._cleanup = undefined
+  constructor(observer: Observer, subscriber: Subscriber, origin: *) {
+    this._state = IDLE
+    this._cleanup = noop
+    this._queue = []
     this._observer = observer
-    this._origin = origin
 
-    var subscriptionObserver = new SubscriptionObserver(this)
+    let subscriptionObserver = new SubscriptionObserver(this)
 
     try {
       // Call the subscriber function
-      let cleanup = subscriber.call(undefined, subscriptionObserver)
-
-      // The return value must be undefined, null, a subscription object, or a function
-      if (cleanup != null) {
-        if (typeof cleanup.unsubscribe === 'function') {
-          cleanup = cleanup.unsubscribe
-        } else if (typeof cleanup !== 'function') {
-          throw new TypeError(String(cleanup) + ' is not a function')
-        }
-
-        this._cleanup = cleanup
-      }
+      this._cleanup = subscriber.call(origin, subscriptionObserver) || noop
     } catch (error) {
-      // If an error occurs during startup, then attempt to send the error
-      // to the observer
       subscriptionObserver.error(error)
-    } finally {
-      if (this.closed) {
-        cleanupSubscription(this)
-      }
     }
-  }
 
-  close(): void {
-    this._observer = ClosedObserver
+    if (this._state === IDLE) {
+      this._state = WORKING
+    }
   }
 
   get closed(): boolean {
-    return this._observer == ClosedObserver
+    return this._state === CLOSED
   }
 
   get unsubscribe(): * {
-    return handleUnsubscribe.bind(null, this)
+    return disposeSubscription.bind(null, this)
   }
 }
 
 export class Observable {
-  _subscriber: Function
-  _subscriptions: Set<Subscription>
+  _subscriber: SubscriptionObserver => ?Cleanup
 
-  constructor(subscriber: Subscriber) {
-    this._subscriber = subscriber
-    this._subscriptions = new Set()
+  constructor(subscriber?: Subscriber) {
+    this._subscriber = subscriber || noop
   }
 
-  subscribe(
-    next: Object | Function,
-    error?: Function,
-    complete?: Function
-  ): Subscription {
-    let subscription = new Subscription(
-      new Observer(next, error, complete),
-      this._subscriber,
-      this
-    )
-
-    this._subscriptions.add(subscription)
+  subscribe(next: *, error?: *, complete?: *, cancel?: *): Subscription {
+    let observer = new Observer(next, error, complete, cancel)
 
-    return subscription
-  }
-
-  get cancel(): (reason?: *) => void {
-    return reason => {
-      this._subscriptions.forEach(s => handleCancel(s, reason))
-    }
+    return new Subscription(observer, this._subscriber, this)
   }
 
   map(fn: (*) => *, scope: any): Observable {
@@ -156,18 +129,23 @@ export class Observable {
     }
 
     return new Observable(observer => {
-      return this.subscribe({
-        next(value) {
-          try {
-            value = fn.call(scope, value)
-          } catch (e) {
-            return observer.error(e)
-          }
-          observer.next(value)
-        },
-        error: observer.error,
-        complete: observer.complete
-      })
+      return this.subscribe(
+        value => observer.next(fn.call(scope, value)),
+        observer.error,
+        observer.complete,
+        observer.cancel
+      )
+    })
+  }
+
+  every(fn: (subject: this) => void, scope: any): * {
+    let dispatch = fn.bind(scope, this)
+
+    return this.subscribe({
+      next: dispatch,
+      error: dispatch,
+      complete: dispatch,
+      cancel: dispatch
     })
   }
 
@@ -181,105 +159,136 @@ export class Observable {
       for (var i = 0; i < arguments.length; ++i) {
         observer.next(arguments[i])
       }
-
-      observer.complete()
-    })
-  }
-
-  static wrap(source: *): Observable {
-    if (source && typeof source === 'object') {
-      let builder = getObservable(source)
-
-      if (builder) {
-        return builder.call(source)
-      }
-
-      if (typeof source.then === 'function') {
-        return Observable.fromPromise(source)
-      }
-    }
-
-    return new Observable(observer => {
-      observer.next(source)
       observer.complete()
     })
   }
+}
 
-  static fromPromise(promise: Promise<*>): Observable {
-    return new Observable(observer => {
-      promise
-        .then(observer.next)
-        .then(observer.complete)
-        .catch(observer.error)
-    })
-  }
+function disposeSubscription(subscription: Subscription) {
+  closeSubscription(subscription)
+  cleanupSubscription(subscription)
 }
 
-export function getObservable(obj: any) {
-  return obj && obj[observable]
+const ClosedObserver = new Observer()
+
+function closeSubscription(subscription: Subscription): void {
+  subscription._observer = ClosedObserver
+  subscription._queue.length = 0
+  subscription._state = CLOSED
 }
 
-function cleanupSubscription(subscription) {
+function cleanupSubscription(subscription: Subscription): void {
   let cleanup = subscription._cleanup
 
-  subscription._origin._subscriptions.delete(subscription)
+  // Drop the reference to the cleanup function so that we won't call it
+  // more than once
+  subscription._cleanup = noop
 
-  if (cleanup) {
-    // Drop the reference to the cleanup function so that we won't call it
-    // more than once
-    subscription._cleanup = undefined
-
-    // Call the cleanup function
+  if (typeof cleanup === 'function') {
     cleanup()
+  } else if (cleanup instanceof Subscription) {
+    cleanup.unsubscribe()
   }
 }
 
-function handleNext(subscription: Subscription, value?: *) {
-  if (subscription._observer) {
-    subscription._observer.next(value)
-  }
-}
+function flushSubscription(subscription: Subscription): void {
+  let queue = subscription._queue
 
-function handleComplete(subscription: Subscription): void {
-  let observer = subscription._observer
+  subscription._queue = []
+  subscription._state = WORKING
 
-  subscription.close()
+  for (let i = 0; i < queue.length; i++) {
+    if (subscription.closed) {
+      break
+    }
 
-  observer.complete()
+    notifySubscription(subscription, queue[i].type, queue[i].value)
+  }
 
-  cleanupSubscription(subscription)
+  subscription._state = IDLE
 }
 
-function handleError(subscription: Subscription, error?: *): void {
-  if (subscription.closed) {
-    // TODO: this doesn't feel right. If a user double errors() we
-    // should warn, but not raise. It should mean that they did
-    // something wrong
-    throw error
-  }
-
+function notifySubscription(subscription: *, type: *, value: *) {
   let observer = subscription._observer
 
-  subscription.close()
-
-  observer.error(error)
-
-  cleanupSubscription(subscription)
-}
+  // This is what triggers subscription callbacks
+  try {
+    switch (type) {
+      case NEXT:
+        observer.next(value)
+        break
+      case ERROR:
+        closeSubscription(subscription)
+        observer.error(value)
+        break
+      case COMPLETE:
+        closeSubscription(subscription)
+        observer.complete()
+        break
+      case CANCEL:
+        closeSubscription(subscription)
+        observer.cancel()
+        break
+      default:
+        assert(
+          false,
+          `Unrecognized type ${type}. This is an error internal to Microcosm. ` +
+            `Please file an issue: https://github.com/vigetlabs/microcosm/issues`
+        )
+    }
+  } catch (e) {
+    scheduler().raise(e)
+  }
 
-function handleUnsubscribe(subscription: Subscription): void {
   if (subscription.closed) {
-    return
+    cleanupSubscription(subscription)
   }
-
-  subscription.close()
-  cleanupSubscription(subscription)
 }
 
-function handleCancel(subscription: Subscription, reason: ?*): void {
-  if (subscription._observer) {
-    subscription._observer.cancel(reason)
+function onNotify(subscription: Subscription, type: string, value: *) {
+  // On cancel, empty out the queue for subscriptions
+  if (type === CANCEL) {
+    subscription._queue.length = 0
   }
 
-  handleUnsubscribe(subscription)
+  switch (subscription._state) {
+    case CLOSED:
+      // If closed, do nothing
+      break
+    case BUFFERING:
+      // Until the scheduler is ready to send out an update, push
+      // notifications into a backlog
+      subscription._queue.push({ type, value })
+      break
+    case WORKING:
+      // Go ahead and send messages that result from subscription
+      // callbacks while the subscription is processing
+      notifySubscription(subscription, type, value)
+      break
+    case IDLE:
+      // For any other phase, we are now buffering. Initiate a new
+      // queue and push a job to flush it eventually.
+      assert(
+        subscription._queue.length <= 0,
+        'This subscription has outstanding work, however it is trying ' +
+          'to begin buffering. This is an error internal to Microcosm. ' +
+          `Please file an issue: https://github.com/vigetlabs/microcosm/issues`
+      )
+
+      subscription._state = BUFFERING
+      subscription._queue.push({ type, value })
+      scheduler().push(flushSubscription.bind(undefined, subscription))
+      break
+    default:
+      // There are two cases for which we can get into this state:
+      //   1. A subscription is closed
+      //   2. We've made a mistake and assigned an invalid state.
+      // Both are exceptions. Throw in strict mode
+      assert(
+        false,
+        `Observable subscription is in an invalid state.` +
+          'This is an error internal to Microcosm. ' +
+          `Please file an issue: https://github.com/vigetlabs/microcosm/issues`
+      )
+  }
 }
diff --git a/packages/microcosm/src/scheduler.js b/packages/microcosm/src/scheduler.js
new file mode 100644
index 00000000..032d8abf
--- /dev/null
+++ b/packages/microcosm/src/scheduler.js
@@ -0,0 +1,106 @@
+/**
+ * @fileoverview The central scheduler for Microcosm work
+ * This is based upon Scheduler from the ChooJS project
+ * https://github.com/choojs/nanoscheduler/blob/master/index.js
+ */
+
+import assert from 'assert'
+
+const hasWindow = typeof window !== 'undefined'
+const root = hasWindow ? window : global
+
+export function scheduler() {
+  // TODO: This should probably be based upon the repo. Some sort of
+  // root context
+  if (!root.__microcosmScheduler) {
+    root.__microcosmScheduler = new Scheduler(hasWindow)
+  }
+
+  return root.__microcosmScheduler
+}
+
+const untilEmpty = {
+  timeRemaining: () => 1
+}
+
+function idleFallback(callback) {
+  setTimeout(callback, 0, untilEmpty)
+}
+
+class Scheduler {
+  constructor(hasWindow) {
+    let hasIdle = hasWindow && 'requestIdleCallback' in root
+
+    this.method = hasIdle ? root.requestIdleCallback.bind(root) : idleFallback
+    this.tick = this.tick.bind(this)
+    this.scheduled = false
+    this.queue = []
+    this.errorCallbacks = []
+  }
+
+  push(cb) {
+    assert.equal(
+      typeof cb,
+      'function',
+      'scheduler.push: callback should be a function'
+    )
+
+    this.queue.push(cb)
+    this.schedule()
+  }
+
+  schedule() {
+    if (this.scheduled) {
+      return
+    }
+
+    this.scheduled = true
+    this.method(this.tick)
+  }
+
+  tick(idleDeadline) {
+    try {
+      while (this.queue.length && idleDeadline.timeRemaining() > 0) {
+        this.queue.shift()()
+      }
+    } catch (error) {
+      this.raise(error)
+    }
+
+    this.scheduled = false
+
+    if (this.queue.length) {
+      this.schedule()
+    }
+  }
+
+  flush() {
+    this.tick(untilEmpty)
+  }
+
+  onError(fn) {
+    this.errorCallbacks.push(fn)
+  }
+
+  raise(error) {
+    if (this.errorCallbacks.length) {
+      this.errorCallbacks.forEach(callback => callback(error))
+      this.errorCallbacks.length = 0
+    } else {
+      setTimeout(() => {
+        throw error
+      })
+    }
+  }
+
+  then(pass, fail) {
+    return new Promise((resolve, reject) => {
+      this.push(resolve)
+      this.onError(reject)
+    }).then(pass, fail)
+  }
+
+  get size() {
+    return this.queue.length
+  }
+}
diff --git a/packages/microcosm/src/type-checks.js b/packages/microcosm/src/type-checks.js
new file mode 100644
index 00000000..6e4ffa41
--- /dev/null
+++ b/packages/microcosm/src/type-checks.js
@@ -0,0 +1,21 @@
+/**
+ * @flow
+ */
+
+import { observable } from './symbols'
+
+export function isPromise(obj: any) {
+  return obj && typeof obj.then === 'function'
+}
+
+export function isObservable(obj: any) {
+  return obj && obj[observable]
+}
+
+export function isObject(obj: any) {
+  return obj && typeof obj === 'object'
+}
+
+export function isPlainObject(obj: any) {
+  return isObject(obj) && Object.getPrototypeOf(obj) === Object.prototype
+}


Test patch:
diff --git a/packages/microcosm/test/setup.js b/packages/microcosm/test/setup.js
index bf18fc22..d784cdc2 100644
--- a/packages/microcosm/test/setup.js
+++ b/packages/microcosm/test/setup.js
@@ -1,6 +1,10 @@
-import { get } from 'microcosm'
+import { get, scheduler } from 'microcosm'
 import expect from 'expect'
 
+scheduler().onError(error => {
+  throw error instanceof Error ? error : new Error(error)
+})
+
 // Make a strict-only test flag
 it.dev = function(description, test) {
   if (!process.env.BUNDLED) {
diff --git a/packages/microcosm/test/unit/observable.test.js b/packages/microcosm/test/unit/observable.test.js
index b3b22c16..a3f12efb 100644
--- a/packages/microcosm/test/unit/observable.test.js
+++ b/packages/microcosm/test/unit/observable.test.js
@@ -1,4 +1,4 @@
-import { Observable } from 'microcosm'
+import { Observable, scheduler } from 'microcosm'
 
 describe('Observable', function() {
   describe('subscriber', function() {
@@ -14,10 +14,28 @@ describe('Observable', function() {
           expect(reason).toEqual('Terrible things')
         }
       })
+
+      scheduler().flush()
+    })
+
+    it('cleans up any enqueued tasks on unsubscribe', () => {
+      expect.assertions(1)
+
+      let observable = new Observable(observer => {
+        throw 'Terrible things'
+      })
+
+      observable.subscribe({
+        error(reason) {
+          expect(reason).toEqual('Terrible things')
+        }
+      })
+
+      scheduler().flush()
     })
   })
 
-  describe('duplicate closers', function() {
+  describe('duplicate completion', function() {
     it('can not double complete', function() {
       let complete = jest.fn()
 
@@ -28,6 +46,8 @@ describe('Observable', function() {
 
       observable.subscribe({ complete })
 
+      scheduler().flush()
+
       expect(complete).toHaveBeenCalledTimes(1)
     })
 
@@ -42,118 +62,74 @@ describe('Observable', function() {
 
       observable.subscribe({ next, complete })
 
+      scheduler().flush()
+
       expect(next).toHaveBeenCalledTimes(0)
       expect(complete).toHaveBeenCalledTimes(1)
     })
 
-    it('can not double cancel', function() {
-      let cancel = jest.fn()
+    it('can not complete after erroring', function() {
+      let complete = jest.fn()
 
       let observable = new Observable(observer => {
-        observer.cancel()
-        observer.cancel()
+        observer.error()
+        observer.complete()
       })
 
-      observable.subscribe({ cancel })
-
-      expect(cancel).toHaveBeenCalledTimes(1)
-    })
-
-    it('raises on the second error call', function() {
-      expect.assertions(3)
-
-      let error = jest.fn()
-
-      let observable = new Observable(observer => {
-        observer.error('one')
-        observer.error('two')
-      })
+      observable.subscribe({ complete })
 
-      try {
-        observable.subscribe({ error })
-      } catch (x) {
-        expect(x).toBe('two')
-      }
+      scheduler().flush()
 
-      expect(error).toHaveBeenCalledTimes(1)
-      expect(error).toHaveBeenCalledWith('one')
+      expect(complete).toHaveBeenCalledTimes(0)
     })
+  })
 
-    it('raises if erroring after completion', function() {
-      expect.assertions(2)
-
-      let error = jest.fn()
+  describe('subscribe', function() {
+    it('subscribes to next as the first argument', () => {
+      let next = jest.fn()
 
       let observable = new Observable(observer => {
+        observer.next(2)
         observer.complete()
-        observer.error('Bad news')
       })
 
-      try {
-        observable.subscribe({ error })
-      } catch (x) {
-        expect(x).toBe('Bad news')
-      }
+      observable.subscribe(next)
 
-      expect(error).toHaveBeenCalledTimes(0)
+      scheduler().flush()
+
+      expect(next).toHaveBeenCalledWith(2)
     })
 
-    it('can not complete after erroring', function() {
+    it('next can be null', () => {
       let complete = jest.fn()
 
       let observable = new Observable(observer => {
-        observer.error()
+        observer.next(2)
         observer.complete()
       })
 
-      observable.subscribe({ complete })
-
-      expect(complete).toHaveBeenCalledTimes(0)
-    })
-
-    it('can not cancel after erroring', function() {
-      let cancel = jest.fn()
-
-      let observable = new Observable(observer => {
-        observer.error()
-        observer.cancel()
-      })
+      observable.subscribe(null, null, complete)
 
-      observable.subscribe({ cancel })
+      scheduler().flush()
 
-      expect(cancel).toHaveBeenCalledTimes(0)
+      expect(complete).toHaveBeenCalledTimes(1)
     })
-  })
 
-  describe('subscribe', function() {
-    it.dev('raises if not given an observer', () => {
-      expect.assertions(1)
+    it('subscribes to error as the second argument', () => {
+      let error = jest.fn()
 
       let observable = new Observable(observer => {
-        throw 'Terrible things'
+        observer.error('Exceptional!')
       })
 
-      try {
-        observable.subscribe(null)
-      } catch (x) {
-        expect(x.message).toBe('Unable to subscribe to null')
-      }
-    })
+      observable.subscribe(() => {}, error)
 
-    it('subscribes to next as the first argument', () => {
-      let next = jest.fn()
+      scheduler().flush()
 
-      let observable = new Observable(observer => {
-        observer.next(2)
-        observer.complete()
-      })
-
-      observable.subscribe(next)
-
-      expect(next).toHaveBeenCalledWith(2)
+      expect(error).toHaveBeenCalledWith('Exceptional!')
     })
 
-    it('subscribes to error as the second argument', () => {
+    it('triggers error() when the subscription observer throws', () => {
       let error = jest.fn()
 
       let observable = new Observable(observer => {
@@ -162,6 +138,8 @@ describe('Observable', function() {
 
       observable.subscribe(() => {}, error)
 
+      scheduler().flush()
+
       expect(error).toHaveBeenCalledWith('Exceptional!')
     })
 
@@ -174,124 +152,119 @@ describe('Observable', function() {
 
       observable.subscribe(() => {}, () => {}, complete)
 
+      scheduler().flush()
+
       expect(complete).toHaveBeenCalled()
     })
   })
 
   describe('cleanup', function() {
-    it('executes a clean up method when cancelled', () => {
-      let cleaned = false
+    it('executes a clean up method when completed', () => {
+      let cleaned = jest.fn()
 
       let observable = new Observable(observer => {
         observer.complete()
-        return () => {
-          cleaned = true
-        }
+        return cleaned
       })
 
       observable.subscribe(n => n)
 
-      expect(cleaned).toBe(true)
-    })
+      scheduler().flush()
 
-    it.dev('raises if cleanup is not a function', () => {
-      expect.assertions(1)
+      expect(cleaned).toHaveBeenCalled()
+    })
 
+    it('noops if not given a function', () => {
       let observable = new Observable(observer => {
         observer.complete()
         return true
       })
 
-      try {
-        observable.subscribe(n => n)
-      } catch (x) {
-        expect(x.message).toEqual('true is not a function')
-      }
+      observable.subscribe(n => n)
+
+      expect(() => scheduler().flush()).not.toThrow()
     })
-  })
 
-  describe('cancellation', function() {
-    it('can kill all active subscriptions', function() {
-      let observer = new Observable(observer => {
-        observer.next(2)
-      })
+    it('the scheduler covers cases when cleanup fails', () => {
+      expect.assertions(1)
 
-      let cancel = jest.fn()
+      let observable = new Observable(observer => {
+        observer.complete()
 
-      observer.subscribe({ cancel })
-      observer.subscribe({ cancel })
+        return () => {
+          throw 'Exceptional!'
+        }
+      })
 
-      observer.cancel('Nevermind')
+      observable.subscribe(n => n)
 
-      expect(cancel).toHaveBeenCalledTimes(2)
-      expect(cancel).toHaveBeenCalledWith('Nevermind')
+      try {
+        scheduler().flush()
+      } catch (error) {
+        expect(error.message).toContain('Exceptional!')
+      }
     })
 
-    it('invokes cleanup', async function() {
-      let observer = new Observable(observer => {
-        observer.next(2)
+    it('unsubscribes subscriptions', () => {
+      let unsubscribe = jest.fn()
 
-        let timer = setTimeout(observer.complete, 0)
+      let range = new Observable(observer => {
+        observer.next(1)
+        observer.next(2)
+        observer.next(3)
+        observer.complete()
 
-        return () => clearTimeout(timer)
+        return unsubscribe
       })
 
-      let complete = jest.fn()
-
-      observer.subscribe({ complete })
+      let observable = new Observable(observer => {
+        return range.subscribe(observer)
+      })
 
-      observer.cancel('Nevermind')
+      observable.subscribe(jest.fn())
 
-      await new Promise(resolve => setTimeout(resolve, 10))
+      scheduler().flush()
 
-      expect(complete).toHaveBeenCalledTimes(0)
+      expect(unsubscribe).toHaveBeenCalled()
     })
-  })
 
-  describe('.wrap', async () => {
-    it('wraps promises', async () => {
+    it('the scheduler covers cases when a nested cleanup method fails', () => {
       expect.assertions(1)
 
-      let wrapped = Observable.wrap(Promise.resolve(true))
+      let range = new Observable(observer => {
+        observer.next(1)
+        observer.next(2)
+        observer.next(3)
+        observer.complete()
 
-      wrapped.subscribe({
-        next: value => {
-          expect(value).toBe(true)
+        return () => {
+          throw 'Exceptional!'
         }
       })
 
-      await wrapped
-    })
-
-    it('wraps primitive values', async () => {
-      expect.assertions(1)
-
-      let wrapped = Observable.wrap(true)
-
-      wrapped.subscribe({
-        next: value => {
-          expect(value).toBe(true)
-        }
+      let observable = new Observable(observer => {
+        return range.subscribe(observer)
       })
 
-      await wrapped
+      observable.subscribe(jest.fn())
+
+      try {
+        scheduler().flush()
+      } catch (error) {
+        expect(error.message).toContain('Exceptional!')
+      }
     })
   })
 
-  describe('.map', () => {
+  describe('::map', () => {
     it('operates like array.map over a stream of values', () => {
-      let observable = new Observable(observer => {
-        for (var i = 0; i < 5; i++) {
-          observer.next(i)
-        }
-        observer.complete()
-      })
-
-      let double = observable.map(value => value * 2)
+      let double = Observable.of(0, 1, 2, 3, 4).map(value => value * 2)
       let answers = []
 
       double.subscribe(value => answers.push(value))
 
+      scheduler().flush()
+
       expect(answers).toEqual([0, 2, 4, 6, 8])
     })
 
@@ -303,19 +276,37 @@ describe('Observable', function() {
         }
       }
 
+      let double = Observable.of(0, 1, 2, 3, 4).map(scope.multiply, scope)
+      let answers = []
+
+      double.subscribe(value => answers.push(value))
+
+      scheduler().flush()
+
+      expect(answers).toEqual([0, 2, 4, 6, 8])
+    })
+
+    it('passes along errors', () => {
       let observable = new Observable(observer => {
         for (var i = 0; i < 5; i++) {
           observer.next(i)
         }
-        observer.complete()
+        throw 'Exceptional!'
       })
 
-      let double = observable.map(scope.multiply, scope)
-      let answers = []
+      let error = jest.fn()
 
-      double.subscribe(value => answers.push(value))
+      observable.map(n => n).subscribe({ error })
 
-      expect(answers).toEqual([0, 2, 4, 6, 8])
+      scheduler().flush()
+
+      expect(error).toHaveBeenCalledWith('Exceptional!')
+    })
+
+    it('raises if not given a function', () => {
+      expect(() => Observable.of(0, 1, 2, 3, 4).map()).toThrow(
+        'undefined is not a function'
+      )
     })
   })
 })