feat(state$): state$ now only emits subsequent values if the state sh…

…allowly is different (e.g. prevValue !== nextValue). It still emits the current state immediately on subscribe regardless, as it did before, similar to BehaviorSubject. Closes #497

src/StateObservable.js

@@ -1,4 +1,4 @@
-import { Observable } from 'rxjs';
+import { Observable, Subject } from 'rxjs';
 
 // Used as a placeholder value so we can tell
 // whether or not the real state has been set,
@@ -17,26 +17,32 @@ export const UNSET_STATE_VALUE = {};
 // screw things up and it would just be a footgun.
 export class StateObservable extends Observable {
   constructor(stateSubject, store) {
-    super();
-    this.source = stateSubject;
+    super(subscriber => {
+      const subscription = this.__notifier.subscribe(subscriber);
+      if (this.__value !== UNSET_STATE_VALUE && subscription && !subscription.closed) {
+        subscriber.next(this.__value);
+      }
+      return subscription;
+    });
+
     // If you're reading this, keep in mind that this is
     // NOT part of the public API and will be removed!
     this.__store = store;
     this.__value = UNSET_STATE_VALUE;
+    this.__notifier = new Subject();
 
-    this.source.subscribe(value => {
-      this.__value = value;
+    this.__subscription = stateSubject.subscribe(value => {
+      // We only want to update state$ if it has actually changed since
+      // redux requires reducers use immutability patterns.
+      // This is basically what distinctUntilChanged() does but it's so simple
+      // we don't need to pull that code in
+      if (value !== this.__value) {
+        this.__value = value;
+        this.__notifier.next(value);
+      }
     });
   }
 
-  _subscribe(subscriber) {
-    const subscription = super._subscribe(subscriber);
-    if (this.__value !== UNSET_STATE_VALUE && subscription && !subscription.closed) {
-      subscriber.next(this.__value);
-    }
-    return subscription;
-  }
-
   get value() {
     if (this.__value === UNSET_STATE_VALUE) {
       if (process.env.NODE_ENV !== 'production') {
@@ -48,12 +54,6 @@ export class StateObservable extends Observable {
     }
   }
 
-  lift(operator) {
-    const observable = new StateObservable(this);
-    observable.operator = operator;
-    return observable;
-  }
-
   getState() {
     if (process.env.NODE_ENV !== 'production') {
       require('./utils/console').deprecate('calling store.getState() in your Epics is deprecated and will be removed. The second argument to your Epic is now a stream of state$ (a StateObservable), instead of the store. To imperatively get the current state use state$.value instead of getState(). Alternatively, since it\'s now a stream you can compose and react to state changes.\n\n  function <T, R, S, D>(action$: ActionsObservable<T>, state$: StateObservable<S>, dependencies?: D): Observable<R>\n\nLearn more: https://redux-observable.js.org/MIGRATION.html');

test/StateObservable-spec.js

@@ -1,9 +1,20 @@
-/* globals describe it */
+/* globals describe it beforeEach afterEach */
 import { expect } from 'chai';
+import sinon from 'sinon';
 import { StateObservable } from '../';
 import { Observable, Subject } from 'rxjs';
 
 describe('StateObservable', () => {
+  let spySandbox;
+
+  beforeEach(() => {
+    spySandbox = sinon.sandbox.create();
+  });
+
+  afterEach(() => {
+    spySandbox.restore();
+  });
+
   it('should exist', () => {
     expect(StateObservable.prototype).to.be.instanceof(Observable);
   });
@@ -34,4 +45,35 @@ describe('StateObservable', () => {
     input$.next('second');
     expect(state$.value).to.equal('second');
   });
+
+  it('should only update when the next value shallowly differs', () => {
+    const input$ = new Subject();
+    const state$ = new StateObservable(input$);
+    const next = spySandbox.spy();
+    state$.subscribe(next);
+
+    expect(state$.value).to.equal(undefined);
+    expect(next.callCount).to.equal(0);
+
+    const first = { value: 'first' };
+    input$.next(first);
+    expect(state$.value).to.equal(first);
+    expect(next.callCount).to.equal(1);
+    expect(next.getCall(0).args).to.deep.equal([first]);
+
+    input$.next(first);
+    expect(state$.value).to.equal(first);
+    expect(next.callCount).to.equal(1);
+
+    first.value = 'something else';
+    input$.next(first);
+    expect(state$.value).to.equal(first);
+    expect(next.callCount).to.equal(1);
+
+    const second = { value: 'second' };
+    input$.next(second);
+    expect(state$.value).to.equal(second);
+    expect(next.callCount).to.equal(2);
+    expect(next.getCall(1).args).to.deep.equal([second]);
+  });
 });