RxJS - Introduction

 

What are Streams?

"Stream" of values (zero to infinite)

Ex:

 

// Emit multiple values and never complete

document.addEventListener('click', () => {

    console.log('click');

  });

  // Emit multiple values and never complete

  setInterval(() => {

  console.log('setInterval');

  }, 1000);

  // Emit one value and complete

  setTimeout(() => {

    console.log('setTimout(3000)');

  }, 3000);

Why using RxJS?

To solve the callback hell issues (nested callback functions)

Ex:

 

  // Add 'click' event

  document.addEventListener('click', () => {

    console.log('click');

  // Counting by every second

  let counter = 0;

    const counterInterval = setInterval(

      () => {

        console.log(counter);

        counter++;

      }, 1000

    );

    // Clear counting interval after 3 seconds

    setTimeout(

() => {

      console.log('done');

      clearInterval(counterInterval);

      }, 3000);

    });

}

You may notice that the code above is so nested, and it will not scale well in complexity in the future.

Using RXJS(with Operators), it can combine multiple stream of values together in maintainable way.

How to use?

1. Define the stream.

 

const interval$ = interval(1000);

2. Subscribe (with three callbacks).

 

  const sub = interval$.subscribe(

    val => console.log(val),

    err => console.log(err),

    () => console.log('finished'),

  );

3. Unsubscribe (cancel the observable executions).

  sub.unsubscribe();

NOTE:

* err and complete are mutual exclusive

* once there is an error or it turn into the complete stage, then the observable will not emit any value anymore, and it is called Observable Contract.

Refactoring the previous example to solve 'callback hell'

Ex:

  fromEvent(document, 'click')

    .pipe(

      debounceTime(400),

      tap(() => console.log('click')),

      concatMap(() => interval(1000)),

      take(3),

      finalize(() => console.log('done')),

    )

    .subscribe(console.log);