State management by RxJS

rxjs-state-management.ts

import { Observable, Subject, BehaviorSubject } from 'rxjs'

interface State {
    counter: { count: number },
    user: { name: string, age: number }
}

const models = {
    count(intent: { inc$: Observable<number>, dec$: Observable<number> }) {
        type S = State['counter']
        return Observable.of((_s: S) => ({ count: 0 })).merge(
            intent.inc$.map(n => (s: S) => ({ count: s.count + n })),
            intent.dec$.map(n => (s: S) => ({ count: s.count - n })),
        )
    },
    user(intent: { update$: Observable<{ name?: string, age?: number }> }) {
        type S = State['user']

        return Observable.of((_: S) => ({ name: 'unknown', age: 0 })).merge(
            intent.update$.pluck<Partial<S>, string>('name')
                .filter(Boolean)
                .map((name => (s: S) => ({ name, age: s.age }))),

            intent.update$.pluck<Partial<S>, number>('age')
                .filter(Boolean)
                .map((age => (s: S) => ({ name: s.name, age }))),
        )
    },
}

const actions = {
    inc$: new Subject<number>(),
    dec$: new Subject<number>(),
    update$: new Subject<{ name?: string, age?: number }>(),
}

const intents = {
    inc$: actions.inc$.map(n => n * 10),
    dec$: actions.dec$.map(n => n * 10),
    update$: actions.update$,
}

const state$ = new BehaviorSubject<State>({} as any)

/* subscribe */
const subscription = attachMutations(state$, [
    models.count(intents).let(makeStateful).map(s => ({ counter: s })),
    models.user(intents).let(makeStateful).map(s => ({ user: s })),
])

// unsubscribe all mutation
state$.filter(s => s === null).subscribe(() => {
    subscription.unsubscribe()
})

state$.map(state => state.user)
    .distinctUntilChanged()
    .subscribe(console.log)

state$.pluck<any, number>('counter', 'count')
    .distinctUntilChanged()
    .subscribe(console.log)

/* kick action */
actions.inc$.next(1)
actions.inc$.next(1)
actions.dec$.next(1)
actions.inc$.next(10)

actions.update$.next({ name: 'taro' })
actions.update$.next({ age: 10 })
actions.update$.next({ name: 'taro yamada', age: 20 })

//
// ─── LIB ────────────────────────────────────────────────────────────────────────
//

function makeStateful<T>(reducer$: Observable<(state: T) => T>) {
    return reducer$.scan((acc, fn) => fn(acc), {} as T)
        .distinctUntilChanged()
        .shareReplay(1)
}

// tslint:disable:no-shadowed-variable
function attachMutations<T>(state$: Subject<T>, mutations: Observable<Partial<T>>[]) {
    return Observable.merge(...mutations)
        .withLatestFrom(state$, assign)
        .subscribe({ next, error, complete })

    function assign(next: Partial<T>, cur: T) {
        return Object.assign({}, cur, next)
    }
    function next(value: T) {
        return state$.next(value)
    }
    function error(err: any) {
        console.error(err)
    }
    function complete() {
        /* noop */
    }
}

参考 : Use RxJS with React - Michal Zalecki

Use RxJS with React - Michal Zaleckiでは、actionからreducerまでにcurrentStateの参照が一切取得できないが、現実的にはintent層(reduxのmiddleware層)でcurrentStateに応じた処理(HTTP request等の副作用)を行うユースケースがある。

intent層でstate$.next()を呼ばない前提で、reduxのようにクロージャーでstateを隠さずintentで参照可能にしている。

initial stateはmodelの中でObservable.of(() => initialState)をするならば、const state$ = new BehaviorSubject<State>({} as any) でよい気がする.

逆にmodelの中でinitial stateを用意しない場合、intentを受けたobservableが返すreducer関数の中でstateの参照が必要なため、 BehaviorSubjectの中でinitial stateは必須になる。

https://github.com/cotttpan/frxx