RxJs simple state machine

sekretk — Tue, 14 Jul 2020 13:23:42 +0000

This is example of state service with RxJs on typescript

Requirements

To have more than one source of initial state.
Changes can comes from different sources and at unpredictable time
To have possibility to save partial state
To notify state consumers only about special state parts

Implementation

After reading requirements one of the best solution is using RxJs.

Let me show how it's possible

some infrastructural code

export interface IUserState{
    user: string,
    theme: string,
    mode: number
}

const DEFAULT_USER_STATE: IUserState = {
    user: 'John Do',
    theme: 'light',
    mode: 1
}

//awaiting time for persistence response
const DEFAULT_STATE_DELAY = 2000;

//emulation for persistence service feed
const persistanceObservable = 
   of({user: 'Bart Simpson'} as IUserState)
     .pipe(delay(4000));

first i'll create observable of consumer request to change state

private stateChangeRequests$: Subject<Partial<IUserState>> 
    = new Subject<Partial<IUserState>>();

changeState(partialState: Partial<IUserState>): void {
    this.stateChangeRequests$.next((partialState));
}

next construct feed for state from DEFAULT and persisntance:

merge(
   of(DEFAULT_USER_STATE).pipe(delay(DEFAULT_STATE_DELAY)),
   persistanceObservable
)

now we have observable of consumer requests and feed of initial state. Lets merge all together:

merge(
   of(DEFAULT_USER_STATE).pipe(delay(DEFAULT_STATE_DELAY)),
   persistanceObservable
).pipe(exhaustMap(initState => this.stateChangeRequests$.pipe(
            takeUntil(this.dispose$),
            scan((acc, curr) => ({ ...acc, ...curr } 
               as IUserState), initState),
            tap(this.handleState),
            shareReplay(1)
        )));

I've add takeUntil(this.dispose$) in aim to stop service and complete consumer subscriptions
tap(this.handleState) allow us to react in service to result state and shareReplay(1) constrain it to ONLY ONE CALL

OK, OK, BUT HOW TO USE?

Example of usage:

const consoleHandler = (prefix: any) => (value: any) =>
     console.log(`At ${Date.now() - startTime}: 
          ${prefix} ${JSON.stringify(value)}`);

stateMachine.stateObservable().subscribe(
     consoleHandler('Consumer, result state: '))

stateMachine.statePartialObservable(map(state =>
    state.user)).subscribe(
       consoleHandler('User consumer, result state: '))

of(true).pipe(delay(5000)).subscribe(() =>
    stateMachine.changeState({mode: 2}))

SO? WHERE IS PROFIT?

for me i see some advantages:

Service can manipulate with initial state and populate it from any source (maybe add some localstorage if persistence not responding in timeout)
It's possible to add reaction on result state object or only on part of it. It's possible to use same mechanics as consumer: save partial state and subscribe for it. (in this example is used tap operator)
NO STATE! there are no variable _state, no any assignment for state field. All magic goes in scan operator and you can't mess it up

Conclusion

In compare with non-reactive approach code is much more agile and testable (everything in pipe is no-side-effect-function, can be separated and tested), and more durable (there is nothing to mess with)