feat(state$): state$ is now an Observable not a Subject

package.json

@@ -15,7 +15,7 @@
     "build:tests": "rimraf temp && babel test -d temp",
     "clean": "rimraf lib temp dist",
     "check": "npm run lint && npm run test",
-    "test": "npm run lint && npm run build && npm run build:tests && mocha temp && npm run test:typings",
+    "test": "npm run build:tests && mocha temp",
     "test:typings": "tsc --strict index.d.ts test/typings.ts --outDir temp --target ES5 --moduleResolution node --lib dom,es2015 && cd temp && node typings.js",
     "shipit": "npm run clean && npm run build && npm run lint && npm test && scripts/preprepublish.sh && npm publish",
     "docs:clean": "rimraf _book",

src/StateObservable.js

@@ -0,0 +1,71 @@
+import { Observable } from 'rxjs';
+
+// Used as a placeholder value so we can tell
+// whether or not the real state has been set,
+// even if the real state was `undefined`
+export const UNSET_STATE_VALUE = {};
+
+// We can't use BehaviorSubject because when we create
+// and give state$ to your epics your reducers have not
+// yet run, so there is no initial state yet until the
+// `@@redux/INIT` action is emitted. BehaviorSubject
+// requires an initial value (which would be undefined)
+// and if an epic subscribes to it on app boot it would
+// synchronously emit that undefined value incorrectly.
+// We also don't want to expose a Subject to the user at
+// all because then they could do `state$.next()` and
+// screw things up and it would just be a footgun.
+export class StateObservable extends Observable {
+  constructor(stateSubject, store) {
+    super();
+    this.source = stateSubject;
+    // 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.source.subscribe(value => {
+      this.__value = 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') {
+        require('./utils/console').warn('You accessed state$.value inside one of your Epics, before your reducers have run for the first time, so there is no state yet. You\'ll need to wait until after the first action (@@redux/INIT) is dispatched or by using state$ as an Observable.');
+      }
+      return undefined;
+    } else {
+      return this.__value;
+    }
+  }
+
+  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');
+    }
+    return this.__value;
+  }
+
+  dispatch(...args) {
+    if (process.env.NODE_ENV !== 'production') {
+      require('./utils/console').deprecate('calling store.dispatch() directly 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. Instead of calling store.dispatch() in your Epic, emit actions through the Observable your Epic returns.\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');
+    }
+    return this.__store.dispatch(...args);
+
+  }
+}

src/createEpicMiddleware.js

@@ -1,7 +1,7 @@
-import { Subject } from 'rxjs';
+import { Subject, BehaviorSubject } from 'rxjs';
 import { map, switchMap } from 'rxjs/operators';
 import { ActionsObservable } from './ActionsObservable';
