Reactive functional programming

core ideas

Observable

from http://reactivex.io/rxjs/manual/overview.html#subscribing-to-observables

Observables are like functions with zero arguments, but generalize those to allow multiple values.

observer.next(42) is like return 42

 var foo = Rx.Observable.create(function (observer) {
   console.log('Hello');
   observer.next(42);
 });

 foo.subscribe(function (x) {
   console.log(x);
 });

Subscribing to an Observable is analogous to calling a Function.

 console.log('before');
 foo.subscribe(function (x) {
   console.log(x);
 });
 console.log('after');

Observables are able to deliver values either synchronously or asynchronously

each oberver.next is like a return statement

Observables can be created with create, but usually we use the so-called creation operators, like of, from, interval, etc.

Subscribing to an Observable is like calling a function, providing callbacks where the data will be delivered to.

The Observable does not even maintain a list of attached Observers.

In an Observable Execution, zero to infinite Next notifications may be delivered. If either an Error or Complete notification is delivered, then nothing else can be delivered afterwards.

When you subscribe, you get back a Subscription, which represents the ongoing execution. Just call unsubscribe() to cancel the execution.

Observers are just objects with three callbacks, one for each type of notification that an Observable may deliver.

Here, all three types of callbacks are provided as 3 arguments, the first callback argument as the next handler

 observable.subscribe(
   x => console.log('Observer got a next value: ' + x),
   err => console.error('Observer got an error: ' + err),
   () => console.log('Observer got a complete notification')
 );

Subscription

A Subscription essentially just has an unsubscribe() function to release resources or cancel Observable executions.

(be carefull not to think you can observable.unsubscribe())

 var observable = Rx.Observable.interval(1000);
 var subscription = observable.subscribe(x => console.log(x));
 // Later:
 // This cancels the ongoing Observable execution which
 // was started by calling subscribe with an Observer.
 subscription.unsubscribe();

Subject

A Subject is like an Observable, but can multicast to many Observers. Subjects are like EventEmitters

they maintain a registry of many listeners.

Since a Subject is an Observer, this also means you may provide a Subject as the argument to the subscribe of any Observable, like the example below shows:

 var subject = new Rx.Subject();

 subject.subscribe({
   next: (v) => console.log('observerA: ' + v)
 });
 subject.subscribe({
   next: (v) => console.log('observerB: ' + v)
 });

 var observable = Rx.Observable.from([1, 2, 3]);

 observable.subscribe(subject); // You can subscribe providing a Subject

Which executes as:

 observerA: 1
 observerB: 1
 observerA: 2
 observerB: 2
 observerA: 3
 observerB: 3

With the approach above, we essentially just converted a unicast Observable execution to multicast, through the Subject. This demonstrates how Subjects are the only way of making any Observable execution be shared to multiple Observers

Multicasted Observables

A multicasted Observable uses a Subject under the hood to make multiple Observers see the same Observable execution.

 var source = Rx.Observable.from([1, 2, 3]);
 var subject = new Rx.Subject();
 var multicasted = source.multicast(subject);

 // These are, under the hood, `subject.subscribe({...})`:
 multicasted.subscribe({
   next: (v) => console.log('observerA: ' + v)
 });
 multicasted.subscribe({
   next: (v) => console.log('observerB: ' + v)
 });

 // This is, under the hood, `source.subscribe(subject)`:
 multicasted.connect();

Because connect() does source.subscribe(subject) under the hood, connect() returns a Subscription, which you can unsubscribe from in order to cancel the shared Observable execution.