Mini-CycleJS with RxJS

demo.js

import Rx from "rxjs/Rx";
import cycleRun from "./mini-cycle";


function loggerDriver(source$) {
  source$.subscribe((text) => console.log(text));

  return Rx.Observable.empty();
}

function main() {
  return {
    logger: Rx.Observable.of("hello"),
  };
}

cycleRun(main, {
  logger: loggerDriver, 
});

mini-cycle.js

import Rx from "rxjs/Rx";


/**
 * @param {function(Object):Object}     fn
 * @param {Object.<string, Observable>} drivers
 */
export default function cycleRun(fn, drivers) {
  /** @type {Object.<string, Subject>} */
  const sinksSubjects = {};
  /** @type {Object.<string, Observable>} */
  const sources = {};

  objectForEach(drivers, (driver, key) => {
    const sinkSubject = new Rx.Subject();
    const sink$ = sinkSubject.asObservable();

    sinksSubjects[key] = sinkSubject;
    sources[key] = driver(sink$);
  });

  /** @type {Object.<string, Observable>} */
  const sinks$$ = fn(sources);

  objectForEach(sinks$$, (sink$, key) => {
    const sinkSubject = sinksSubjects[key];
    sink$.subscribe(sinkSubject);
  });
}

object-for-each.js

export default function objectForEach(obj, fn) {
  if (obj != null) {
    Object.keys(obj)
      .forEach((key) => {
        fn(obj[key], key, obj);
      });
  }

  return obj;
}