flux using rxjs inspired by redux + redux-observable

example.ts

import { Observable } from 'rxjs'
import { Dispatcher, select, createActionFactory } from './index'
import { bootFlux } from './boot-flux'

/* actions */
interface Actions {
    INCREMENT: number
    DECREMENT: number
    INCREMENT_ASYNC_REQUEST: any
    INCREMENT_ASYNC_COMMIT: number
}
const ACTIONS = createActionFactory<Actions>()


interface State {
    counter: { count: number }
}

function reducer(ev: Dispatcher) {
    type S = State['counter']
    const onInit$ = Observable.of((_: S): S => ({ count: 0 }))

    const onIncrement$ = select(ev, ACTIONS.INCREMENT, ACTIONS.INCREMENT_ASYNC_COMMIT)
        .map(action => action.payload)
        .map(n => (s: S): S => ({ count: s.count + n }))

    const onDecrement$ = select(ev, ACTIONS.DECREMENT)
        .map(action => action.payload)
        .map(n => (s: S): S => ({ count: s.count + n }))

    return Observable.merge(
        onInit$, onIncrement$, onDecrement$,
    )
}

// tslint:disable:no-shadowed-variable
function asyncIncrement(ev: Dispatcher, state$: Observable<State>) {
    return select(ev, ACTIONS.INCREMENT_ASYNC_REQUEST)
        .delay(1000)
        // .do(() => console.log('currrentState', state.getValue()))
        .switchMap(() => state$.first())
        .map(x => ACTIONS.INCREMENT_ASYNC_COMMIT(x.counter.count + 1))
}

function logger(ev: Dispatcher, state$: Observable<any>) {
    return select(ev, '*').withLatestFrom(state$)
        .do(([action, state]) => console.log(action, state))
        .mapTo(null)
}

// /* boot */
const reducers = { counter: reducer }
const epics = [asyncIncrement, logger]

const { dispatch, state$ } = bootFlux(reducers, epics, { wildcard: true })

state$.debounceTime(1).subscribe(state => console.log('state: ', state))

dispatch(ACTIONS('INCREMENT', 1))
dispatch(ACTIONS.INCREMENT(1))
dispatch(ACTIONS.INCREMENT_ASYNC_REQUEST(null))

rx-flux.ts

import { Observable } from 'rxjs/Observable'
import { BehaviorSubject } from 'rxjs/BehaviorSubject'
import 'rxjs/add/operator/scan'
import 'rxjs/add/operator/distinctUntilChanged'
import 'rxjs/add/operator/withLatestFrom'
import 'rxjs/add/operator/filter'
import { existy } from '@cotto/utils.ts'
import { Dispatcher, DispatcherOptions, Action } from './index'

export interface Epic<S> {
    (dispatcher: Dispatcher, state$: BehaviorSubject<S>): Observable<Action | null>
}
export interface Reducer<S> {
    (state: S): S
}
export type ReducerMap<S, K extends keyof S> = {
    [P in K]: (ev: Dispatcher) => Observable<Reducer<S[K]>>
}
export interface FluxOptions extends DispatcherOptions {
    /*  */
}
export function bootFlux<S, K extends keyof S>(reducermap: ReducerMap<S, K>, epics: Epic<S>[] = [], opts: FluxOptions = {}) {
    const state$ = new BehaviorSubject<S>({} as any)
    const dispatcher = new Dispatcher(opts)
    const dispatch = dispatcher.dispatch.bind(dispatcher) as Dispatcher['dispatch']
    const callReducer = (state: S[K], reducer: Reducer<S[K]>) => reducer(state)
    const patch = (s: S) => state$.next(s)

    for (const scope in reducermap) {
        const reducerCreator = reducermap[scope]
        reducerCreator(dispatcher)
            .scan(callReducer, {} as any)
            .distinctUntilChanged()
            .withLatestFrom(state$, (next, cur) => ({ ...cur as any, [scope]: next }))
            .subscribe(patch)
    }
    for (const epic of epics) {
        epic(dispatcher, state$)
            .filter(existy)
            .subscribe(dispatch)
    }

    return { state$, dispatch }
}