RxJS: How would I "manually" update an Observable?

Question

In RX, Observer and Observable are distinct entities. An observer subscribes to an Observable. An Observable emits items to its observers by calling the observers' methods. If you need to call the observer methods outside the scope of Observable.create() you can use a Subject, which is a proxy that acts as an observer and Observable at the same time.

You can do like this:

var eventStream = new Rx.Subject();

var subscription = eventStream.subscribe(
   function (x) {
        console.log('Next: ' + x);
    },
    function (err) {
        console.log('Error: ' + err);
    },
    function () {
        console.log('Completed');
    });

var my_function = function() {
  eventStream.next('foo'); 
}

You can find more information about subjects here:

• https://github.com/ReactiveX/rxjs/blob/master/docs_app/content/guide/subject.md
• http://reactivex.io/documentation/subject.html

I believe Observable.create() does not take an observer as callback param but an emitter. So if you want to add a new value to your Observable try this instead:

var emitter;
var observable = Rx.Observable.create(e => emitter = e);
var observer = {
  next: function(next) {
    console.log(next);
  },
  error: function(error) {
    console.log(error);
  },
  complete: function() {
    console.log("done");
  }
}
observable.subscribe(observer);
emitter.next('foo');
emitter.next('bar');
emitter.next('baz');
emitter.complete();

//console output
//"foo"
//"bar"
//"baz"
//"done"

Yes Subject makes it easier, providing Observable and Observer in the same object, but it's not exactly the same, as Subject allows you to subscribe multiple observers to the same observable when an observable only send data to the last subscribed observer, so use it consciously. Here's a JsBin if you want to tinker with it.