Repository: vigetlabs/microcosm
Pull request #519 (2018-06-18 12:35:46)

Pass next when completing with a payload. Improve efficiency
**What**

As a part of the Microcosm 13 update, actions are moving away from a custom data type and embrace the [Observable](https://github.com/tc39/proposal-observable) standard interface. Unfortunately, that causes a few API incompatibilities that I'm trying to smooth over before a full release of 13.x.

In Microcosm 12.x, you can complete an action with a payload like:

```javascript
action.resolve('stuff')
```

As far as I understand, with [Observables](https://github.com/tc39/proposal-observable), the correct way to accomplish this is something like:

```javascript
observer.next('stuff')
observer.complete()
```

While I've respected that for Observables themselves, this PR updates the Subject API to allow passing a payload to `complete`. This improves ergonomics and compatibility with usage in Microcosm 12.x.

**Other related work**

The drawback to this approach is that an extra action status is dispatched. I have alleviated that by adding a `filter` method to the Observable prototype that Domains use to avoid dispatch if an action update has no impact on their state. Essentially:

```javascript
// In a Domain
// History is a tree of actions that reprocess in a consistent order as they change
history 
  .filter(action => {
    // Can I process this action at all?
    return this.shouldListenTo(action)
  })
  .subscribe(action => {
    // Update the domain's state value
    this.dispatch(action)
  })
```

This actually ended up being a greater efficiency update than I anticipated, allowing me to unskip a few tests. Pretty neat. I've been wanting to add `filter` to the core API, but was wary of doing so until I could dogfood it internally. Now I can!

codecov-io: # [Codecov](https://codecov.io/gh/vigetlabs/microcosm/pull/519?src=pr&el=h1) Report
> :exclamation: No coverage uploaded for pull request base (`master@fbd3d38`). [Click here to learn what that means](https://docs.codecov.io/docs/error-reference#section-missing-base-commit).
> The diff coverage is `94.44%`.

[![Impacted file tree graph](https://codecov.io/gh/vigetlabs/microcosm/pull/519/graphs/tree.svg?width=650&height=150&src=pr&token=AaE0FLz9sU)](https://codecov.io/gh/vigetlabs/microcosm/pull/519?src=pr&el=tree)

```diff
@@            Coverage Diff            @@
##             master     #519   +/-   ##
=========================================
  Coverage          ?   93.83%           
=========================================
  Files             ?       32           
  Lines             ?      795           
  Branches          ?      159           
=========================================
  Hits              ?      746           
  Misses            ?       43           
  Partials          ?        6
```


| [Impacted Files](https://codecov.io/gh/vigetlabs/microcosm/pull/519?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [packages/microcosm/src/ledger.js](https://codecov.io/gh/vigetlabs/microcosm/pull/519/diff?src=pr&el=tree#diff-cGFja2FnZXMvbWljcm9jb3NtL3NyYy9sZWRnZXIuanM=) | `100% <100%> (ø)` | |
| [packages/microcosm/src/agent.js](https://codecov.io/gh/vigetlabs/microcosm/pull/519/diff?src=pr&el=tree#diff-cGFja2FnZXMvbWljcm9jb3NtL3NyYy9hZ2VudC5qcw==) | `100% <100%> (ø)` | |
| [packages/microcosm/src/domain.js](https://codecov.io/gh/vigetlabs/microcosm/pull/519/diff?src=pr&el=tree#diff-cGFja2FnZXMvbWljcm9jb3NtL3NyYy9kb21haW4uanM=) | `100% <100%> (ø)` | |
| [packages/microcosm/src/lifecycle.js](https://codecov.io/gh/vigetlabs/microcosm/pull/519/diff?src=pr&el=tree#diff-cGFja2FnZXMvbWljcm9jb3NtL3NyYy9saWZlY3ljbGUuanM=) | `100% <100%> (ø)` | |
| [packages/microcosm/src/subject.js](https://codecov.io/gh/vigetlabs/microcosm/pull/519/diff?src=pr&el=tree#diff-cGFja2FnZXMvbWljcm9jb3NtL3NyYy9zdWJqZWN0Lmpz) | `100% <100%> (ø)` | |
| [packages/microcosm/src/coroutine.js](https://codecov.io/gh/vigetlabs/microcosm/pull/519/diff?src=pr&el=tree#diff-cGFja2FnZXMvbWljcm9jb3NtL3NyYy9jb3JvdXRpbmUuanM=) | `100% <100%> (ø)` | |
| [packages/microcosm/src/observable.js](https://codecov.io/gh/vigetlabs/microcosm/pull/519/diff?src=pr&el=tree#diff-cGFja2FnZXMvbWljcm9jb3NtL3NyYy9vYnNlcnZhYmxlLmpz) | `88.69% <90%> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/vigetlabs/microcosm/pull/519?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/vigetlabs/microcosm/pull/519?src=pr&el=footer). Last update [fbd3d38...4cc1b64](https://codecov.io/gh/vigetlabs/microcosm/pull/519?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

nhunzaker: @zporter Thanks! Honestly, as stop and go as this has been, they've been super helpful for me too.

Patch:
diff --git a/packages/microcosm/src/agent.js b/packages/microcosm/src/agent.js
index 9e4b56bb..cc727cb1 100644
--- a/packages/microcosm/src/agent.js
+++ b/packages/microcosm/src/agent.js
@@ -21,7 +21,9 @@ export class Agent extends Subject {
 
     this._preHistory()
 
-    let tracker = repo.history.subscribe(this.receive.bind(this))
+    let tracker = repo.history
+      .filter(this._shouldListenTo, this)
+      .subscribe(this.receive.bind(this))
 
     repo.subscribe({
       complete: () => {
@@ -33,10 +35,6 @@ export class Agent extends Subject {
     this.setup(this.repo, this.options)
   }
 
-  _preHistory() {
-    // NOOP by default
-  }
-
   /**
    * Setup runs right after an agent is added to a Microcosm. It
    * receives that repo and any options passed as the second argument.
@@ -54,4 +52,14 @@ export class Agent extends Subject {
    * Called whenever an agent receives a new action
    */
   receive(action: Subject): void {}
+
+  // Private -------------------------------------------------- //
+
+  _preHistory() {
+    // NOOP by default
+  }
+
+  _shouldListenTo(action: Subject): boolean {
+    return true
+  }
 }
diff --git a/packages/microcosm/src/coroutine.js b/packages/microcosm/src/coroutine.js
index 88767152..6fb960f5 100644
--- a/packages/microcosm/src/coroutine.js
+++ b/packages/microcosm/src/coroutine.js
@@ -35,10 +35,7 @@ export function coroutine(
   }
 
   if (isPromise(body)) {
-    body
-      .then(action.next)
-      .then(() => action.complete())
-      .catch(action.error)
+    body.then(action.complete).catch(action.error)
     return
   }
 
diff --git a/packages/microcosm/src/domain.js b/packages/microcosm/src/domain.js
index 4e8a02fa..c537a19a 100644
--- a/packages/microcosm/src/domain.js
+++ b/packages/microcosm/src/domain.js
@@ -72,13 +72,6 @@ export class Domain<State: any = Object> extends Agent {
   }
 
   receive(action: Subject): void {
-    // Avoid a situation where history dispatches before a domain
-    // is fully set up. Domains move into the "next" state when
-    // construction is finished (getInitialState())
-    if (this.status === 'start') {
-      return
-    }
-
     let next = this._rollforward(action)
 
     if (next !== this.payload) {
@@ -96,6 +89,24 @@ export class Domain<State: any = Object> extends Agent {
     return this.serialize(this.valueOf())
   }
 
+  _shouldListenTo(action: Subject): boolean {
+    // Avoid a situation where history dispatches before a domain
+    // is fully set up. Domains move into the "next" state when
+    // construction is finished (getInitialState())
+    if (this.status === START) {
+      return false
+    }
+
+    // If this domain has processed this action previously, the ledger
+    // needs to roll back the state change
+    if (this._ledger.has(action)) {
+      return true
+    }
+
+    // Otherwise, only listen to this action if there are registrations
+    return this._resolve(action).length > 0
+  }
+
   _resolve(action: Subject): DomainHandler<State>[] {
     switch (action.meta.key) {
       case RESET_KEY:
@@ -123,11 +134,11 @@ export class Domain<State: any = Object> extends Agent {
   }
 
   _rollforward(action: Subject): State {
-    let state = this._ledger.rebase(action)
     let focus = action
+    let state = this._ledger.rebase(focus)
 
     while (focus) {
-      if (!focus.disabled && focus.status !== 'start') {
+      if (!focus.disabled) {
         state = this._dispatch(state, focus)
         this._ledger.set(focus, state)
       }
diff --git a/packages/microcosm/src/ledger.js b/packages/microcosm/src/ledger.js
index 75a505ce..4f0d87c2 100644
--- a/packages/microcosm/src/ledger.js
+++ b/packages/microcosm/src/ledger.js
@@ -28,6 +28,10 @@ export class Ledger<State> {
     }
   }
 
+  has(action: Subject): boolean {
+    return this._versions.has(action)
+  }
+
   recall(action: ?Subject): State {
     let focus = action
 
diff --git a/packages/microcosm/src/lifecycle.js b/packages/microcosm/src/lifecycle.js
index 23f98923..55ea4ad3 100644
--- a/packages/microcosm/src/lifecycle.js
+++ b/packages/microcosm/src/lifecycle.js
@@ -17,16 +17,14 @@ export const RESET = tag(parse.bind(null), '$reset')
 
 export const PATCH = tag(parse.bind(null), '$patch')
 
-export const INITIAL_STATE = '$initialState'
-
-export const INACTIVE = 'inactive'
-
 export const START = 'start'
 
-export const COMPLETE = 'complete'
-
 export const NEXT = 'next'
 
 export const ERROR = 'error'
 
+export const COMPLETE = 'complete'
+
+export const CANCEL = 'cancel'
+
 export const UNSUBSCRIBE = 'unsubscribe'
diff --git a/packages/microcosm/src/observable.js b/packages/microcosm/src/observable.js
index 272cdc1f..8b252c62 100644
--- a/packages/microcosm/src/observable.js
+++ b/packages/microcosm/src/observable.js
@@ -8,6 +8,7 @@ import { observable } from './symbols'
 import { noop } from './empty'
 import { scheduler } from './scheduler'
 import { isObject } from './type-checks'
+import { NEXT, ERROR, COMPLETE, CANCEL } from './lifecycle'
 
 // A subscription is constructing
 const START = 0
@@ -20,11 +21,6 @@ const WORKING = 3
 // A subscription is done
 const CLOSED = 4
 
-const NEXT = 'next'
-const ERROR = 'error'
-const COMPLETE = 'complete'
-const CANCEL = 'cancel'
-
 export interface Unsubscribable {
   unsubscribe(): any;
 }
@@ -152,6 +148,25 @@ export class Observable {
     })
   }
 
+  filter(fn: (*) => boolean, scope: any): Observable {
+    if (typeof fn !== 'function') {
+      throw new TypeError(fn + ' is not a function')
+    }
+
+    return new Observable(observer => {
+      return this.subscribe(
+        value => {
+          if (fn.call(scope, value)) {
+            observer.next(value)
+          }
+        },
+        observer.error,
+        observer.complete,
+        observer.cancel
+      )
+    })
+  }
+
   flatMap(fn: (*) => Observable, scope: any): Observable {
     if (typeof fn !== 'function') {
       throw new TypeError(fn + ' is not a function')
@@ -248,32 +263,28 @@ function notifySubscription(subscription: *, type: *, value: *) {
   let observer = subscription._observer
 
   // 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)
+  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`
+      )
   }
 
   if (subscription.closed) {
diff --git a/packages/microcosm/src/subject.js b/packages/microcosm/src/subject.js
index c30707a5..ba8b87d3 100644
--- a/packages/microcosm/src/subject.js
+++ b/packages/microcosm/src/subject.js
@@ -22,6 +22,7 @@ import {
 } from './observable'
 import { merge } from './data'
 import { isPromise, isObservable } from './type-checks'
+import { START, NEXT, COMPLETE, ERROR, CANCEL } from './lifecycle'
 
 export class Subject extends Observable {
   meta: { key: *, status: string, origin: * }
@@ -34,7 +35,7 @@ export class Subject extends Observable {
   constructor(payload?: *, meta?: Object) {
     super()
 
-    this.meta = merge({ key: 'subject', status: 'start' }, meta)
+    this.meta = merge({ key: 'subject', status: START }, meta)
     this.payload = payload
     this.disabled = false
 
@@ -46,17 +47,17 @@ export class Subject extends Observable {
     this._observers.add(observer)
 
     switch (this.status) {
-      case 'next':
+      case NEXT:
         observer.next(this.payload)
         break
-      case 'complete':
+      case COMPLETE:
         observer.next(this.payload)
         observer.complete()
         break
-      case 'error':
+      case ERROR:
         observer.error(this.payload)
         break
-      case 'cancel':
+      case CANCEL:
         observer.cancel()
         break
     }
@@ -70,9 +71,9 @@ export class Subject extends Observable {
 
   get closed(): boolean {
     switch (this.status) {
-      case 'error':
-      case 'complete':
-      case 'cancel':
+      case ERROR:
+      case COMPLETE:
+      case CANCEL:
         return true
       default:
         return false
@@ -80,24 +81,24 @@ export class Subject extends Observable {
   }
 
   get next() {
-    return update.bind(null, this, 'next')
+    return update.bind(null, this, NEXT)
   }
 
   get complete() {
-    return update.bind(null, this, 'complete')
+    return update.bind(null, this, COMPLETE)
   }
 
   get error() {
-    return update.bind(null, this, 'error')
+    return update.bind(null, this, ERROR)
   }
 
   get cancel() {
-    return update.bind(null, this, 'cancel')
+    return update.bind(null, this, CANCEL)
   }
 
   get clear() {
     return () => {
-      this.meta.status = 'start'
+      this.meta.status = START
       this.payload = null
       this._observers.forEach(observer => observer.unsubscribe())
     }
@@ -162,10 +163,7 @@ export class Subject extends Observable {
 function fromPromise(promise: Promise<*>): Subject {
   let subject = new Subject()
 
-  promise
-    .then(subject.next)
-    .then(() => subject.complete())
-    .catch(subject.error)
+  promise.then(subject.complete).catch(subject.error)
 
   return subject
 }
@@ -181,6 +179,12 @@ function update(subject: Subject, status: string, value: *): void {
     return
   }
 
+  // Allow passing a value to complete. This is a decent compromise
+  // For actions that simply complete
+  if (status === COMPLETE && arguments.length > 2) {
+    update(subject, NEXT, value)
+  }
+
   subject.meta.status = status
 
   if (arguments.length > 2) {
@@ -189,13 +193,13 @@ function update(subject: Subject, status: string, value: *): void {
 
   subject._observers.forEach(observer => {
     switch (status) {
-      case 'next':
+      case NEXT:
         return observer.next(subject.payload)
-      case 'error':
+      case ERROR:
         return observer.error(subject.payload)
-      case 'complete':
+      case COMPLETE:
         return observer.complete()
-      case 'cancel':
+      case CANCEL:
         return observer.cancel()
     }
   })


Test patch:
diff --git a/packages/microcosm/test/integration/efficiency.test.js b/packages/microcosm/test/integration/efficiency.test.js
index 942f9b4f..7fe314ed 100644
--- a/packages/microcosm/test/integration/efficiency.test.js
+++ b/packages/microcosm/test/integration/efficiency.test.js
@@ -61,7 +61,7 @@ describe('Efficiency', function() {
     expect(handler).toHaveBeenCalledTimes(1)
   })
 
-  it.skip('actions only dispatch duplicatively to address races', () => {
+  it('actions only dispatch duplicatively to address races', () => {
     const repo = new Microcosm()
     const handler = jest.fn()
     const action = () => () => {}
@@ -80,6 +80,6 @@ describe('Efficiency', function() {
     two.complete(1)
     one.complete(2)
 
-    expect(handler).toHaveBeenCalledTimes(2)
+    expect(handler).toHaveBeenCalledTimes(3)
   })
 })
diff --git a/packages/microcosm/test/integration/reexecuting-domain-handlers.test.js b/packages/microcosm/test/integration/reexecuting-domain-handlers.test.js
index 08440d35..5734b38b 100644
--- a/packages/microcosm/test/integration/reexecuting-domain-handlers.test.js
+++ b/packages/microcosm/test/integration/reexecuting-domain-handlers.test.js
@@ -1,7 +1,7 @@
 import { Microcosm } from 'microcosm'
 
 describe('Re-executing domain handlers', function() {
-  it.skip('does not re-invoke a domain handler if it would result in no change', function() {
+  it('does not re-invoke a domain handler if it would result in no change', function() {
     let repo = new Microcosm()
 
     let counterOneCalls = 0
diff --git a/packages/microcosm/test/unit/history/updater.test.js b/packages/microcosm/test/unit/history/updater.test.js
index b4113fc5..94ec05ec 100644
--- a/packages/microcosm/test/unit/history/updater.test.js
+++ b/packages/microcosm/test/unit/history/updater.test.js
@@ -13,6 +13,8 @@ describe('History updater', function() {
     await repo.push('one', 1)
     await repo.push('two', 2)
 
-    expect(spy).toHaveBeenCalledTimes(2)
+    // Twice for next
+    // Twice for complete
+    expect(spy).toHaveBeenCalledTimes(4)
   })
 })
diff --git a/packages/microcosm/test/unit/observable.test.js b/packages/microcosm/test/unit/observable.test.js
index a3f12efb..dfe3578f 100644
--- a/packages/microcosm/test/unit/observable.test.js
+++ b/packages/microcosm/test/unit/observable.test.js
@@ -309,4 +309,65 @@ describe('Observable', function() {
       )
     })
   })
+
+  describe('::filter', () => {
+    it('operates like array.filter over a stream of values', () => {
+      let isOdd = n => n % 2
+      let double = Observable.of(0, 1, 2, 3, 4).filter(isOdd)
+      let answers = []
+
+      // Filter is lazy. It shouldn't trigger an iteration until
+      // subscribe is invoked
+      expect(answers).toEqual([])
+
+      double.subscribe(value => answers.push(value))
+      scheduler().flush()
+
+      expect(answers).toEqual([1, 3])
+    })
+
+    it('can configure scope', () => {
+      let Maths = {
+        threshold: 2,
+        greaterThan(value) {
+          return value > this.threshold
+        }
+      }
+
+      let greaterThan2 = Observable.of(2, 4, 6, 8).filter(
+        Maths.greaterThan,
+        Maths
+      )
+      let answers = []
+
+      greaterThan2.subscribe(value => answers.push(value))
+
+      scheduler().flush()
+
+      expect(answers).toEqual([4, 6, 8])
+    })
+
+    it('passes along errors', () => {
+      let observable = new Observable(observer => {
+        for (var i = 0; i < 5; i++) {
+          observer.next(i)
+        }
+        throw 'Exceptional!'
+      })
+
+      let error = jest.fn()
+
+      observable.filter(n => n).subscribe({ error })
+
+      scheduler().flush()
+
+      expect(error).toHaveBeenCalledWith('Exceptional!')
+    })
+
+    it('raises if not given a function', () => {
+      expect(() => Observable.of(0, 1, 2, 3, 4).filter()).toThrow(
+        'undefined is not a function'
+      )
+    })
+  })
 })