Today I Learned - Rocky Kev

TIL about RxJS Observables

POSTED ON:

TAGS: javascript

RxJS is a reactive programming library. It makes composing asynchronous and event-based programs very easy by using observable streams. Think of it as an alternative to Promises... An analogy I often use is to imagine a promise that can resolve multiple times.

It provides one core type, the Observable, satellite types (Observer, Subjects) and operators.

An Observable takes in a stream of data and emits multiple bits of data over time.

They don't do much until called upon, aka "subscribed" to. To create an Observable, we create a function, and this function will return the data.

// Import fromEvent method to create Observable
import { fromEvent } from 'rxjs/observable/fromEvent';

// Create an observable with click event
var observable = fromEvent(window, 'click');

let count = 0;

// Subscribe to the observable
observable.subscribe(() => console.log(`Hey! You clicked the window ${++count} times`));

You can also cancel observables.

Real-life Analogy

Let's look at youtube.com for a second. I'll use it as an example to explain this pattern's two core participants: Observer and Observable.

Your browsing on youtube and you find a new channel. You want to receive notifications anytime some new content is published into that youtube channel, but for that, you'll need to hit the subscribe button first. You are the Observer, consuming the content posted (published) by that youtube channel. This makes the youtube channel the Observable. Another vital aspect to see is the multiplicity of this relationship. There's one youtube channel to many youtube users (subscribers).

via https://goodguydaniel.com/blog/reactive-programming-fundamentals#observer-pattern-real-life-analogy

Some key words:

Observable: They deliver 3 types of notifications.

Observers: An observer is something that watches an observable. They get the Observable data and fires off a callback function based on the notification.

Subscription: When we subscribe to an Observable, we get a subscription object in return which can be used to unsubscribe from the observable, to stop receiving any further data events from the observable.

Operators: Operators offer a way to manipulate values from an observable, returning a transformed value for the subscriber or a new observable of transformed values.

import { range } from 'rxjs/observable/range';

// Creating an Observable using 'range'
const observable = range(1, 10)

//Creating an observer
const observer = {
    next: (val:number) => console.log(val),
    error: (err:TypeError) => console.log(err),
    complete: () => console.log('No more data in stream')
}

// Subscribing the observable to get the data from the data stream
observable.subscribe(observer);

// Output :
// 1
// 2
// 3
// 4
// 5
// 6
// 7
// 8
// 9
// 10
// No more data in stream

Use cases

A common example would be handling web socket messages. The socket becomes the observable and each message becomes one of the sequence of values. I also think using observables works quite well in auto complete / type ahead situations.

Why use RxJS over Promises?

So a promise is straightforward.

A JavaScript Promise is an object that produces a single value, asynchronously. Data goes in, and a single value is emitted and used. Easy, straightforward way to handle incoming data.

Promises are eager, meaning that a promise will start doing whatever task you give it as soon as the promise constructor is invoked.

console.log('Start');
const fetchItem = new Promise(resolve => {
  console.log('Inside Promise');
  resolve();
});
console.log('Finished');

// OUTPUT
// Start
// Inside Promise
// Finished

You can avoid that by wrapping your promise in a function, and call your promise when it's go time.

console.log('Start');
function runPromise() {
  const fetchItem = new Promise(resolve => {
    console.log('Inside Promise');
    resolve();
  });
  return fetchItem;
}
console.log('Finished');

// OUTPUT
// Start
// Finished

Observables don't execute their implementation unless they are subscribed.

What is the difference between RxJs and the Observables Pattern?

RxJS is a library it follows the Observables pattern!

The observer pattern is a software design pattern in which an object, named the subject, maintains a list of its dependents, called observers, and notifies them automatically of any state changes, usually by calling one of their methods.

Via Wikipedia

More simply, it is basically an object that keeps track of "observers" that are interested in knowing when something changes and notifies them each time there is a change.

An example of this that you might be familiar with is the event listener pattern used in JS. The element you call addEventListener() on is the "observable", and the observers are the methods that are called when the event is triggered.

In a more formal observable pattern, you'd usually have some interface that you'd have methods on that would get triggered, then pass in that whole observable object, but that's just a technical implementation detail.

This article actually covers how RxJs works under the hood!

So in the Observables Pattern, it follows a Subjects, Actions, and Observers pattern.

tl;dr - Subjects notify the Observer on events. The Observer takes the data/action, and passes it over to the subject.

class Subject {
    constructor() {
        /**
         * The list of threads observed based on each user's instance.
         * this will contain a list of observers.
         */
        this.observers = [];
    }

    isSubscribed(f) {
        /* Help us check if the observer for an user is already subscribed */
        return this.observers.filter(subscriber => subscriber === f).length;
    }

    subscribe(f) {
        /* Verifies that the user is not subscribed already */
        if (this.isSubscribed(f)) return;

        /* If the user is not subscribed adds the function to the list */
        this.observers.push(f);
    }

    unsubscribe(f) {
        /**
         * returns a new array of functions without the one passed as argument,
         * Basically unsubscribing the user from that thread.
         */
        this.observers = this.observers.filter(subscriber => subscriber !== f);
    }

    notify(data) {
        /**
         * notifies the user, it passes the data to each observer
         * set in the list so that it's updated.
         */
        this.observers.forEach(observer => observer.update(data));
    }
}

export default Subject;

References:
https://stackoverflow.com/questions/34497343/redux-rxjs-any-similarities
https://tusharsharma.dev/posts/js-promises-eager-not-lazy
https://auth0.com/blog/javascript-promises-vs-rxjs-observables/
https://enmascript.com/articles/2019/03/09/diving-into-the-great-observer-pattern-in-javascript
https://www.reddit.com/r/learnjavascript/comments/rxid5v/comment/hrie90y/?utm_source=share&utm_medium=web2x&context=3

Related TILs

Tagged: javascript

TIL what is npm Script

Despite their high usage they are not particularly well optimized and add about 400ms of overhead. In this article we were able to bring that down to ~22ms.

TIL fancy methods to transform Javascript Objects

You can use Object.entries(), Object.keys(), Object.fromEntries()...

TIL how to hide your JS code

ONE THING TO NOTE: Encrypting a script is stronger than obfuscation, both methods are still not adequate to protect secret content. Honestly, I don't think it's worth it on a production site, and instead just go with pure server-side if you want security. But it's fascinating.