micro-reactor.js

function Dispatcher(subscribe){
  var subscriptions = [];

  this.push = function(e){
    subscriptions.forEach(function(s){
      s(e);
    });
  };

  this.subscribe = function(sink){
    subscriptions = subscriptions.concat(sink);
    if(subscriptions.length == 1){
      subscribe(this.push);
    }
  }.bind(this);
}

function EventStream(subscribe){
  var dispatcher = new Dispatcher(subscribe);
  this.subscribe = dispatcher.subscribe;
}

EventStream.prototype.onValue = function(fn){
  return this.subscribe(function(e){
    return fn(e.value);
  });
};

EventStream.prototype.onError = function(fn){
  return this.subscribe(function(e){
    if(e.failed){
      return fn(e.value);
    }
  });
};

EventStream.prototype.onDone = function(fn){
  return this.subscribe(function(e){
    if(e.done){
      return fn();
    }
  });
};

EventStream.prototype.push = function(value){
  this.subscribers.forEach(function(sub){
    sub(value);
  });
};

function Event(value){
  this.value = value;
  this.failed = false;
  this.done = false;
}

function next(value){
  return new Event(value);
}

function error(msg){
  var e = new Event(msg);
  e.failed = true;
  return e;
}

function done(){
  var e = new Event();
  e.done = true;
  return e;
}

function reactor(){
  var sink;

  var r = new EventStream(function(s){
    sink = s;
  });

  r.push = function(value){
    sink(next(value));
  };

  r.error = function(err){
    sink(error(err));
  };

  r.done = function(){
    sink(done());
  };

  return r;
}

var signals = {
  changed: reactor(),
  started: reactor(),
  completed: reactor()
};

var doChange = function(){
  signals.changed.push({ old: "blah", "new": "diddy" });
};

var doStart = function(){
  signals.started.push();
};

var doComplete = function(){
  signals.completed.push({ blah: "diddy" });
  setTimeout(signals.completed.done, 6000);
};

var doError = function(){
  signals.started.error("Could not start");
};

signals.started.onValue(function(){
  console.log("started");
});

signals.changed.onValue(function(changes){
  console.log("changed", changes);
});

signals.completed.onValue(function(data){
  console.log("completed with: ", data);
});

signals.started.onError(function(error){
  console.log("error starting: ", error);
});

signals.completed.onDone(function(){
  console.log("done completing");
});

setTimeout(doStart, 1000);
setTimeout(doChange, 3000);
setTimeout(doComplete, 5000);
setTimeout(doChange, 7000);
setTimeout(doError, 9000);