An idea of side-effect-free epic like redux-saga for redux-observable

epic.js

// Explicit in-out of side-effect actions.
function epic(action$, store) {
  const fooReq$ = action$.ofType('FOO')
    .map(action => call('FOO_REQ', webapi.getFoo, action.payload.id));
  const foo$ = action$.ofType('FOO_REQ')
    .map(foo => ({ type: 'FOO_FETCHED', payload: { foo } }));

  return Observable.merge(
    fooReq$,
    foo$
  );
}

// Custom operator to hide side-effect actions.
const CALL_REQUEST = Symbol('CALL_REQUEST');
const CALL_RESPONSE = Symbol('CALL_RESPONSE');
const callMiddleware = store => next => action => {
  if (action[CALL_REQUEST]) {
    const payload = action[CALL_REQUEST].payload;
    // TODO: Handle errors.
    // TODO: Handle other async primitives like thunk.
    payload[0].apply(null, payload.slice(1))
      .then(result => next({
        [CALL_RESPONSE]: {
          id: action[CALL_REQUEST].id,
          payload: result
        }
      }));
  } else {
    next(action);
  }
};
function callOperator(action$, call$, callback) {
  return Observable.create(subscriber => {
    const id = uuid();
    const source = this;
    const responseSub = action$
      .filter(a => a[CALL_RESPONSE] && a[CALL_RESPONSE].id === id)
      .map(a => a[CALL_RESPONSE].payload)
      .subscribe(subscriber);
    const requestSub = source.subscribe(
      value => {
        try {
          call$.next({
            [CALL_REQUEST]: { id, payload: callback(value) }
          });
        } catch (e) {
          subscriber.error(e);
        }
      },
      err => subscriber.error(err),
      () => subscriber.complete()
    );
    // https://github.com/ReactiveX/rxjs/issues/1583
    return new Subscription(() => {
      [responseSub, requestSub].forEach(s => s.unsubscribe());
    });
  });
}
Observable.prototype.call = callOperator;
function epic(action$, store) {
  const call$ = new Subject();

  const foo$ = action$.ofType('FOO')
    .call(action$, call$, action => [webapi.getFoo, action.payload.id])
    .map(foo => ({ type: 'FOO_FETCHED', payload: { foo } }));

  return Observable.merge(
    call$,
    foo$
  );
}