How to use is-observable - 9 common examples

  const reducersArr = [].concat(flatened).filter(item => isObservable(item));
  // console.log('RA: ', reducersArr);

const createState = (reducerS, initialState = Rx.Observable.of({})) => {
  if (!isObservable(reducerS)) {
    throw new Error(`createState expects first argument - reducer - to be Observable
    but it is ${reducerS === null ? 'null' : typeof reducerS}`
    );
  }

  if (!isObservable(initialState) && typeof initialState !== 'object') {
    throw new Error(`createState expects second argument - initialState
      to be Observable or Object
      but it is ${initialState === null ? 'null' : typeof initialState}`
    );
  }

  const initialStateS = isObservable(initialState) ? initialState : Rx.Observable.of(initialState);

  // TODO: make create state universal (if not scoped, it will use the whole state)
  // TODO: scan function make external so it can be tested
  return initialStateS
    .merge(reducerS)
    .scan((state, [ scope, reducer ]) =>
      immutable.set(state, scope, reducer(objectPath(state, scope)))
    )
    .publishReplay(1)
    .refCount();
};

test('create with just state', t => {
  t.plan(3);
  const states = [{ some: 'state' }, { someOther: 'other state' }];
  const statesCopy = [].concat(states);
  const state$ = Rx.Observable.from(states);
  const loggerStream$ = createLoggerStream(state$);

  t.true(isObservable(loggerStream$));

  return loggerStream$.subscribe(val => {
    t.deepEqual(val, { streamName: 'state', payload: statesCopy.shift() });
  });
});

const isObservable = (thing: any): thing is Observable => isSomeObservable(thing) || isZenObservable(thing)

const createPushMessageFunctions = (subjects) => {
  if (isObservable(subjects)) return messagesToStream(subjects);

  if (typeof subjects !== 'object' || subjects === null) {
    throw new Error(
      `messagesToStreams expected an object or a function,
      instead received: ${subjects === null ? 'null' : typeof subjects}.`
    );
  }

  return Object.keys(subjects).reduce((acc, key) => {
    const subject = subjects[key];
    if (isObservable(subject)) {
      acc[key] = messagesToStream(subject);
    }
    return acc;
  }, {});
};

  const otherToLog = other.filter(item => isObservable(item));

const createLoggerStream = (state$, ...other) => {
  if (!state$ && !isObservable(state$)) {
    throw new Error(
      'createLoggerStream: you need to pass at least state$ stream.'
    );
  }
  const otherToLog = other.filter(item => isObservable(item));

  const stateToLog$ = state$.map(state => ({
    streamName: 'state',
    payload: state,
  }));
  const toLog = [stateToLog$, ...otherToLog];

  return Rx.Observable
    .merge(...toLog)
    .distinctUntilChanged((a, b) => deepEqual(a, b))
    .publishReplay(1)

export function asObservable (obj: any): Observable {
  if (isObservable(obj)) {
    return obj
  }

  return Observable.of(obj)
}