Skip to content

Commit

Permalink
Browse files Browse the repository at this point in the history
feat(replaceEpic): Dispatches an EPIC_END action when you replaceEpic…
…() (#75)

so that you can listen for it and emit any state cleanup
  • Loading branch information
jayphelps committed Jul 24, 2016
1 parent a26f429 commit fef6f80
Show file tree
Hide file tree
Showing 7 changed files with 85 additions and 51 deletions.
2 changes: 1 addition & 1 deletion docs/basics/Epics.md
@@ -1,7 +1,7 @@
# Epics

>##### Not familiar with Observables/RxJS v5?
> redux-observable requires an understanding of Observables with RS v5. If you're new to Reactive Programming with S v5, head over to [http://reactivex.io/rxjs/](http://reactivex.io/rxjs/) to familiarize yourself first.
> redux-observable requires an understanding of Observables with RxJS v5. If you're new to Reactive Programming with RxJS v5, head over to [http://reactivex.io/rxjs/](http://reactivex.io/rxjs/) to familiarize yourself first.
An **Epic** is the core primitive of redux-observable.

Expand Down
1 change: 0 additions & 1 deletion package.json
Expand Up @@ -87,7 +87,6 @@
"gitbook-plugin-theme-default": "1.0.4",
"json-server": "^0.8.14",
"mocha": "^2.4.5",
"promise": "^7.1.1",
"redux": "^3.5.1",
"rimraf": "^2.5.2",
"rxjs": "^5.0.0-beta.6",
Expand Down
1 change: 1 addition & 0 deletions src/EPIC_END.js
@@ -0,0 +1 @@
export const EPIC_END = '@@redux-observable/EPIC_END';
41 changes: 25 additions & 16 deletions src/createEpicMiddleware.js
@@ -1,35 +1,44 @@
import { Subject } from 'rxjs/Subject';
import { BehaviorSubject } from 'rxjs/BehaviorSubject';
import { from } from 'rxjs/observable/from';
import { switchMap } from 'rxjs/operator/switchMap';
import { ActionsObservable } from './ActionsObservable';
import { EPIC_END } from './EPIC_END';

export function createEpicMiddleware(epic) {
const actionsSubject = new Subject();
const action$ = new ActionsObservable(actionsSubject);
const epic$ = new BehaviorSubject(epic);
let store;

const epicMiddleware = store => next => {
epic$.filter(epic => typeof epic === 'function')
.switchMap(epic => epic(action$, store))
.subscribe(store.dispatch);
const epicMiddleware = _store => {
store = _store;

return action => {
if (typeof action === 'function') {
if (typeof console !== 'undefined' && typeof console.warn !== 'undefined') {
console.warn('DEPRECATION: Using thunkservables with redux-observable is now deprecated in favor of the new "Epics" feature. See http://redux-observable.js.org/docs/FAQ.html#why-were-thunkservables-deprecated');
}

const out$ = from(action(action$, store));
return out$.subscribe(store.dispatch);
} else {
const result = next(action);
actionsSubject.next(action);
return result;
return next => {
if (typeof epic === 'function') {
epic$::switchMap(epic => epic(action$, store))
.subscribe(store.dispatch);
}

return action => {
if (typeof action === 'function') {
if (typeof console !== 'undefined' && typeof console.warn !== 'undefined') {
console.warn('DEPRECATION: Using thunkservables with redux-observable is now deprecated in favor of the new "Epics" feature. See http://redux-observable.js.org/docs/FAQ.html#why-were-thunkservables-deprecated');
}

const out$ = from(action(action$, store));
return out$.subscribe(store.dispatch);
} else {
const result = next(action);
actionsSubject.next(action);
return result;
}
};
};
};

epicMiddleware.replaceEpic = epic => {
store.dispatch({ type: EPIC_END });
epic$.next(epic);
};

Expand Down
1 change: 1 addition & 0 deletions src/index.js
@@ -1,3 +1,4 @@
export { createEpicMiddleware } from './createEpicMiddleware';
export { ActionsObservable } from './ActionsObservable';
export { combineEpics } from './combineEpics';
export { EPIC_END } from './EPIC_END';
73 changes: 42 additions & 31 deletions test/createEpicMiddleware-spec.js
@@ -1,22 +1,30 @@
/* globals describe it */
import 'babel-polyfill';
import { expect } from 'chai';
import sinon from 'sinon';
import { createStore, applyMiddleware } from 'redux';
import { createEpicMiddleware } from '../';
import * as Rx from 'rxjs';
import Promise from 'promise';
import 'babel-polyfill';
import { createEpicMiddleware, EPIC_END } from '../';
import $$observable from 'symbol-observable';

const { Observable } = Rx;
// We need to import the operators separately and not add them to the Observable
// prototype, otherwise we might accidentally cover-up that the source we're
// testing uses an operator that it does not import!
import { of } from 'rxjs/observable/of';
import { mergeStatic, merge } from 'rxjs/operator/merge';
import { mapTo } from 'rxjs/operator/mapTo';
import { delay } from 'rxjs/operator/delay';
import { takeUntil } from 'rxjs/operator/takeUntil';
import { take } from 'rxjs/operator/take';
import { filter } from 'rxjs/operator/filter';
import { map } from 'rxjs/operator/map';
import { startWith } from 'rxjs/operator/startWith';

describe('createEpicMiddleware', () => {
it('should accept a epic argument that wires up a stream of actions to a stream of actions', () => {
const reducer = (state = [], action) => state.concat(action);
const epic = (action$, store) =>
Observable.merge(
action$.ofType('FIRE_1').mapTo({ type: 'ACTION_1' }),
action$.ofType('FIRE_2').mapTo({ type: 'ACTION_2' })
mergeStatic(
action$.ofType('FIRE_1')::mapTo({ type: 'ACTION_1' }),
action$.ofType('FIRE_2')::mapTo({ type: 'ACTION_2' })
);

const middleware = createEpicMiddleware(epic);
Expand All @@ -38,18 +46,19 @@ describe('createEpicMiddleware', () => {
it('should allow you to replace the root epic with middleware.replaceEpic(epic)', () => {
const reducer = (state = [], action) => state.concat(action);
const epic1 = action$ =>
Observable.merge(
Observable.of({ type: 'EPIC_1' }),
action$.ofType('FIRE_1').mapTo({ type: 'ACTION_1' }),
action$.ofType('FIRE_2').mapTo({ type: 'ACTION_2' }),
action$.ofType('FIRE_GENERIC').mapTo({ type: 'EPIC_1_GENERIC' })
mergeStatic(
of({ type: 'EPIC_1' }),
action$.ofType('FIRE_1')::mapTo({ type: 'ACTION_1' }),
action$.ofType('FIRE_2')::mapTo({ type: 'ACTION_2' }),
action$.ofType('FIRE_GENERIC')::mapTo({ type: 'EPIC_1_GENERIC' }),
action$.ofType(EPIC_END)::mapTo({ type: 'CLEAN_UP_AISLE_3' })
);
const epic2 = action$ =>
Observable.merge(
Observable.of({ type: 'EPIC_2' }),
action$.ofType('FIRE_3').mapTo({ type: 'ACTION_3' }),
action$.ofType('FIRE_4').mapTo({ type: 'ACTION_4' }),
action$.ofType('FIRE_GENERIC').mapTo({ type: 'EPIC_2_GENERIC' })
mergeStatic(
of({ type: 'EPIC_2' }),
action$.ofType('FIRE_3')::mapTo({ type: 'ACTION_3' }),
action$.ofType('FIRE_4')::mapTo({ type: 'ACTION_4' }),
action$.ofType('FIRE_GENERIC')::mapTo({ type: 'EPIC_2_GENERIC' })
);

const middleware = createEpicMiddleware(epic1);
Expand All @@ -75,6 +84,8 @@ describe('createEpicMiddleware', () => {
{ type: 'ACTION_2' },
{ type: 'FIRE_GENERIC' },
{ type: 'EPIC_1_GENERIC' },
{ type: EPIC_END },
{ type: 'CLEAN_UP_AISLE_3' },
{ type: 'EPIC_2' },
{ type: 'FIRE_3' },
{ type: 'ACTION_3' },
Expand All @@ -92,7 +103,7 @@ describe('createEpicMiddleware', () => {
const middleware = createEpicMiddleware();
const store = createStore(reducer, applyMiddleware(middleware));

store.dispatch(() => Observable.of({ type: 'ASYNC_ACTION_1' }));
store.dispatch(() => of({ type: 'ASYNC_ACTION_1' }));

expect(console.warn.calledOnce).to.equal(true);
expect(
Expand All @@ -109,8 +120,8 @@ describe('createEpicMiddleware', () => {

const store = createStore(reducer, applyMiddleware(middleware));

store.dispatch(() => Observable.of({ type: 'ASYNC_ACTION_1' }).delay(10));
store.dispatch(() => Observable.of({ type: 'ASYNC_ACTION_2' }).delay(20));
store.dispatch(() => of({ type: 'ASYNC_ACTION_1' })::delay(10));
store.dispatch(() => of({ type: 'ASYNC_ACTION_2' })::delay(20));

// HACKY: but should work until we use TestScheduler.
setTimeout(() => {
Expand Down Expand Up @@ -215,15 +226,15 @@ describe('createEpicMiddleware', () => {
const store = createStore(reducer, applyMiddleware(middleware));

store.dispatch(
(action$) => Observable.of({ type: 'ASYNC_ACTION_2' })
.delay(10)
.takeUntil(action$.filter(action => action.type === 'ASYNC_ACTION_ABORT'))
.merge(
(action$) => of({ type: 'ASYNC_ACTION_2' })
::delay(10)
::takeUntil(action$::filter(action => action.type === 'ASYNC_ACTION_ABORT'))
::merge(
action$
.map(action => ({ type: action.type + '_MERGED' }))
.take(1)
::map(action => ({ type: action.type + '_MERGED' }))
::take(1)
)
.startWith({ type: 'ASYNC_ACTION_1' })
::startWith({ type: 'ASYNC_ACTION_1' })
);

store.dispatch({ type: 'ASYNC_ACTION_ABORT' });
Expand All @@ -247,8 +258,8 @@ describe('createEpicMiddleware', () => {

const store = createStore(reducer, applyMiddleware(middleware));

const action2 = (action$) => Observable.of({ type: 'ASYNC_ACTION_2' });
const action1 = (action$) => Observable.of({ type: 'ASYNC_ACTION_1' }, action2);
const action2 = (action$) => of({ type: 'ASYNC_ACTION_2' });
const action1 = (action$) => of({ type: 'ASYNC_ACTION_1' }, action2);

store.dispatch(action1);

Expand Down
17 changes: 15 additions & 2 deletions test/typings.ts
Expand Up @@ -2,6 +2,8 @@ import { createEpicMiddleware, Epic, combineEpics,
EpicMiddleware, ActionsObservable } from '../index';
import { Action } from 'redux';
import { Observable } from 'rxjs/Observable';
import 'rxjs/add/observable/of';
import 'rxjs/add/operator/mapTo';

const epic1: Epic = (action$, store) =>
action$.ofType('FIRST')
Expand All @@ -17,8 +19,19 @@ const epic2: Epic = (action$, store) =>
payload: store.getState()
});

const rootEpic1: Epic = combineEpics(epic1, epic2);
const rootEpic2: Epic = combineEpics(epic1, epic2);
const epic3: Epic = action$ =>
action$.ofType('THIRD')
.mapTo({
type: 'third'
});

const epic4: Epic = () =>
Observable.of({
type: 'third'
});

const rootEpic1: Epic = combineEpics(epic1, epic2, epic3, epic4);
const rootEpic2: Epic = combineEpics(epic1, epic2, epic3, epic4);

const epicMiddleware: EpicMiddleware = createEpicMiddleware(rootEpic1);

Expand Down

0 comments on commit fef6f80

Please sign in to comment.