RxJS Observables, a gentle introduction.

For a long time I struggled to make head or tails of what are Observables, how to use them and how to apply them in Angular apps whilst following best patterns. I’ll try to explain to the best of my understanding what I now know about Observables as I get to learn much more about them.

Introduction

Observables are an implementation of Reactive Extensions for JavaScript popularly known as RxJS which is concerned with data streams and propagation of change. With this in mind, Observables can be viewed as lazy push collections of multiple values. The push paradigm or protocol embodies the idea that the producer determines when to send data to the consumer who is unaware of when the data will arrive.

Observables can produce a single or multiple values and push them to the Observers that is the Consumers.

Anatomy of an Observable

Observables can be generalized as functions with no arguments but can produce multiple values synchronously or asynchronously. Observables can be said to be lazy computations, for example, if you don’t call a function, nothing happens. If you don’t subscribe to an observable, no values are emitted. But keep in mind (in the context of producers) unlike functions that produce a single values and complete, observables can produce multiple values.

Subscribing to an Observable can be viewed to be the same as calling a Function.

Observables (Producers) are created using new Observable or a creation operator, are subscribed to with an Observer (Consumer), execute to deliver next / error / complete notifications to the Observer and then the execution may be disposed. These four aspects are all part of an Observable instance.

Key Observable concepts to understand:

• Creating Observables
• Subscribing to Observables
• Executing the Observable
• Disposing Observables

Creating Observables

The Observable constructor (hint: class instantiation is happening) takes one argument: the subscribe function.

In the following example we create an Observable explicitly which is uncommon but here it is for the purposes of gaining some understanding of what goes on in the plumb works.

import { Observable } from 'rxjs';  
const observable = new Observable ( function subscribe(subscriber){
   subscriber.next('Hello World');
});

When creating the new instance of an Observable we pass a function as a param value. This subscribe function has a subscriber param and inside subscribe we describe (for the sake of understanding) in its internals what happens when we subscribe to the observable aka execute the observable if we think of it as a function that can return a stream of values.

Observable.d.ts

Observables can be created with new Observable. Most commonly, observables are created using creation functions, like of, from, interval, etc.

From an Angular perspective for example, making a http call, the http module will take care of plumbing the internals of what is to be executed and wait for us to subscribe in order to execute the Observable. Creation functions like of, from, interval etc. also do so behind the curtains.

Subscribing to Observables

observable.subscribe(x => console.log(x));

When calling observable.subscribe, the function subscribe in new Observable(function subscribe(subscriber) {...}) is run for that given subscriber. This method .subscribe(Observer) tells the Observable to start emitting values to the Observer (described in the next section) that watches for emissions and notifications from an Observable.

You may note thatobservable.subscribe and subscribe in new Observable(function subscribe(subscriber) {...}) have the same name but are essentially different in terms of what they achieve.

Subscribing to an Observable is like calling a function whilst providing callbacks where the data will be delivered to when the observable is executed.

Executing Observables

Code inside new Observable(function subscribe(subscriber) {...}) is executed for each Observer that subscribes. The execution produces multiple values over time, either synchronously or asynchronously.

An Observer defines an interface with callback functions for each type of Observable notification namely: next, error, and complete.

1. next callback is used to process the emitted item.
2. error callback is used to implement exception handling.
3. complete callback is used to perform cleanup when the Observable is complete.

// Define an explicit observer (uncommon)
const observer = {
  next: (res)=> console.log(`A response ${res} was emitted`),
  error: (err) => console.log(`Error occurred: ${err}`),
  complete: () => console.log(`No more response, yaaaa!`)
};
// Pass the Observer into the subscribe (uncommon)
const sub = observable.subscribe(observer);
//----------------------------------------------------------
// Or you can 
// Pass the next callback function directly as the only function
const sub = observable.subscribe(
   res => console.log(`A response ${res} was emitted`)
);
//----------------------------------------------------------
// Or you can
// Pass an Observer object
const sub = observable.subscribe({
  next: (res)=> console.log(`A response ${res} was emitted`),
  error: (err) => console.log(`Error occurred: ${err}`),
  complete: () => console.log(`No more response, yaaaa!`)
});

Disposing Observables

Observable Executions may be infinite and since each execution is exclusive to one Observer only, once the Observer is done receiving values, it has to have a way to stop the execution in order to avoid what is popularly known memory leaks.

When observable.subscribe is called, the Observer passed to .subscribe gets attached to the new Observable execution. This call returns an object of type Subscription which represents the ongoing execution.

const subscription = observable.subscribe(x => console.log(x));

Using subscription.unsubscribe() you can cancel the ongoing execution.

const subscription = observable.subscribe(x => console.log(x)); 
// Cancel subscription
subscription.unsubscribe();

I hope that helps in understanding Observables and illustrates how RxJS concepts of Observable, Subscription, and Observer work.