WIP: feat(lib): implement RxAutomaton inspired by the slide

src/client/example_rxautomaton.ts

@@ -0,0 +1,66 @@
+import * as Rx from 'rxjs';
+import {Automaton} from '../lib/RxAutomaton';
+
+// tslint:disable-next-line: no-namespace
+declare global {
+    interface Window {
+        auto: Automaton<State, Input>;
+        input: Rx.Subject<number>;
+    }
+}
+
+const input = new Rx.Subject<number>();
+
+interface State {
+    current: number;
+}
+
+const enum OpCode {
+    Increment,
+    Decrement,
+}
+
+interface Input {
+    type: OpCode;
+    value: number;
+}
+
+const inputStream = input.map((value): Input => {
+    return {
+        type: OpCode.Increment,
+        value,
+    };
+});
+
+window.auto = new Automaton<State, Input>({ current: 0 }, inputStream, (state: State, input: Input) => {
+    switch (input.type) {
+        case OpCode.Increment:
+            return {
+                state: {
+                    current: state.current + input.value,
+                },
+                input: Rx.Observable.of<Input>({
+                    type: OpCode.Decrement,
+                    value: input.value,
+                }).delay(500),
+            };
+
+        case OpCode.Decrement:
+            return {
+                state: {
+                    current: state.current - input.value,
+                },
+                input: Rx.Observable.empty<Input>(),
+            };
+    }
+});
+
+window.auto.state().asObservable().subscribe((state) => {
+    console.log(`new state: ${state.current}`);
+}, (e) => {
+    console.error(e);
+});
+
+window.input = input;
+
+input.next(1);

src/lib/RxAutomaton.ts

@@ -0,0 +1,143 @@
+import {
+    Observable,
+    Subject,
+    Subscription,
+} from 'rxjs';
+import {ReactiveProperty} from './ReactiveProperty';
+
+const enum TransitionType {
+    Success,
+    Failure,
+}
+
+type TransitionSucess<TState, TInput> = {
+    type: TransitionType.Success,
+    next: TState,
+    input: Observable<TInput>,
+};
+
+type TransitionFailure<TState, TInput> = {
+    type: TransitionType.Failure,
+    from: TState,
+    input: TInput,
+    err: Error,
+};
+
+type TransitionResult<TState, TInput> = TransitionSucess<TState, TInput> | TransitionFailure<TState, TInput>;
+
+function isTransitionSucess<TState, TInput>(v: TransitionResult<TState, TInput>): v is TransitionSucess<TState, TInput> {
+    return v.type === TransitionType.Success;
+}
+
+type NextMapping<TState, TInput> = (state: TState, input: TInput) => {
+    state: TState,
+    input: Observable<TInput>;
+};
+
+/**
+ *  Inspired by https://speakerdeck.com/inamiy/reactive-state-machine-japanese?slide=65
+ */
+export class Automaton<TState, TInput> {
+    private _state: ReactiveProperty<TState>;
+    private _disposer: Subscription;
+
+    constructor(initial: TState, input: Observable<TInput>, mapping: NextMapping<TState, TInput>) {
+        const state = new ReactiveProperty(initial);
+        const nextState: Observable<TState> = transitionState(state.asObservable(), input, mapping);
+        this._state = state;
+        this._disposer = nextState.subscribe(state);
+    }
+
+    state(): ReactiveProperty<TState> {
+        return this._state;
+    }
+}
+
+function transitionState<TState, TInput>(state: Observable<TState>,
+                                         input: Observable<TInput>,
+                                         mapping: NextMapping<TState, TInput>): Observable<TState> {
+    const inputPipe = new Subject<Observable<TInput>>();
+    const nextInput: Observable<TInput> = inputPipe.flatMap((inner) => inner);
+    const grandInput = input.merge<TInput>(nextInput);
+
+    type Result = TransitionResult<TState, TInput>;
+    type Success = TransitionSucess<TState, TInput>;
+
+    const transition: Observable<Result> = grandInput
+        .withLatestFrom(state, (input: TInput, from: TState) => {
+            return {
+                input,
+                from,
+            };
+        }).map((container) => {
+            return callStateMapper<TState, TInput>(mapping, container);
+        });
+
+    const postTransition: Observable<Result> = transition
+        .do((result: Result) => {
+            switch (result.type) {
+                case TransitionType.Success:
+                    inputPipe.next(result.input);
+                    break;
+                case TransitionType.Failure:
+                    console.error(result);
+                    break;
+                default:
+                    throw new RangeError('undefined TransitionType');
+            }
+        })
+        .do((result: Result) => {
+            let type: string;
+            switch (result.type) {
+                case TransitionType.Success:
+                    type = 'Success';
+                    break;
+                case TransitionType.Failure:
+                    type = 'Failure';
+                    break;
+                default:
+                    throw new RangeError('undefined TransitionType');
+            }
+            console.group();
+                console.log(`type: ${type}`);
+                console.dir(result);
+            console.groupEnd();
+        });
+
+    const successTransition = postTransition.filter<Result, Success>(isTransitionSucess);
+
+    return successTransition.map((container) => {
+        if (container.type === TransitionType.Success) {
+            return container.next;
+        }
+        else {
+            throw new TypeError('unreachable');
+        }
+    });
+}
+
+function callStateMapper<TState, TInput>(mapping: NextMapping<TState, TInput>,
+                                         container: { from: TState, input: TInput }): TransitionResult<TState, TInput> {
+    const { input, from, } = container;
+    let next: {
+        state: TState,
+        input: Observable<TInput>;
+    };
+    try {
+        next = mapping(from, input);
+    }
+    catch (err) {
+        return {
+            type: TransitionType.Failure,
+            from,
+            input,
+            err,
+        } as TransitionFailure<TState, TInput>;
+    }
+
+    return {
+        type: TransitionType.Success,
+        next: next.state,
+        input: next.input,
+    } as TransitionSucess<TState, TInput>;
+}

