Rxjs

Creating observable

import { Observable } from 'rxjs'

const observable = new Observable((subscriber) => {
    const id = setInterval(() => { 
        subscriber.next("hi");
    }, 1000);
})

Creating observer

Observer is to subscribe to this, we can do

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

The value of hi will then be populated into this function;

To stop publishing, observable can then call .complete(). For example:

import { Observable } from 'rxjs'

const observable = new Observable((subscriber) => {
    const id = setInterval(() => { 
        subscriber.next("hi");
        subscriber.complete();
    }, 1000);
})

Because of this, the client only see hi and stop.

The observer itself can also be expanded as an object


const observer: Observer<number> = {
  next(x: number) {
    console.log(x);
  },
  error(err: string) {
    console.error(err);
  },
  complete() {
    console.log("done");
  }
}

observable.subscribe(observer)

Without this, we just simply call next

Subscription

Output of .subscribe is Subscription.

import { Observable, type Observer } from "rxjs";

const observable = new Observable<number>((client) => {
  setInterval(() => {
    client.next(1);
  }, 1000)
})


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

setTimeout(() => subscription.unsubscribe(), 5000);

This allows us to unsubscribe and stuff.

To clear interval, we can do something like this

import { Observable } from "rxjs";

const observable = new Observable<number>((client) => {
  const id = setInterval(() => {
    client.next(1);
  }, 1000);

  return () => {
    console.log("clear interval")
    clearInterval(id);
  }
})


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

setTimeout(() => subscription.unsubscribe(), 5000);

the return function of observable is auto-called by rxjs.

Operation

Piping

pipe take 1 observable and return another observable and leave the original observable alone. Normally is used for transforming.

For example:

import { of, first, map } from "rxjs";
of(1, 2, 3, 4).pipe(map((x) => x * 2), first()).subscribe((x) => console.log(x));

of create a new observable, we then feed this into map function. The syntax pipe(op1, op2) is similar to op2(op1()) so the output of map() goes to the input of first.

Creation operator

import { interval } from 'rxjs';

const observable = interval(1000 /* number of milliseconds */);

This is for creating an observer.