feat(ofType): expose ofType as lettable operator (#343)

Supports using ofType directly in lettable operator pipelines.

Closes #186

docs/basics/Epics.md

@@ -98,6 +98,20 @@ const pingEpic = action$ =>
     .mapTo({ type: 'PONG' });
 ```
 
+You can also use the `ofType()` operator directly as a [lettable operator](https://github.com/ReactiveX/rxjs/blob/master/doc/lettable-operators.md):
+
+```js
+import { ofType } from 'redux-observable';
+import { delay, mapTo } from 'rxjs/operators'; // rxjs v5.5+
+
+const pingEpic = action$ =>
+  action$.pipe(
+    ofType('PING'),
+    delay(1000), // Asynchronously wait 1000ms then continue
+    mapTo({ type: 'PONG' })
+  );
+```
+
 ***
 
 ### Try It Live!

index.d.ts

@@ -51,4 +51,6 @@ export declare function createEpicMiddleware<T extends Action, S, D = any>(rootE
 export declare function combineEpics<T extends Action, S, D = any>(...epics: Epic<T, S, D>[]): Epic<T, S, D>;
 export declare function combineEpics<E>(...epics: E[]): E;
 
+export declare function ofType<T extends Action>(...keys: T['type'][]): (source: Observable<T>) => Observable<T>;
+
 export declare const EPIC_END: '@@redux-observable/EPIC_END';

package.json

@@ -97,7 +97,7 @@
     "mocha": "^3.0.1",
     "redux": "^3.5.2",
     "rimraf": "^2.5.4",
-    "rxjs": "^5.0.0",
+    "rxjs": "^5.5.0",
     "sinon": "^2.3.3",
     "typescript": "^2.1.4",
     "webpack": "^2.2.1",

src/ActionsObservable.js

@@ -1,7 +1,8 @@
 import { Observable } from 'rxjs/Observable';
 import { of } from 'rxjs/observable/of';
 import { from } from 'rxjs/observable/from';
-import { filter } from 'rxjs/operator/filter';
+import { letProto } from 'rxjs/operator/let';
+import { ofType } from './operators';
 
 export class ActionsObservable extends Observable {
   static of(...actions) {
@@ -24,18 +25,6 @@ export class ActionsObservable extends Observable {
   }
 
   ofType(...keys) {
-    return this::filter(({ type }) => {
-      const len = keys.length;
-      if (len === 1) {
-        return type === keys[0];
-      } else {
-        for (let i = 0; i < len; i++) {
-          if (keys[i] === type) {
-            return true;
-          }
-        }
-      }
-      return false;
-    });
+    return this::letProto(ofType(...keys));
   }
 }

src/index.js

@@ -2,3 +2,4 @@ export { createEpicMiddleware } from './createEpicMiddleware';
 export { ActionsObservable } from './ActionsObservable';
 export { combineEpics } from './combineEpics';
 export { EPIC_END } from './EPIC_END';
+export { ofType } from './operators';

src/operators.js

@@ -0,0 +1,19 @@
+import { filter } from 'rxjs/operator/filter';
+
+export function ofType(...keys) {
+  return function ofTypeOperatorFunction(source) {
+    return source::filter(({ type }) => {
+      const len = keys.length;
+      if (len === 1) {
+        return type === keys[0];
+      } else {
+        for (let i = 0; i < len; i++) {
+          if (keys[i] === type) {
+            return true;
+          }
+        }
+      }
+      return false;
+    });
+  };
+}

test/operators-spec.js

@@ -0,0 +1,56 @@
+/* globals describe it */
+import { expect } from 'chai';
+import { Subject } from 'rxjs/Subject';
+import { ofType } from '../';
+
+describe('operators', () => {
+  describe('ofType', () => {
+    it('should filter by action type', () => {
+      let actions = new Subject();
+      let lulz = [];
+      let haha = [];
+
+      actions.pipe(ofType('LULZ')).subscribe(x => lulz.push(x));
+      actions.pipe(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 }]);
+    });
+
+    it('should filter by multiple action types', () => {
+      let actions = new Subject();
+      let lulz = [];
+      let haha = [];
+
+      actions.pipe(ofType('LULZ', 'LARF')).subscribe(x => lulz.push(x));
+      actions.pipe(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: 'LARF', i: 1 });
+
+      expect(lulz).to.deep.equal([{ type: 'LULZ', i: 0 }, { type: 'LARF', i: 1 }]);
+      expect(haha).to.deep.equal([]);
+
+      actions.next({ type: 'HAHA', i: 0 });
+
+      expect(lulz).to.deep.equal([{ type: 'LULZ', i: 0 }, { type: 'LARF', i: 1 }]);
+      expect(haha).to.deep.equal([{ type: 'HAHA', i: 0 }]);
+    });
+  });
+});

test/typings.ts

@@ -2,14 +2,15 @@ import { expect } from 'chai';
 import { createStore, applyMiddleware, MiddlewareAPI, Action } from 'redux';
 import { Observable } from 'rxjs/Observable';
 import { ajax } from 'rxjs/observable/dom/ajax';
+import { map } from 'rxjs/operators/map';
 import { asap } from 'rxjs/scheduler/asap';
 import 'rxjs/add/observable/of';
 import 'rxjs/add/operator/mapTo';
 import 'rxjs/add/operator/map';
 import 'rxjs/add/operator/mergeMap';
 
 import { createEpicMiddleware, Epic, combineEpics,
-  EpicMiddleware, ActionsObservable } from '../';
+  EpicMiddleware, ActionsObservable, ofType } from '../';
 
 interface State {
   foo: string
@@ -70,8 +71,17 @@ const epic7: Epic<FluxStandardAction, State, Dependencies> = (action$, store, de
       payload: dependencies.func(payload)
     }));
 
-const rootEpic1: Epic<FluxStandardAction, State> = combineEpics<FluxStandardAction, State>(epic1, epic2, epic3, epic4, epic5, epic6, epic7);
-const rootEpic2 = combineEpics(epic1, epic2, epic3, epic4, epic5, epic6, epic7);
+const epic8: Epic<FluxStandardAction, State, Dependencies> = (action$, store, dependencies) =>
+    action$.pipe(
+      ofType('EIGHTH'),
+      map(({ type, payload }) => ({
+        type: 'eighth',
+        payload
+      }))
+    )
+
+const rootEpic1: Epic<FluxStandardAction, State> = combineEpics<FluxStandardAction, State>(epic1, epic2, epic3, epic4, epic5, epic6, epic7, epic8);
+const rootEpic2 = combineEpics(epic1, epic2, epic3, epic4, epic5, epic6, epic7, epic8);
 
 const dependencies: Dependencies = {
   func(value: string) { return `func-${value}`}
@@ -118,6 +128,7 @@ store.dispatch({ type: 'SECOND' });
 store.dispatch({ type: 'FIFTH', payload: 'fifth-payload' });
 store.dispatch({ type: 'SIXTH', payload: 'sixth-payload' });
 store.dispatch({ type: 'SEVENTH', payload: 'seventh-payload' });
+store.dispatch({ type: 'EIGHTH', payload: 'eighth-payload' });
 
 expect(store.getState()).to.deep.equal([
   { "type": "@@redux/INIT" },
@@ -157,7 +168,10 @@ expect(store.getState()).to.deep.equal([
   { "type": "sixth", "payload": "sixth-payload" },
   { "type": "SEVENTH", "payload": "seventh-payload" },
   { "type": "seventh", "payload": "func-seventh-payload" },
-  { "type": "seventh", "payload": "func-seventh-payload" }
+  { "type": "seventh", "payload": "func-seventh-payload" },
+  { "type": "EIGHTH", "payload": "eighth-payload" },
+  { "type": "eighth", "payload": "eighth-payload" },
+  { "type": "eighth", "payload": "eighth-payload" }
 ]);
 
 const input$ = Observable.create(() => {});