Référence : rxjs.dev

• ReactiveX.io
• RxJS est la version JavaScript
• Observable Pattern
• Asynchronous programming

getDataFromLocalMemory()
    .skip(10)
    .take(5)
    .map({ s -> return s + " transformed" })
    .subscribe({ println "onNext => " + it })

Un observable cold est un observable qui émet quand il y a un observeur. Par opposition, un observable hot émet même s'il n'y a pas d'observeur. Avec un observable hot, quand on commence à écouter, on a les valeurs à ce moment, les valeurs passées ne sont pas accessibles.

Exemples de cold observables: Netflix Movie, vidéo Youtube (on-demand), lecture de fichier, requête HTTP, requête à une base de données. Exemple de hot observable: Film dans un cinéma, Youtube Live.

of(1, 2, 3)
  .pipe(count())
    .subscribe(
      val => console.log(val),
      e => console.log(e),
      () => console.log('complete')
    );

Les opérateurs sont des fonctions pures (sauf: tap).

Use case: Obtenir des ensembles de valeurs et les émettre en lots selon les conditions données.

• Buffering: Émettre un tableau de valeurs
• Windowing: Émettre un observable de valeurs

L'opérateur buffer() prend un observable en tant que paramètre. Lorsque cet observable émet, c'est le signal pour émettre les valeurs mis en tampon du premier observable.

Dans l'exemple suivant, des valeurs sont émises par interval(100) à toutes les dixièmes de seconde. Les valeurs sont mis en tampon jusqu'à ce qu'une valeur du second observable interval(1000) soit émise, soit à toutes les secondes.

À noter que si le premier observable se termine, le reste des valeurs qui n'ont pas fait partie d'un buffer sont alors émits dans un dernier buffer.

interval(100)
  .pipe(
    buffer(interval(1000)),
    take(3)
  )
  .subscribe(d => console.log(d));

[1, 2, 3, 4, 5, 6, 7, 8]
[9, 10, 11, 12, 13, 14, 15, 16, 17]
[18, 19, 20, 21, 22, 23, 24, 25, 26]

BufferCount émet la séquence une fois le bufferSize atteint. Le deuxième paramètre startBufferEvery (optionnel) dit quand repartir une nouvelle séquence.

bufferCount<number>(bufferSize: number, startBufferEvery?: number)

of(1,2,3,4,5,6,7,8,9)
  .pipe(
    bufferCount(4)
  )
  .subscribe(sequence => console.log(sequence));

[1,2,3,4]
[5,6,7,8]
[9]

of(1,2,3,4,5,6,7,8,9)
  .pipe(
    bufferCount(4, 2)
  )
  .subscribe(sequence => console.log(sequence));

[1,2,3,4]
[3,4,5,6]
[5,6,7,8]
[7,8,9]
[9]

Les exemples sont prises dans le contexte d'une application Angular, mais devrait convenir à tout autre contexte Javascript.

observable$.pipe(
  combineLatest(anotherObservable$),
  map(([obs1Result, obs2Result]) => {
    // do something with results.
  })
).subscribe();

Différence avec withLatestFrom:

• A.withLatestFrom(B): Utiliser cet opérateur pour obtenir la valeur la plus récente de l'observable B quand A émet. Cela va nous donner les deux valeurs dans un tableau [a, b].
• A.combineLatest(B): Utiliser cet opérateur pour obtenir la valeur la plus récente de l'observable A et B chaque fois que A ou B émet. Cela va nous donner les deux valeurs dans un tableau [a, b].

Exemple de distinct() utilisé dans le contexte de NgRx. Celui-ci permet d'envoyer une seule requête au lieu d'envoyer plusieurs requêtes successives (par exemple par plusieurs clics successifs).

La signature du dictinct() est :

• keySelector: Fonction optionnelle pour dire sur quelle valeur faire un distinct.
• flushes: Observable optionnel qui permet de flusher le HashSet interne de l'opérateur.

@Effect()
requestNewPassword$: Observable<Action> = this.actions$.pipe(
  ofType<RequestResetPassword>(SomeActionTypes.REQUEST_PIZZA_FETCH),
  distinct(
    () => SomeActionTypes.REQUEST_PIZZA_FETCH,
    this.actions$.pipe(
      ofType<RequestResetPassword>(
        SomeActionTypes.REQUEST_PIZZA_SUCCESS,
        SomeActionTypes.REQUEST_PIZZA_ERROR,
        SomeActionTypes.REQUEST_PIZZA_RESET_STATE
      )
    )
  ),
  mergeMap(action =>
    // Call to service for request.
  )
);

Permet de faire comme les .then() avec les promesses, c'est-à-dire faire séquentiellement les opérations.

    this.someService.isReady()
      .pipe(
        takeUntil(this.destroyed$),
        filter(isReady => isReady),
        flatMap(() => this.catService.getCat(1)),
        flatMap(cat => this.foodService.getFood(cat)),
        tap(food => (this.food = food))
      )
      .subscribe();

• rxjs.dev
• Pluralsight Learning RxJS Operators by Example Playbook