src/lib/test/test_RxAutomaton.ts

@@ -0,0 +1,83 @@
+import * as assert from 'assert';
+import {
+    Observable,
+    Subject,
+} from 'rxjs';
+
+import {Automaton} from '../RxAutomaton';
+
+describe('RxAutomaton', () => {
+
+    describe('Automaton', () => {
+        describe('state()', () => {
+
+            describe('get initial state', () => {
+                const INITIAL_STATE = 0;
+
+                let resultBySubscribe: number;
+                let resultByGetter: number;
+
+                before((done) => {
+                    const input = new Subject<number>();
+                    const m = new Automaton<number, number>(INITIAL_STATE, input, (state: number, _: number) => {
+                        return {
+                            state,
+                            input: Observable.empty<number>(),
+                        };
+                    });
+
+                    const state = m.state();
+                    resultByGetter = state.value();
+                    state.asObservable().subscribe((state) => {
+                        resultBySubscribe = state;
+                    }, done, done);
+
+                    state.complete();
+                });
+
+                it('initial state from subscription', () => {
+                    assert.deepStrictEqual(resultBySubscribe, INITIAL_STATE);
+                });
+
+                it('initial state from getter', () => {
+                    assert.deepStrictEqual(resultByGetter, INITIAL_STATE);
+                });
+            });
+
+            describe('set state from outer', () => {
+                const seq: Array<number> = [];
+                const mapperSeq: Array<number> = [];
+
+                before((done) => {
+                    const input = new Subject<number>();
+                    const m = new Automaton<number, number>(0, input, (state: number, _: number) => {
+                        mapperSeq.push(state);
+
+                        return {
+                            state,
+                            input: Observable.empty<number>(),
+                        };
+                    });
+
+                    const state = m.state();
+                    state.asObservable().subscribe((state) => {
+                        seq.push(state);
+                    }, done, done);
+
+                    state.setValue(1);
+                    state.setValue(2);
+                    state.setValue(3);
+                    state.complete();
+                });
+
+                it('state should be updated', () => {
+                    assert.deepStrictEqual(seq, [0, 1, 2, 3]);
+                });
+
+                it('mapper should not call', () => {
+                    assert.deepStrictEqual(mapperSeq, []);
+                });
+            });
+        });
+    });
+});

src/lib/test_manifest.js

@@ -4,4 +4,5 @@ import './test/test_FetchDriver_fetch';
 import './test/test_FetchDriver_get';
 import './test/test_FetchDriver_post';
 import './test/test_FetchDriver_utils';
+import './test/test_RxAutomaton';
 import './test/test_ViewContext';