RxJS: введение

• Реактивный подход
• Observable
• Операторы Rx
• Демо
• Subjects

            from([{x: 11, y: 99}, {x: 91, y: 151}, {x: 152, y: 106}]).pipe(
                map((value) => { ... }),
                filter((value) => { ... }),
                reduce((acc, curr) => { ... })
            

                ).subscribe(result => {
                console.log(result);
            });
            
        

Субъект изменяет значения.
Также называется Observable, наблюдаемый объект.

Наблюдатель (Observer) реагирует на изменения.

• последовательный обход составных объектов
• не раскрывается внутреннее представление

Observable – наблюдаемый объект, доставляет изменения в подписку.

Observer – объект-наблюдатель,
опциональные обработчики next/error/complete

Функция создания определяет,
как наблюдатель будет получать значения.

Subscribtion – подписка.
Старт получения значений.

unsubscribe – завершение получения значений.

Нет подписки — нет потока

Каждый observer создаст свой поток значений
Подписки не зависят друг от друга

• cоздание
• преобразование одного потока
• комбинирование потоков
• Observable высшего порядка

            import {of} from 'rxjs';
            of('foo');
            // ---foo-|
            

             
            import {from} from 'rxjs';
            from(['foo', 'bar', 'baz']);
            // ---foo---bar---baz-|
            
            

             
            import {timer} from 'rxjs';
            timer(100, 500);
            // -0---1---2---3--...
            
        

            import {from} from 'rxjs';
            const promise = new Promise((resolve) => { resolve('foo')});
            from(promise);
            // ----------------foo-|
            

             
            import {fromEvent} from 'rxjs';
            const btnElement = document.getElementById('btn');
            fromEvent(btnElement, 'click');
            // --evt-evt------evt----evt--...
            
        

            .pipe(
        

            ).subscribe(...)
        

            timer(0, 250).pipe(
                take(40),
                filter(() => Math.random() < 0.3),
                debounceTime(500)
            ).subscribe(renderNext);
        

            const streamA$ = timer(0, 1500).pipe(take(4));
            const streamB$ = timer(800, 1000).pipe(
                take(6),
                map((i) => 'abcdef'[i])
            );
            merge(streamA$, streamB$)
                .subscribe(renderNext);
        

            const streamA$ = timer(0, 1500).pipe(take(4));
            const streamB$ = timer(800, 1000).pipe(
                take(6),
                map((i) => 'abcdef'[i])
            );
            combineLatest(streamA$, streamB$).pipe(
                map(([a, b]) => a + b)
            ).subscribe(renderNext);
        

            timer(0, 1500).pipe(
                take(3),
                map((i) => 'abcd'[i]),
                mergeMap(letter => {
                    return timer(0, 625).pipe(
                        take(6),
                        map(digit => letter + digit));
                })
            ).subscribe(renderNext);
        

            const getPromise = (value) => new Promise((resolve) => {
                setTimeout(() => {
                    resolve(value);
                }, 1500);
            });
             
            from(getPromise('⏰'))
                .subscribe(renderNext);
        

SwitchMap автоматически отписывает внутренний поток — вызывает subscription.unsubscribe()

Можем написать отменяемые запросы:

• Angular HttpClient
• Fetch + AbortController
• XMLHttpRequest -> xhr.abort()

            const request$ = Observable.create((observer) => {
            

                const controller = new AbortController();
                const signal = controller.signal;
            
            

             
                fetch(url, {signal}).then(res => res.json())
            
            

                    .then(data => {
                            observer.next(data);
                            observer.complete();
                        });
            
            

             
                return function unsubscribe() {
                    controller.abort();
                };
            });
            
        

Работаем с потоками,
а не с промежуточными состояниями

• видна вся последовательность обработки
• минимум сайд эффектов
• самодостаточный компонент

Объединение Observable и Observer.
Передает значение всем подписчикам

            const subscription = streamA$.pipe(
                map(...),
                switchMap(...)
            ).subscribe(...);
            

            ...
            onDestroy() {
                    subscription.unsubscribe();
                    subscription2.unsubscribe();
                    subscription3.unsubscribe();
                    subscription4.unsubscribe();
                

                    ...
                }
                
            
        

            const destroy$ = new Subject();
            

            streamA$.pipe(
                map(...),
                switchMap(...),
                takeUntil(destroy$)
            ).subscribe()
            ...
            
            

            onDestroy() {
                destroy$.next();
            }
            
        

            class StateService {
              private authSubject = new BehaviorSubject<boolean>(false);
            

             
              setAuthState(state: boolean) {
                this.authSubject.next(state);
              }
            }
            
        

            class StateService {
              private authSubject = new BehaviorSubject<boolean>(false);
              setAuthState(state: boolean) {...}
            

             
              getStateChange(): Observable<boolean> {
                return this.authSubject.asObservable();
              }
            }
            
        

            class StateService {
              private authSubject = new BehaviorSubject<boolean>(false);
              setAuthState(state: boolean) {...}
              getStateChange(): Observable<boolean> {...}
             
            

              getCurrentState(): boolean {
                return this.authSubject.getValue();
              }
            }
            
        

• Push / pull
• Observable = Observer + Iterator
• Observable - паттерн для работы с потоками данных
• Операторы: преобразования, комбинирования потоков
• Observable высшего порядка
• Как сделать отменяемые запросы
• Поле поиска через цепочку операторов
• Subjects – один поток на множество подписчиков