Higher-Order Observable Mapping w RxJS

Observable oraz strumień danych

Jednymi z najważniejszych elementów biblioteki RxJS są Observable oraz strumienie danych.

Oba te pojęcia są podobne, jednak występują między nimi subtelne różnice:

Podsumowując, Observable mają szerszy zakres zastosowań niż strumienie danych i mają dodatkowe funkcje, pozwalające na kontrolowanie czasu przepływu wartości i sterowanie błędami.

Operatory w RxJS

Prawdziwą potęgę biblioteki RxJS stanowią operatory. Operatory pozwalają przekształcać, filtrować, łączyć strumienie danych i zarządzać nimi. RxJS dostarcza szereg operatorów, takich jak: map, filter, reduce, debounceTime, distinctUntilChanged itp., które pozwalają na przetwarzanie danych w strumieniach. Wybór odpowiedniego operatora jest ważny, gdyż źle dobrany operator może wpłynąć na funkcjonowanie systemu. Niżej omawiam najważniejsze z nich.

Higher-Order Observable Mapping

Biblioteka RxJS zawiera szczególny typ operatorów, tzw. higher-order operators, czyli operatory wyższego rzędu, które pozwalają na złożone manipulowanie strumieniami danych. Higher-order operators to operatory, które zawierają w argumentach inne operatory lub Observable i zwracają nowe Observable mające zastosowanie do pierwotnego strumienia danych.

Istnienie higher-order operators w bibliotece RxJS jest uzasadnione z kilku powodów:

• Zapewniają one prostsze i bardziej czytelne sposoby manipulowania strumieniami danych. Dzięki nim można złożyć wiele operacji na strumieniu w jeden łańcuch, co ułatwia czytanie i zrozumienie kodu.
• Umożliwiają bardziej elastyczne manipulowanie strumieniami danych. Wiele higher-order operators pozwala na dynamiczne tworzenie nowych strumieni danych w zależności od zdarzeń zachodzących w pierwotnym strumieniu. Dzięki temu można łatwo dostosować przetwarzanie danych do zmieniających się wymagań.
• Higher-order operators pozwalają na wielokrotne wykorzystanie kodu. Wiele operacji na strumieniach danych jest podobnych i można je zastosować w wielu miejscach w aplikacji. Higher-order operators ułatwiają użycie tych samych operacji w różnych częściach kodu, co pozwala na uniknięcie powtórzeń.

Podsumowując, higher-order operators w bibliotece RxJS pozwalają na bardziej elastyczne, czytelne i wielokrotne wykorzystanie kodu do manipulowania strumieniami danych. Dzięki nim programiści mogą złożyć wiele operacji na strumieniu w jeden łańcuch, co ułatwia zarządzanie danymi w aplikacji.

RxJS udostępnia nam cztery operatory wyższego rzędu:

SwitchMap

Operator switchMap emituje wartości wewnętrznego Observable, ale każda nowa wartość emitowana przez strumień źródłowy powoduje anulowanie poprzedniego wewnętrznego Observable i utworzenie nowego. Wewnętrzny Observable zwrócony z funkcji mapującej jest zastępowany nowym wewnętrznym Observable, co oznacza, że jeśli źródłowy strumień emituje wartości z częstotliwością wyższą niż wewnętrzny Observable, operator switchMap przerywa działanie i zastępuje poprzedni wewnętrzny Observable. SwitchMap możemy zastosować w przypadku, gdy chcemy odwołać wcześniejsze żądania i wysłać tylko najnowsze żądanie, np. dotyczące filtrowania wyników wyszukiwania.

import { fromEvent } from 'rxjs'; 

import { switchMap } from 'rxjs/operators'; 

 

Pobieranie danych z serwera 

function fetchData(keyword: string) { 

  return fetch(`https://api.github.com/search/repositories?q=${keyword}`).then(res => res.json()); 

} 
 

Strumień wejściowy z danymi z pola tekstowego

const input$ = fromEvent(document.querySelector('input'), 'input'); 

Operator switchMap kończy subskrypcję poprzedniego strumienia i zwraca tylko strumień wyjściowy ostatniego wyemitowanego strumienia

const result$ = input$.pipe( 

  switchMap(event => fetchData(event.target.value)) 

); 

Subskrypcja strumienia wyjściowego

result$.subscribe( 

  data => console.log(data), 

  err => console.error(err), 

  () => console.log('Complete') 

); 

W tym przykładzie operator switchMap przekształca każdą wartość strumienia wejściowego (event z pola tekstowego) na strumień wyjściowy, który pobiera dane z serwera GitHub API na podstawie frazy wyszukiwania. Dzięki temu, kiedy użytkownik wprowadza nową frazę, operator switchMap anuluje subskrypcję poprzedniego strumienia wyjściowego i zwraca tylko wynik dla ostatniej frazy.

MergeMap

Operator mergeMap przetwarza wszystkie wartości w strumieniu źródłowym jednocześnie, emitując wartości wewnętrzne Observable. Operator tworzy wiele wewnętrznych Observables, co oznacza, że wynikowy strumień wyjściowy może emitować wartości w dowolnej kolejności. Przykładowym zastosowaniem operatora mergeMap może być równoległe pobieranie danych z wielu źródeł.

Operator mergeMap jest szczególnie przydatny w sytuacjach, gdy potrzebujemy wykonać operację asynchroniczną dla każdego elementu strumienia wejściowego i uzyskać strumień wyjściowy składający się z wyników tych operacji.

