Repository: vigetlabs/microcosm
Pull request #513 (2018-04-25 21:03:33)

[microcosm] Async, scheduled, subjects
This commit brings the asychronous behavior of Observables over to
Subjects. It is also able to consolidate some behavior by extending
Observable.

The async behavior of Subjects is different in that they track a
present value. When subscribed to, Subjects immediately emit an
update. All future work is scheduled.

This is a really nice system because it allows Presenter view models
to immediately have a value, and for state to be synchronized by the
time Effects dispatch.


Patch:
diff --git a/packages/microcosm/src/subject.js b/packages/microcosm/src/subject.js
index 6322c3e3..2a9010c8 100644
--- a/packages/microcosm/src/subject.js
+++ b/packages/microcosm/src/subject.js
@@ -1,140 +1,119 @@
-// @flow
+/**
+ * @fileoverview A subject is a type of observable that is useful when:
+ *
+ * 1. You need a value right away.
+ * 2. A job should run only once, but be subscribed to many times.
+ *
+ * Unlike Observables, when you subscribe to a Subject you immediately
+ * receive a "next" callback with the current payload. This is important
+ * for a couple of use cases that require immediate access to state:
+ *
+ * 1. Domains
+ * 2. Actions
+ * 3. Presenter view models
+ *
+ * @flow
+ */
+import assert from 'assert'
 import { type Microcosm } from './microcosm'
-import { Observable, Observer, getObservable } from './observable'
+import {
+  Observable,
+  type SubscriptionObserver,
+  type Cleanup
+} from './observable'
 import { observable } from './symbols'
-import { noop, EMPTY_SUBSCRIPTION } from './empty'
 import { set, merge } from './data'
+import { isPromise, isObservable, isObject, isPlainObject } from './type-checks'
 
