feat(ofType): add operator to provided actions observable

This adds an `ofType` operator to the actions observable provided to the dispatched function that will be available anywhere
in the operator chain for that observable. It basically filters by the action's type, which is a common need.

src/actions-observable.js

@@ -0,0 +1,19 @@
+import { Observable } from 'rxjs/Observable';
+import { filter } from 'rxjs/operator/filter';
+
+export class ActionsObservable extends Observable {
+  constructor(actionsSubject) {
+    super();
+    this.source = actionsSubject;
+  }
+
+  lift(operator) {
+    const observable = new ActionsObservable(this);
+    observable.operator = operator;
+    return observable;
+  }
+
+  ofType(key) {
+    return this::filter((action) => action.type === key);
+  }
+}

src/index.js

@@ -1,13 +1,15 @@
 import { Subject } from 'rxjs/Subject';
 import { from } from 'rxjs/observable/from';
+import { ActionsObservable as Actions } from './actions-observable';
 
 export function reduxObservable() {
   let actions = new Subject();
+  let actionsObs = new Actions(actions);
 
   let middleware = (store) => (next) => {
     return (action) => {
       if (typeof action === 'function') {
-        let obs = from(action(actions, store));
+        let obs = from(action(actionsObs, store));
         let sub = obs.subscribe(next);
         return sub;
       } else {
@@ -19,3 +21,5 @@ export function reduxObservable() {
 
   return middleware;
 }
+
+export const ActionsObservable = Actions;

test/actions-observable-spec.js

@@ -0,0 +1,51 @@
+/* globals describe it */
+import { expect } from 'chai';
+import { ActionsObservable, reduxObservable } from '../';
+import { createStore, applyMiddleware } from 'redux';
+import { of } from 'rxjs/observable/of';
+import { Subject } from 'rxjs/Subject';
+
+describe('ActionsObservable', () => {
+  it('should exist', () => {
+    expect(ActionsObservable).to.be.a('function');
+  });
+
+  it('should be the type provided to a dispatched function', () => {
+    let middleware = reduxObservable();
+    let reducer = (state, action) => {
+      return state;
+    };
+
+    let store = createStore(reducer, applyMiddleware(middleware));
+
+    store.dispatch((arg1) => {
+      expect(arg1).to.be.an.instanceof(ActionsObservable);
+      return of({ type: 'WEEE' });
+    });
+  });
+
+  it('should have a ofType operator that filters by action type', () => {
+    let actions = new Subject();
+    let actionsObs = new ActionsObservable(actions);
+    let lulz = [];
+    let haha = [];
+
+    actionsObs.ofType('LULZ').subscribe(x => lulz.push(x));
+    actionsObs.ofType('HAHA').subscribe(x => haha.push(x));
+
+    actions.next({ type: 'LULZ', i: 0 });
+
+    expect(lulz).to.deep.equal([{ type: 'LULZ', i: 0 }]);
+    expect(haha).to.deep.equal([]);
+
+    actions.next({ type: 'LULZ', i: 1 });
+
+    expect(lulz).to.deep.equal([{ type: 'LULZ', i: 0 }, { type: 'LULZ', i: 1 }]);
+    expect(haha).to.deep.equal([]);
+
+    actions.next({ type: 'HAHA', i: 0 });
+
+    expect(lulz).to.deep.equal([{ type: 'LULZ', i: 0 }, { type: 'LULZ', i: 1 }]);
+    expect(haha).to.deep.equal([{ type: 'HAHA', i: 0 }]);
+  });
+});