import { from, Observable } from 'rxjs'; 

import { mergeMap } from 'rxjs/operators'; 

Definicja funkcji zwracającej strumień asynchroniczny

function makeHttpRequest(id: number): Observable {

return fetch(https://jsonplaceholder.typicode.com/posts/${id}).then(res => res.json());

}

Strumień wejściowy z listą identyfikatorów

const ids$ = from([1, 2, 3, 4, 5]); 

 

Operator mergeMap przekształca każdy element strumienia wejściowego na strumień wyjściowy

const posts$ = ids$.pipe( 

  mergeMap(id => makeHttpRequest(id)) 

); 

Subskrypcja strumienia wyjściowego

posts$.subscribe( 

  post => console.log(post), 

  err => console.error(err), 

  () => console.log('Complete') 

); 

W tym przykładzie operator mergeMap przekształca każdy element strumienia wejściowego (identyfikator posta) na strumień wyjściowy reprezentujący wynik asynchronicznej operacji pobierania danych z serwera dla danego posta. W efekcie otrzymujemy strumień wyjściowy składający się z pobranych postów, a nie z identyfikatorów.

ContactMap

Operator concatMap emituje wartości wewnętrznego Observable w kolejności, w jakiej są one dostarczane. Operator nie tworzy wielu wewnętrznych Observables, dopóki poprzedni Observable nie zakończy działania. W przypadku, gdy w strumieniu źródłowym pojawi się kolejna wartość, zanim zakończy działanie poprzedni wewnętrzny Observable. Operator concatMap umieszcza wartości kolejnego wewnętrznego Observable w kolejce i przetwarza je dopiero po zakończeniu poprzedniego Observable.

import { of } from 'rxjs'; 

import { concatMap, delay } from 'rxjs/operators'; 

Strumień wejściowy z liczbami

const source$ = of(1, 2, 3, 4, 5); 

Operator concatMap przekształca każdy element strumienia wejściowego na nowy strumień wyjściowy z opóźnieniem w zależności od wartości elementu

const example$ = source$.pipe( 

  concatMap(value => of(`Delayed by: ${value * 1000}ms`).pipe(delay(value * 1000))) 

); 

Subskrypcja strumienia wyjściowego

example$.subscribe(console.log); 

W powyższym przykładzie concatMap przekształca każdą wartość strumienia wejściowego (1, 2, 3, 4, 5) na strumień wyjściowy, który zostaje opóźniony o wartość elementu w sekundach i zwrócony w kolejności ich pojawiania się w źródłowym strumieniu. Dzięki temu pierwszy element (1) zostanie zwrócony po jednej sekundzie, drugi element (2) po dwóch sekundach itd.

ExhaustMap

Operator exhaustMap przekształca każdy element strumienia wejściowego (źródłowego) na nowy strumień i ignoruje wszystkie kolejne elementy, zanim zakończy działanie strumień wyjściowy.

ExhaustMap jest przydatny, gdy jest nam potrzebne, żeby w danym czasie był emitowany tylko jeden strumień wyjściowy. W przypadku, gdy pojawiają się nowe elementy w strumieniu wejściowym przed zakończeniem strumienia wyjściowego, są one ignorowane aż do zakończenia działania poprzedniego strumienia wyjściowego.

Przykładowym zastosowaniem operatora exhaustMap może być ograniczenie liczby żądań sieciowych w sytuacji, gdy w aplikacji użytkownik kliknie wiele razy na przycisk, ale zwrócone zostanie tylko jedno zapytanie.

 

import { interval } from 'rxjs'; 

import { exhaustMap, take } from 'rxjs/operators'; 

Strumień wejściowy z wartościami

const source$ = interval(1000).pipe(take(4)); 

Operator exhaustMap przekształca każdą wartość strumienia wejściowego na nowy strumień wyjściowy, który zostaje zwrócony dopiero po zakończeniu działania jego poprzednika. 

const example$ = source$.pipe( 

  exhaustMap(value => interval(500).pipe(take(3))) 

); 

Subskrypcja strumienia wyjściowego 

example$.subscribe(console.log); 

Powyższy przykład obrazuje sytuacje, gdy operator exhaustMap przekształca każdą wartość strumienia wejściowego (0, 1, 2, 3) na strumień wyjściowy emitujący wartości z opóźnieniem 0,5 sekundy. Każdy kolejny strumień wyjściowy zostanie zignorowany, dopóki poprzedni nie zostanie zakończony, dlatego drugi strumień wyjściowy z wartościami (0, 1, 2) nie zostanie wyemitowany, ponieważ pierwszy strumień z wartościami (0, 1, 2) nadal jest w trakcie emisji.

Podsumowując:

• exhaustMap ignoruje nowe zdarzenia, jeśli obecny strumień wciąż trwa. Jeśli w trakcie trwania strumienia wejściowego operator otrzyma kolejne zdarzenie, to zostanie ono pominięte, aż do momentu, gdy strumień wejściowy się zakończy.
• mergeMap przetwarza wszystkie wartości w strumieniu źródłowym jednocześnie, emitując wartości wewnętrznych Observables.
• concatMap emituje wartości wewnętrznego Observable w kolejności, w jakiej są one emitowane.
• switchMap emituje wartości wewnętrznego Observable, ale zastępuje poprzedni wewnętrzny Observable nowym, gdy źródłowy strumień emituje nowe wartości.

FAQ:  

Styled components oraz useContext

Użycie hooków w nowoczesnych aplikacjach frontendowych

Przeczytaj artykuł