RxJS: Operator combineLatest

Do czego służy combineLatest?

Operator combineLatest przyjmuje jako parametr listę obiektów typu Observable. Jako wynik zwraca wartości wysłane przez te obiekty.

Żeby combineLatest zwrócił wynik, musi zostać spełniony jeden warunek: wszystkie przekazane do operatora obiekty Observable muszą wysłać swoje dane. Na przykład: przekazujemy do combineLatest dwa strumienie danych. Jeden strumień wysłał dane, ale drugi jeszcze nie. W tym przypadku combineLatest nie zwróci żadnych danych (patrz przykład 1).

Dodatkowo combineLatest zwróci kolejny raz dane, jeśli któryś ze strumieni także wysłał nową porcję danych (patrzy przykład 2). W tym przypadku wystarczy, że tylko jeden obiekt Observable wyemituje dane.

Przykłady użycia combineLatest

Przykład 1

import { of, combineLatest} from 'rxjs';
import { delay } from 'rxjs/operators';

combineLatest([
    of(1),
    of(2).pipe(delay(1000)),
    of(3).pipe(delay(2000)),
]).subscribe(console.log); // [1, 2, 3]

Jako wynik zwróci: [1, 2, 3] ale po 3 sekundach. Zauważ, że pierwszy strumień zwraca dane od razu, drugi po sekundzie, w końcu trzeci po dwóch sekundach. combineLatest wysyła dane, gdy wszystkie przekazane do niego obiekty Observable wyślą swoje dane.

Przykład 2

W HTML dodaj dwa input'y:

<input type="text" id="field1" />
<input type="text" id="field2" />

Kod:

import { fromEvent, combineLatest} from 'rxjs';
import { map } from 'rxjs/operators';

function createKeyupListener(inputId) {
    return fromEvent(document.getElementById(inputId), 'keyup').pipe(
        map((event) => event.target.value)
    );
}
const field1$ = createKeyupListener('field1');
const field2$ = createKeyupListener('field2');

combineLatest([
    field1$,
    field2$
]).pipe(
    map(([a, b]) => {
        return parseFloat(a) + parseFloat(b)
    })
).subscribe(console.log);

Gdy wpiszemy w jedno pole wartość nic się nie stanie. Wprowadzając dane do drugiego pola combineLatest wyemituje sygnał z przetworzonymi danymi. Jeśli ponownie wprowadzisz dane do któregoś pola to combineLatest od razu wyśle nowy wynik. Czyli tylko za pierwszym razem combineLatest czeka, aż wszystkie przekazane do niego strumienie wyślą dane. Później wystarczy, że nowe dane zostaną wyemitowane przez jeden ze strumieni.