-import { StateSubject } from './StateSubject';
+import { StateObservable } from './StateObservable';
 import { EPIC_END } from './EPIC_END';
 
 const defaultAdapter = {
@@ -34,7 +34,8 @@ export function createEpicMiddleware(rootEpic, options = defaultOptions) {
       // https://github.com/redux-observable/redux-observable/issues/389
       require('./utils/console').warn('this middleware is already associated with a store. createEpicMiddleware should be called for every store.\n\n See https://goo.gl/2GQ7Da');
     }
-    const state$ = new StateSubject(_store);
+    const stateInput$ = new Subject();
+    const state$ = new StateObservable(stateInput$, _store);
     store = _store;
 
     return next => {
@@ -67,7 +68,7 @@ export function createEpicMiddleware(rootEpic, options = defaultOptions) {
 
         // It's important to update the state$ before we emit
         // the action because otherwise it would be stale!
-        state$.next(store.getState());
+        stateInput$.next(store.getState());
         input$.next(action);
 
         return result;

src/index.js

@@ -1,6 +1,6 @@
 export { createEpicMiddleware } from './createEpicMiddleware';
 export { ActionsObservable } from './ActionsObservable';
-export { StateSubject } from './StateSubject';
+export { StateObservable } from './StateObservable';
 export { combineEpics } from './combineEpics';
 export { EPIC_END } from './EPIC_END';
 export { ofType } from './operators';

src/utils/console.js

@@ -1,12 +1,15 @@
 const deprecationsSeen = {};
-
-export const warn = (typeof console === 'object' && typeof console.warn === 'function')
-  ? msg => console.warn(msg)
-  : () => {};
+const consoleWarn = (typeof console === 'object' && typeof console.warn === 'function')
+  ? (...args) => console.warn(...args)
+  : () => { };
 
 export const deprecate = msg => {
   if (!deprecationsSeen[msg]) {
     deprecationsSeen[msg] = true;
-    warn(`redux-observable | DEPRECATION: ${msg}`);
+    consoleWarn(`redux-observable | DEPRECATION: ${msg}`);
   }
 };
+
+export const warn = msg => {
+  consoleWarn(`redux-observable | WARNING: ${msg}`);
+};

test/StateObservable-spec.js

@@ -0,0 +1,37 @@
+/* globals describe it */
+import { expect } from 'chai';
+import { StateObservable } from '../';
+import { Observable, Subject } from 'rxjs';
+
+describe('StateObservable', () => {
+  it('should exist', () => {
+    expect(StateObservable.prototype).to.be.instanceof(Observable);
+  });
+
+  it('should mirror the source subject', () => {
+    const input$ = new Subject();
+    const state$ = new StateObservable(input$);
+    let result = null;
+
+    state$.subscribe(state => {
+      result = state;
+    });
+
+    expect(result).to.equal(null);
+    input$.next('first');
+    expect(result).to.equal('first');
+    input$.next('second');
+    expect(result).to.equal('second');
+  });
+
+  it('should cache last state on the `value` property', () => {
+    const input$ = new Subject();
+    const state$ = new StateObservable(input$);
+
+    expect(state$.value).to.equal(undefined);
+    input$.next('first');
+    expect(state$.value).to.equal('first');
+    input$.next('second');
+    expect(state$.value).to.equal('second');
+  });
+});

test/createEpicMiddleware-spec.js

@@ -3,9 +3,9 @@ import 'babel-polyfill';
 import { expect } from 'chai';
 import sinon from 'sinon';
 import { createStore, applyMiddleware } from 'redux';
-import { createEpicMiddleware, combineEpics, ActionsObservable, EPIC_END, ofType } from '../';
+import { createEpicMiddleware, combineEpics, ActionsObservable, StateObservable, EPIC_END, ofType } from '../';
 import { of, empty, merge } from 'rxjs';
-import { mapTo, map, ignoreElements } from 'rxjs/operators';
+import { mapTo, map, ignoreElements, distinctUntilChanged } from 'rxjs/operators';
 
 describe('createEpicMiddleware', () => {
   it('should provide epics a stream of action$ and a stream of state$', (done) => {
@@ -15,7 +15,7 @@ describe('createEpicMiddleware', () => {
     const mockMiddleware = store => next => action => {
       expect(epic.calledOnce).to.equal(true);
       expect(epic.firstCall.args[0]).to.be.instanceOf(ActionsObservable);
-      expect(epic.firstCall.args[1]).to.be.instanceof(StateSubject);
+      expect(epic.firstCall.args[1]).to.be.instanceof(StateObservable);
       done();
     };
     const store = createStore(reducer, applyMiddleware(epicMiddleware, mockMiddleware));
@@ -42,12 +42,22 @@ describe('createEpicMiddleware', () => {
   it('should warn about improper use of dispatch function', () => {
     sinon.spy(console, 'warn');
     const reducer = (state = [], action) => state.concat(action);
-    const epic = (action$, state$) => action$.pipe(
+    const epic = (action$, store) => action$.pipe(
       ofType('PING'),
-      map(() => state$.dispatch({ type: 'PONG' })),
+      map(() => store.dispatch({ type: 'PONG' })),
       ignoreElements()
     );
 
+    const middleware = createEpicMiddleware(epic);
+    const store = createStore(reducer, applyMiddleware(middleware));
+
+    store.dispatch({ type: 'PING' });
+
+    expect(console.warn.callCount).to.equal(1);
+    expect(console.warn.getCall(0).args[0]).to.equal('redux-observable | DEPRECATION: calling store.dispatch() directly 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. Instead of calling store.dispatch() in your Epic, emit actions through the Observable your Epic returns.\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');
+    console.warn.restore();
+  });
+
   it('should update state$ after an action goes through reducers but before epics', () => {
     const actions = [];
     const reducer = (state = 0, action) => {
@@ -60,15 +70,16 @@ describe('createEpicMiddleware', () => {
       }
     };
     const epic = (action$, state$) =>
-      mergeStatic(
-        action$.ofType('PING'),
-        state$::distinctUntilChanged()
-      )
-        ::map(input => ({
+      merge(
+        action$.pipe(ofType('PING')),
+        state$.pipe(distinctUntilChanged())
+      ).pipe(
+        map(input => ({
           type: 'PONG',
           state: state$.value,
           input
-        }));
+        }))
+      );
 
     const epicMiddleware = createEpicMiddleware(epic);
     const store = createStore(reducer, applyMiddleware(epicMiddleware));
@@ -115,7 +126,7 @@ describe('createEpicMiddleware', () => {
     store.dispatch({ type: 'PING' });
 
     expect(console.warn.callCount).to.equal(1);
-    expect(console.warn.getCall(0).args[0]).to.equal('You accessed state$.value inside one of your Epics, before your reducers have run for the first time, so there is no state yet. You\'ll need to wait until after the first action (@@redux/INIT) is dispatched or by using state$ as an Observable.');
+    expect(console.warn.getCall(0).args[0]).to.equal('redux-observable | WARNING: You accessed state$.value inside one of your Epics, before your reducers have run for the first time, so there is no state yet. You\'ll need to wait until after the first action (@@redux/INIT) is dispatched or by using state$ as an Observable.');
     expect(store.getState()).to.deep.equal([{
       type: '@@redux/INIT'
     }, {
@@ -134,20 +145,22 @@ describe('createEpicMiddleware', () => {
       actions.push(action);
       return state;
     };
-    const epic = (action$, state$) => action$
-      .ofType('PING')
-      ::map(() => ({
-        type: 'RESULT',
-        state: state$.getState()
-      }));
+    const epic = (action$, state$) =>
+      action$.pipe(
+        ofType('PING'),
+        map(() => ({
+          type: 'RESULT',
+          state: state$.getState()
+        }))
+      );
 
     const middleware = createEpicMiddleware(epic);
     const store = createStore(reducer, applyMiddleware(middleware));
 
     store.dispatch({ type: 'PING' });
 
     expect(console.warn.callCount).to.equal(1);
-    expect(console.warn.getCall(0).args[0]).to.equal('redux-observable | DEPRECATION: 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 StateSubject), 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$: StateSubject<S>, dependencies?: D): Observable<R>\n\nLearn more: https://goo.gl/WWNYSP');
+    expect(console.warn.getCall(0).args[0]).to.equal('redux-observable | DEPRECATION: 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');
     expect(actions[actions.length - 1].state).to.equal(store.getState());
     console.warn.restore();
   });
@@ -181,7 +194,7 @@ describe('createEpicMiddleware', () => {
     ]);
   });
 
-  it('should not swallow exceptions thrown by reducers', () => {
+  it('exceptions thrown in reducers as part of an epic-dispatched action should go through HostReportErrors', (done) => {
     const reducer = (state = [], action) => {
       switch (action.type) {
         case 'ACTION_1':
@@ -203,10 +216,16 @@ describe('createEpicMiddleware', () => {
       );
     const middleware = createEpicMiddleware(epic);
     const store = createStore(reducer, applyMiddleware(middleware));
+    process.prependOnceListener('uncaughtException', function (err) {
+      expect(err.message).to.equal('some error');
+      done();
+    });
 
+    // rxjs v6 does not rethrow synchronously instead emitting on
+    // HostReportErrors e.g. window.onerror or process.on('uncaughtException')
     expect(() => {
       store.dispatch({ type: 'FIRE_1' });
-    }).to.throw('some error');
+    }).to.not.throw('some error');
   });
 
   it('should throw if you don\'t provide a rootEpic', () => {
@@ -219,16 +238,17 @@ describe('createEpicMiddleware', () => {
     }).to.throw('You must provide a root Epic to createEpicMiddleware');
   });
 
-  it('should throw if you provide a root epic that doesn\'t return anything', () => {
+  it('should throw if you provide a root epic that doesn\'t return anything', (done) => {
     sinon.spy(console, 'error');
 
     const rootEpic = () => {};
     const epicMiddleware = createEpicMiddleware(rootEpic);
     createStore(() => {}, applyMiddleware(epicMiddleware));
 
-    expect(console.error.callCount).to.equal(1);
-    expect(console.error.getCall(0).args[0]).to.equal('Your root Epic "rootEpic" does not return a stream. Double check you\'re not missing a return statement!');
-    console.error.restore();
+    process.prependOnceListener('uncaughtException', function (err) {
+      expect(err.message).to.equal('Your root Epic "rootEpic" does not return a stream. Double check you\'re not missing a return statement!');
+      done();
+    });
   });
 
   it('should allow you to replace the root epic with middleware.replaceEpic(epic)', () => {