Event Composition using events to control logic.

why.js

#!/usr/bin/env node-forward
var Rx = require('rx'),
    EventEmitter = require('events').EventEmitter,
    fromEvent = Rx.Observable.fromEvent;

function throwError(err) {
  return Rx.Observable.throw(err);
} 

// Could be any data source
var e = new EventEmitter(); 

var connectEvent = fromEvent(e, 'connect'),
    dataEvent = fromEvent(e, 'data'),
    errorEvent = fromEvent(e, 'error'),
    endEvent = fromEvent(e, 'end');

// Cross the streams!  
// Note that if an error or end happens, we terminate and clean up handlers!
var stream = dataEvent
  .merge(errorEvent.flatMap(throwError))
  .skipUntil(connectEvent.amb(errorEvent))
  .takeUntil(endEvent.amb(errorEvent));
  
// Start you a processing
var subscription = stream.subscribe(
  function (data) {
    // Do something with the data
  },
  function (err) {
    // Tell the world we've got a problem
  },
  function () {
    // Tell the world we're done 
  });