Rxjs

10.8.js

import { createStore } from 'redux';
import Rx from 'rxjs/Rx';

Rx.Observable.prototype.ofType = function (...types) {
	return this.filter(({ type }) => {
		const len = types.length;
		switch (len) {
			case 0:
				throw new Error('Must specify at least one type! ');
			case 1:
				return type === types[0];
			default:
				return types.indexOf(type) > -1;
		}
	});
};

function createMiddleware(store, epics) { // ❶
	const input$ = new Rx.Subject(); // ❷
	const actions = epics.map(epic => //❸
		epic(input$, store)); // ❹
	const combinedActions$ = Rx.Observable
		.merge(...actions) //❺
		.publish(); // ❻
	combinedActions$.subscribe(input$); //❼
	combinedActions$.subscribe(action => {
		console.log("Store", store); // Store here is an observable!
		return store.dispatch(action) // ERROR!!!
	}); //❽
	const sub = combinedActions$.connect(); //❾
	return {
		dispatch: (action) => input$.next(action), //❿
		unsubscribe: () => sub.unsubscribe() //⓫
	};
}

function createStreamFromStore(store) {
	return Rx.Observable.from(store)
		.map(() => store.getState()) //❶ store.getState() is called twice, so that subscribers always receive the latest state changes.
		.publishBehavior(store.getState()) //❷ publishBehavior() is a flavor of a multicast (hot) operator that emits the latest value to all subscribers
		.refCount(); //❸ Make this stream go live as soon as the first observer subscribes
}

// start here

function simpleReducer(state, action) {
	switch(action.type) {
		case 'LOG':
			return {...state, messages: [...action.payload, 'in Redux!']};
		default:
			return state;
	}
}
const store = createStreamFromStore(
	createStore(simpleReducer, {messages: []})
);

store.subscribe(({messages}) => console.log(messages.join('=>')));

function simpleEpic(action$, store) {
	return action$.ofType('LOG')
		.delay(1000)
		.map(action => ({...action, payload: [...action.payload, 'in Rx!']}));
}

const disposableDispatcher = createMiddleware(store, [simpleEpic]);

disposableDispatcher.dispatch({
	type: 'LOG',
	payload: ['Hello']
});