rxjs.ts

interface LoginRequest {
  kind: "LoginRequest";
  username: string;
  password: string;
}

interface LoginSuccess {
  kind: "LoginSuccess";
  username: string;
}

interface LoginError {
  kind: "LoginError";
  error: string;
}

interface Logout {
  kind: "Logout";
}

interface State {
  username?: string;
}

type Action = LoginRequest | LoginSuccess | LoginError | Logout;

type AffectorFn = <S, A>(
  s: S,
  a: A,
  s$: Observable<S>,
  a$: Observable<A>
) => [S, Observable<A>];

function affector(
  s: State,
  a: Action,
  s$: Observable<State>,
  a$: Observable<Action>
): [State, Observable<Action>] {
  switch (a.kind) {
    case "LoginRequest":
      // the LoginRequest is cancellable if it sees another Login or Logout action while executing
      return [
        s,
        of<Action>({ kind: "LoginSuccess", username: a.username }).pipe(
          delay(1500),
          takeUntil(
            a$.pipe(
              filter(a => {
                switch (a.kind) {
                  case "LoginSuccess":
                  case "LoginRequest":
                  case "Logout":
                    return true;
                }
                return false;
              })
            )
          )
        )
      ];
    case "LoginSuccess":
      return [{ ...s, username: a.username }, empty()];
    case "Logout":
      return [{ ...s, username: null }, empty()];
  }
  return [s, empty()];
}
  
const affect = (
  affector: AffectorFn | any,
  init: State,
  action$$: Subject<Observable<Action>>
): Observable<State> => {
  const state$ = new Observable(observer => {
    console.log("creating observable");
    let current = init;
    const action$ = action$$.pipe(flatMap(a$ => a$));
    observer.next(current);
    const sub = action$
      .pipe(
        tap(a => {
          console.log("running action", a);
        }),
        map(a => {
          return affector(current, a, state$, action$);
        })
      )
      .subscribe(tuple => {
        current = tuple[0];
        observer.next(current);
        action$$.next(tuple[1]);
      });
    return () => {
      sub.unsubscribe();
    };
  });
  // TODO figure out how to turn into a publishBehavior here in rxjs6
  return state$;
};

const action$$ = new Subject<Observable<Action>>();
const state$ = affect(affector, {}, action$$);
 
console.log("subscribing");
state$.subscribe(state => {
  console.log(state);
});

// push a couple of actions
action$$.next(
  of<Action>({
    kind: "LoginRequest",
    username: "supershabam",
    password: "poop"
  })
);
action$$.next(of<Action>({ kind: "Logout" }));