redux-eventemitter2.ts

import { EventEmitter2 } from 'eventemitter2';
import { Stream, fromEvent, mergeArray } from 'most';
import { MiddlewareAPI, Middleware, Dispatch } from 'redux';

// ==================================================================
// EventSource
// ==================================================================
export type Listener<T, K extends keyof T> = (arg: T[K]) => any;

export class TypedEventEmitter<T = any> extends EventEmitter2 {
    on<K extends keyof T>(event: K, listener: Listener<T, K>) {
        return super.on(event, listener);
    }

    once<K extends keyof T>(event: K, listener: Listener<T, K>) {
        return super.once(event, listener);
    }

    off<K extends keyof T>(event: K, listener: Listener<T, K>) {
        return super.removeListener(event, listener);
    }

    emit<K extends keyof T>(event: K, arg: T[K]) {
        return super.emit(event, arg);
    }

    dispatch<K extends keyof T>(event: K, arg: T[K]) {
        return super.emitAsync(event, arg);
    }
}

// ==================================================================
// Listen with stream
// ==================================================================
export function select<T, K extends keyof T>(target: K, src: TypedEventEmitter<T>): Stream<T[K]> {
    return fromEvent(target, src) as any;
}

// ==================================================================
// Redux Middleware
// ==================================================================
export interface RequireCtx {
    eventSource: TypedEventEmitter;
}

export interface MergedCtx<A, S> {
    next: TypedEventEmitter<A>['dispatch'];
    store: MiddlewareAPI<S>;
}

export interface Epic<A, S, C = any> {
    (eventSource: TypedEventEmitter<A>, ctx: MergedCtx<A, S> & C): Stream<any>;
}

export function createEpicMiddleware<C extends RequireCtx>(context: C) {
    return <S = any>(epics: Epic<any, S, C>[]) => {
        const middleware = (store: MiddlewareAPI<S>) => (next: Dispatch<S>) => {
            const emitter = context.eventSource;
            const ctx = Object.assign({}, context, {
                store,
                next: emitter.dispatch.bind(emitter)
            });

            emitter.prependAny((type, payload) => next({ type, payload }));
            mergeArray(epics.map(ep => ep(emitter, ctx))).drain();

            return next;
        };

        return middleware as any as Middleware;
    };
}

////////////////////////////////////////////////////////////////////////////////////////
import { createStore, applyMiddleware, combineReducers } from 'redux';

interface S {
    counter: { count: number };
}

interface ActionMap {
    increment: number;
    decrement: number;
}

type E = Epic<ActionMap, S, { ctx: number }>;

const emitter = new TypedEventEmitter<ActionMap>();

const increment: E = (ev, { store, dispatch }) => {
    return select('increment', ev)
        .map(x => x + 1)
        .tap(x => dispatch('decrement', x));
};

const counterReducer = (s: S['counter'] = { count: 0 }, action: any) => {
    switch (action.type) {
        case 'increment':
            return { count: s.count + action.payload };
        case 'decrement':
            return { count: s.count - action.payload };
        default:
            return s;
    }
};

const reducer = combineReducers({ counter: counterReducer });
const middleware = createEpicMiddleware({ eventSource: emitter, ctx: 1 })([increment]);
const store = createStore(reducer, applyMiddleware(middleware));

store.subscribe(() => console.log(store.getState()));
emitter.dispatch('increment', 1);
emitter.dispatch('increment', 5);