rxjs-saga-auth.js

import { createStore, applyMiddleware } from 'redux';
import { Observable, Subject } from 'rxjs';

const api = type => {
  console.log(`calling API ${type}`);
  return new Promise(res => setTimeout(() => res(), 500));
};

const actionOrder = (actions, order) => actions.every((action, index) => action.type === order[index]);
const actionPredicate = actions => filterableAction => actions.some(action => action === filterableAction.type);
const buildAction = type => () => ({ type });

const AUTH_EXPIRATION = 1000;

const LOG_OUT_ACTIONS_ORDER = ['LOG_IN', 'LOGGED_IN', 'LOG_OUT'];

// User clicked the log in button,
// call the API and respond either success or failure
const authGetTokenSaga = iterable => iterable
  .filter(actionPredicate(['LOG_IN']))
  .flatMap(() => Observable
    .fromPromise(api('get token'))
    .map(buildAction('LOGGED_IN'))
    .catch(reason => Observable.of({type: 'LOG_IN_FAILURE', payload: reason})));

// After the user is successfuly logged in,
// let's schedule an infinite interval stream
// which can be interrupted either by LOG_OUT
// or failure in refreshing (TOKEN_REFRESHING_FAILED)
const authRefreshTokenSaga = iterable => iterable
  .filter(actionPredicate(['LOGGED_IN']))
  .flatMap(() => Observable
    .interval(AUTH_EXPIRATION)
    .flatMap(() => Observable
      .fromPromise(api('refresh token'))
      .map(buildAction('TOKEN_REFRESHED'))
      .catch(() => Observable.of(
        {type: 'TOKEN_REFRESHING_FAILED'}))
    )
    .takeUntil(iterable.filter(actionPredicate(['LOG_OUT', 'TOKEN_REFRESHING_FAILED'])))
  );

// Observe all the actions in specific order
// to determine whether user wants to log out
const authHandleLogOutSaga = iterable => iterable
  .filter(filteringAction => [...LOG_OUT_ACTIONS_ORDER, 'TOKEN_REFRESHING_FAILED']
    .some(action => action === filteringAction.type))
  .bufferCount(LOG_OUT_ACTIONS_ORDER.length)
  .filter(actions => actionOrder(actions, LOG_OUT_ACTIONS_ORDER))
  .flatMap(() => Observable
    .fromPromise(api('log out'))
    .map(buildAction('LOGGED_OUT')));


const authSaga = iterable => Observable.merge(
  authGetTokenSaga(iterable),
  authRefreshTokenSaga(iterable),
  authHandleLogOutSaga(iterable)
);

const sagaMiddleware = saga => {
  const subject = new Subject();

  return store => {
    saga(subject).subscribe(dispatchable => store.dispatch(dispatchable));

    return next => action => {
      next(action);
      subject.next({action, state: store.getState()});
    };
  };
};

const storeFactory = applyMiddleware(sagaMiddleware(authSaga))(createStore);

const store = storeFactory((appState, action) => {
  console.log(action);
  return appState;
});

store.dispatch({type: 'LOG_IN'});
setTimeout(() => {
  store.dispatch({type: 'LOG_OUT'});
}, 10000);

@bgerm - no it's regression as @staltz already mentioned - has been fixed in the repo https://github.com/tomkis1/redux-saga-rxjs/blob/master/src/index.js#L27