redux-most-middeware.ts

import { MiddlewareAPI, Middleware, Dispatch } from 'redux';
import { async } from 'most-subject';
import { mergeArray, Stream } from 'most';
import Dispatcher from '@cotto/dispatcher';
import { noop } from './utils';

interface RequireCtx {
    dispatcher: Dispatcher;
}

interface EpicCtx<S, A = any> {
    dispatch: Dispatcher<A>['dispatch'];
    getState: () => S;
}

export interface Action<T = any, K extends keyof T = any> {
    type: K;
    payload: T[K];
}

export interface Epic<A extends Action, S, C = any> {
    (action$: Stream<A>, context: C & EpicCtx<S, A>): Stream<any>;
}

export default function createEpicMiddleware<C extends RequireCtx>(context: C) {
    return <S = any>(epics: Epic<Action, S, C>[]) => (store: MiddlewareAPI<S>) => (next: Dispatch<S>) => {

        const actionIn$ = async<any>();

        const ctx = Object.assign({}, context, {
            getState: store.getState,
            dispatch: context.dispatcher.dispatch.bind(context.dispatcher)
        });

        // dispatcher.dispatch -> next -> epic(s) -> ↓
        context.dispatcher.subscribe(action => actionIn$.next(next(action)));

        // NOTE: epic(s) -> noop
        // next action from epic: epic -> stream.tap(() => ctx.dispatch('action', payload)) -> ↑
        const description = mergeArray(epics.map(ep => ep(actionIn$, ctx))).subscribe({
            next: noop,
            error: (e) => { throw e; },
            complete: () => description.unsubscribe()
        });

        return next;
    };
}