-export class Subject {
+export class Subject extends Observable {
   meta: { key: *, status: string, origin: Microcosm }
-  payload: *
+  payload: any
   disabled: boolean
-  _observers: Set<*>
-  _observable: Observable
+
+  _observers: SubscriptionObserver[]
+  _subscriber: SubscriptionObserver => ?Cleanup
 
   constructor(payload?: *, meta?: Object) {
+    super()
+
     this.meta = merge({ key: 'subject', status: 'start' }, meta)
     this.payload = payload
     this.disabled = false
 
-    this._observers = new Set()
-
-    this._observable = new Observable(observer => {
-      this._observers.add(observer)
-      return () => this._observers.delete(observer)
-    })
+    this._subscriber = this._multicast
+    this._observers = []
   }
 
-  get started(): boolean {
-    return this.status !== 'start'
-  }
+  _multicast(observer: SubscriptionObserver): Cleanup {
+    this._observers.push(observer)
 
-  get status(): string {
-    return this.meta.status
-  }
+    switch (this.status) {
+      case 'next':
+        observer.next(this.payload)
+        break
+      case 'complete':
+        observer.next(this.payload)
+        observer.complete()
+        break
+      case 'error':
+        observer.error(this.payload)
+        break
+      case 'cancel':
+        observer.cancel()
+        break
+    }
 
-  get completed(): boolean {
-    return this.status === 'complete'
+    return remove.bind(null, this._observers, observer)
   }
 
-  get cancelled(): boolean {
-    return this.status === 'cancel'
+  get status(): string {
+    return this.meta.status
   }
 
   get closed(): boolean {
-    return this.errored || this.completed || this.cancelled
-  }
-
-  get errored(): boolean {
-    return this.status === 'error'
-  }
-
-  get next(): * {
-    if (this.closed) {
-      return noop
-    }
-
-    return value => {
-      this.meta.status = 'next'
-
-      if (value != null) {
-        this.payload = value
-      }
-
-      this._observers.forEach(observer => observer.next(value))
+    switch (this.status) {
+      case 'error':
+      case 'complete':
+      case 'cancel':
+        return true
+      default:
+        return false
     }
   }
 
-  get complete(): * {
-    return value => {
-      this.meta.status = 'complete'
-
-      if (value != null) {
-        this.payload = value
-      }
-
-      this._observers.forEach(observer => observer.complete())
-    }
+  get next() {
+    return update.bind(null, this, 'next')
   }
 
-  get error(): * {
-    return error => {
-      this.payload = error
-      this.meta.status = 'error'
-
-      this._observers.forEach(observer => observer.error(error))
-    }
+  get complete() {
+    return update.bind(null, this, 'complete')
   }
 
-  get cancel(): * {
-    return reason => {
-      this.payload = reason
-      this.meta.status = 'cancel'
+  get error() {
+    return update.bind(null, this, 'error')
+  }
 
-      this._observers.forEach(observer => observer.cancel(reason))
-    }
+  get cancel() {
+    return update.bind(null, this, 'cancel')
   }
 
-  get clear(): * {
+  get clear() {
     return () => {
-      this.payload = null
       this.meta.status = 'start'
+      this.payload = null
 
-      this._observers.forEach(observer => observer.cancel())
-    }
-  }
-
-  subscribe(next: *): * {
-    let observer = new Observer(...arguments)
-
-    if (this.closed) {
-      if (this.completed) {
-        observer.next(this.payload)
+      for (var i = 0; i < this._observers.length; i++) {
+        this._observers[i].unsubscribe()
       }
-
-      // $FlowFixMe - Dynamic keys
-      observer[this.status](this.payload)
-    } else {
-      return new Observable(observer => {
-        if (this.meta.status === 'next') {
-          observer.next(this.payload)
-        }
-
-        return this._observable.subscribe(observer)
-      }).subscribe(observer)
     }
-
-    return EMPTY_SUBSCRIPTION
   }
 
   then(pass: *, fail: *): Promise<*> {
     return new Promise((resolve, reject) => {
-      this.subscribe({
-        complete: () => resolve(this.payload),
-        error: () => reject(this.payload),
-        cancel: () => resolve(this.payload)
-      })
-    }).then(pass, fail)
+      this.subscribe(null, reject, resolve, resolve)
+    })
+      .then(() => this.payload)
+      .then(pass, fail)
   }
 
   // $FlowFixMe - Flow doesn't understand computed keys :-/
@@ -158,87 +137,151 @@ export class Subject {
     return this.payload
   }
 
-  map(fn: (*) => *, scope: any): Observable {
-    return new Observable(observer => {
-      if (this.started) {
-        observer.next(fn.call(scope, this.payload))
+  static hash(obj: *): Subject {
+    let isArray = Array.isArray(obj)
+    let isPojo = isPlainObject(obj)
+
+    if (!isArray && !isPojo) {
+      return Subject.from(obj)
+    }
+
+    let payload = isArray ? [] : {}
+    let subject = new Subject(payload)
+    let building = true
+    let jobs = 0
+
+    function taskFinished() {
+      if (--jobs <= 0) {
+        if (!building) {
+          subject.complete(payload)
+        }
       }
+    }
 
-      if (this.closed) {
-        observer.complete()
-      } else {
-        return this._observable.map(fn, scope).subscribe(observer)
+    function assign(key: string, value: *): void {
+      // $FlowFixMe - Signature of set() is too strict
+      payload = set(payload, key, value)
+
+      if (payload !== subject.payload) {
+        subject.next(payload)
       }
-    })
-  }
+    }
 
-  every(fn: (subject: this) => void, scope: any): * {
-    let dispatch = fn.bind(scope, this)
+    for (var key in obj) {
+      let value = obj[key]
 
-    return this.subscribe({
-      next: dispatch,
-      error: dispatch,
-      complete: dispatch,
-      cancel: dispatch
-    })
-  }
+      if (!isObject(value)) {
+        payload = set(payload, key, value)
+        continue
+      }
 
-  static hash(obj: *): Subject {
-    if (obj instanceof Subject) {
-      return obj
-    }
+      jobs += 1
 
-    let subject = new Subject()
+      if (value instanceof Subject) {
+        payload = set(payload, key, value.payload)
+      }
 
-    if (getObservable(obj)) {
-      obj.subscribe(subject)
-      return subject
-    }
+      let subscription = Subject.hash(value).subscribe({
+        next: assign.bind(null, key),
+        error: subject.error,
+        complete: taskFinished,
+        cancel: taskFinished
+      })
 
-    if (obj == null || typeof obj !== 'object') {
-      subject.next(obj)
-      subject.complete()
-      return subject
+      subject.subscribe({ cancel: subscription.unsubscribe })
     }
 
-    if (typeof obj.then === 'function') {
-      subject.subscribe(Observable.fromPromise(obj).subscribe(subject))
-      return subject
+    if (payload !== subject.payload) {
+      subject.next(payload)
     }
 
-    let keys = Object.keys(obj)
-    let payload = Array.isArray(obj) ? [] : {}
-    let jobs = keys.length
+    building = false
 
-    subject.next(payload)
+    if (jobs <= 0) {
+      subject.complete()
+    }
 
-    function complete() {
-      if (--jobs <= 0) {
-        subject.complete()
-      }
+    return subject
+  }
+
+  static from(source: *): Subject {
+    if (source instanceof Subject) {
+      return source
     }
 
-    function assign(key: string, value: *): void {
-      // $FlowFixMe - Signature of set() is too strict
-      payload = set(payload, key, value)
+    if (isObservable(source)) {
+      return fromObservable(source)
+    }
 
-      if (payload !== subject.payload) {
-        subject.next(payload)
-      }
+    if (isPromise(source)) {
+      return fromPromise(source)
     }
 
-    for (var i = 0, len = keys.length; i < len; i++) {
-      let key = keys[i]
+    return Subject.of(source)
+  }
 
-      let subscription = Observable.wrap(obj[key]).subscribe({
-        next: value => assign(key, value),
-        complete: complete,
-        error: subject.error
-      })
+  // $FlowFixMe This function enumerates on variable arguments
+  static of() {
+    let subject = new Subject()
 
-      subject.subscribe(subscription)
+    for (var i = 0; i < arguments.length; i++) {
+      subject.next(arguments[i])
     }
 
+    subject.complete()
+
     return subject
   }
 }
+
+function fromPromise(promise: Promise<*>): Subject {
+  let subject = new Subject()
+
+  promise
+    .then(subject.next)
+    .then(() => subject.complete())
+    .catch(subject.error)
+
+  return subject
+}
+
+function fromObservable(observable: Observable): Subject {
+  let subject = new Subject()
+  observable.subscribe(subject)
+  return subject
+}
+
+function remove(list: *[], item: *): void {
+  let index = list.indexOf(item)
+
+  if (index >= 0) {
+    list.splice(index, -1)
+  } else {
+    assert(false, 'Attempted to remove an observer that was never added.')
+  }
+}
+
+function update(subject: Subject, status: string, value: *): void {
+  if (subject.closed) {
+    return
+  }
+
+  subject.meta.status = status
+
+  if (arguments.length > 2) {
+    subject.payload = value
+  }
+
+  subject._observers.forEach(observer => {
+    switch (status) {
+      case 'next':
+        return observer.next(subject.payload)
+      case 'error':
+        return observer.error(subject.payload)
+      case 'complete':
+        return observer.complete()
+      case 'cancel':
+        return observer.cancel()
+    }
+  })
+}


Test patch:
diff --git a/packages/microcosm/test/unit/subject.test.js b/packages/microcosm/test/unit/subject.test.js
index f14dfa04..bb006eac 100644
--- a/packages/microcosm/test/unit/subject.test.js
+++ b/packages/microcosm/test/unit/subject.test.js
@@ -1,8 +1,8 @@
-import { Subject, Observable } from 'microcosm'
+import { Subject, Observable, scheduler } from 'microcosm'
 
 describe('Subject', function() {
   describe('next', function() {
-    it('emits an update', () => {
+    it('emits an update', async () => {
       let subject = new Subject()
       let next = jest.fn()
 
@@ -10,12 +10,14 @@ describe('Subject', function() {
       subject.next(1)
       subject.next(2)
 
+      await scheduler()
+
       expect(next).toHaveBeenCalledTimes(2)
       expect(next).toHaveBeenCalledWith(1)
       expect(next).toHaveBeenCalledWith(2)
     })
 
-    it('can be subscribed to mid-update', () => {
+    it('can be subscribed to mid-update', async () => {
       let subject = new Subject()
       let next = jest.fn()
 
@@ -23,12 +25,14 @@ describe('Subject', function() {
       subject.subscribe({ next })
       subject.next(2)
 
+      await scheduler()
+
       expect(next).toHaveBeenCalledTimes(2)
       expect(next).toHaveBeenCalledWith(1)
       expect(next).toHaveBeenCalledWith(2)
     })
 
-    it('does not update the payload if given null', () => {
+    it('does not update the payload if given null', async () => {
       let subject = new Subject()
       let next = jest.fn()
 
@@ -36,12 +40,14 @@ describe('Subject', function() {
       subject.next(1)
       subject.next(null)
 
+      await scheduler()
+
       expect(next).toHaveBeenCalledTimes(2)
       expect(next).toHaveBeenCalledWith(1)
       expect(next).toHaveBeenCalledWith(1)
     })
 
-    it('triggers if next only once after being completed', () => {
+    it('triggers if next only once after being completed', async () => {
       let subject = new Subject()
       let next = jest.fn()
 
@@ -49,11 +55,13 @@ describe('Subject', function() {
       subject.subscribe({ next })
       subject.next(2)
 
+      await scheduler()
+
       expect(next).toHaveBeenCalledTimes(1)
       expect(next).toHaveBeenCalledWith(1)
     })
 
-    it('does not trigger after being completed', () => {
+    it('does not trigger after being completed', async () => {
       let subject = new Subject()
       let next = jest.fn()
 
@@ -61,10 +69,12 @@ describe('Subject', function() {
       subject.complete()
       subject.next(true)
 
+      await scheduler()
+
       expect(next).toHaveBeenCalledTimes(0)
     })
 
-    it('does not trigger after erroring', () => {
+    it('does not trigger after erroring', async () => {
       let subject = new Subject()
       let next = jest.fn()
 
@@ -72,10 +82,12 @@ describe('Subject', function() {
       subject.subscribe({ next })
       subject.next(true)
 
+      await scheduler()
+
       expect(next).toHaveBeenCalledTimes(0)
     })
 
-    it('does not trigger after cancelling', () => {
+    it('does not trigger after cancelling', async () => {
       let subject = new Subject()
       let next = jest.fn()
 
@@ -83,12 +95,14 @@ describe('Subject', function() {
       subject.subscribe({ next })
       subject.next(true)
 
+      await scheduler()
+
       expect(next).toHaveBeenCalledTimes(0)
     })
   })
 
   describe('complete', function() {
-    it('triggers once', () => {
+    it('triggers once', async () => {
       let subject = new Subject()
       let complete = jest.fn()
 
@@ -97,44 +111,52 @@ describe('Subject', function() {
       subject.complete()
       subject.complete()
 
+      await scheduler()
+
       expect(subject.closed).toBe(true)
       expect(complete).toHaveBeenCalledTimes(1)
     })
 
-    it('triggers if already completed', () => {
+    it('triggers if already completed', async () => {
       let subject = new Subject()
       let complete = jest.fn()
 
       subject.complete()
       subject.subscribe({ complete })
 
+      await scheduler()
+
       expect(subject.closed).toBe(true)
       expect(complete).toHaveBeenCalledTimes(1)
     })
 
-    it('does not trigger unsubscribe', () => {
+    it('does not trigger unsubscribe', async () => {
       let subject = new Subject()
       let unsubscribe = jest.fn()
 
       subject.subscribe({ unsubscribe })
       subject.complete(true)
 
+      await scheduler()
+
       expect(unsubscribe).not.toHaveBeenCalled()
     })
   })
 
   describe('error', function() {
-    it('invokes the error again if already errored', () => {
+    it('invokes the error again if already errored', async () => {
       let subject = new Subject()
       let error = jest.fn()
 
       subject.error()
       subject.subscribe({ error })
 
+      await scheduler()
+
       expect(error).toHaveBeenCalled()
     })
 
-    it('does not invoke the error if already completed', () => {
+    it('does not invoke the error if already completed', async () => {
       let subject = new Subject()
       let error = jest.fn()
 
@@ -142,29 +164,21 @@ describe('Subject', function() {
       subject.complete()
       subject.error()
 
+      await scheduler()
+
       expect(error).not.toHaveBeenCalled()
     })
   })
 
   describe('cancel', function() {
-    it('becomes closed when cancelled', function() {
+    it('becomes closed when cancelled', async () => {
       let subject = new Subject()
 
       subject.cancel()
 
-      expect(subject.closed).toBe(true)
-    })
+      await scheduler()
 
-    it('cancel is a one time binding', function() {
-      let subject = new Subject()
-      let cancel = jest.fn()
-
-      subject.subscribe({ cancel })
-
-      subject.cancel()
-      subject.cancel()
-
-      expect(cancel).toHaveBeenCalledTimes(1)
+      expect(subject.closed).toBe(true)
     })
   })
 
@@ -182,13 +196,13 @@ describe('Subject', function() {
 
     it('cancels observers', function() {
       let subject = new Subject()
-      let cancel = jest.fn()
-
-      subject.subscribe({ cancel })
+      let next = jest.fn()
 
+      subject.subscribe({ next })
       subject.clear()
+      subject.next(1)
 
-      expect(cancel).toHaveBeenCalledTimes(1)
+      expect(next).not.toHaveBeenCalled()
     })
   })
 
@@ -242,7 +256,7 @@ describe('Subject', function() {
     it('returns itself', () => {
       let subject = new Subject()
 
-      expect(Observable.wrap(subject)).toBe(subject)
+      expect(Subject.from(subject)).toBe(subject)
     })
   })
 
@@ -258,7 +272,7 @@ describe('Subject', function() {
         }
       })
 
-      await hash
+      await scheduler()
     })
 
     it('works on primitive values', async () => {
@@ -272,7 +286,7 @@ describe('Subject', function() {
         }
       })
 
-      await hash
+      await scheduler()
     })
 
     it('works on other observables', async () => {
@@ -286,7 +300,7 @@ describe('Subject', function() {
         complete
       })
 
-      await hash
+      await scheduler()
 
       expect(next).toHaveBeenCalledWith({ key: 3 })
       expect(complete).toHaveBeenCalled()
@@ -300,12 +314,9 @@ describe('Subject', function() {
         key: Observable.of(1, 1, 1)
       })
 
-      hash.subscribe({
-        next,
-        complete
-      })
+      hash.subscribe({ next, complete })
 
-      await hash
+      await scheduler()
 
       expect(next).toHaveBeenCalledWith({ key: 1 })
       expect(next).toHaveBeenCalledTimes(1)
@@ -327,7 +338,9 @@ describe('Subject', function() {
       subject.next(2)
       subject.complete()
 
-      await hash
+      await scheduler()
+
+      expect(hash.payload).toEqual({ key: 2 })
 
       expect(next).toHaveBeenCalledWith({ key: 1 })
       expect(next).toHaveBeenCalledWith({ key: 2 })
@@ -336,7 +349,7 @@ describe('Subject', function() {
   })
 
   describe('.map', () => {
-    it('works like observable.map', () => {
+    it('works like observable.map', async () => {
       let subject = new Subject()
 
       let double = subject.map(value => value * 2)
@@ -348,10 +361,12 @@ describe('Subject', function() {
         subject.next(i)
       }
 
+      await scheduler()
+
       expect(answers).toEqual([0, 2, 4, 6, 8])
     })
 
-    it('starts with the current value', () => {
+    it('starts with the current value', async () => {
       let subject = new Subject()
 
       let double = subject.map(value => value * 2)
@@ -361,10 +376,12 @@ describe('Subject', function() {
 
       double.subscribe(value => answers.push(value))
 
+      await scheduler()
+
       expect(answers).toEqual([10])
     })
 
-    it('only maps over the current and new iterations', () => {
+    it('only maps over the current and new iterations', async () => {
       let subject = new Subject()
 
       let double = subject.map(value => value * 2)
@@ -376,10 +393,12 @@ describe('Subject', function() {
       subject.next(2)
       subject.next(3)
 
+      await scheduler()
+
       expect(answers).toEqual([2, 4, 6])
     })
 
-    it('only emits the final value when it completes', () => {
+    it('only emits the final value when it completes', async () => {
       let subject = new Subject()
 
       let double = subject.map(value => value * 2)
@@ -392,6 +411,8 @@ describe('Subject', function() {
       subject.next(2)
       subject.next(3)
 
+      await scheduler()
+
       expect(answers).toEqual([2])
     })
   })