RxJS — доступно, в деталях и на практике

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

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

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

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

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

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

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

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

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

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

Подписки не зависят друг от друга.

Дальше — 127+ операторов. 127, Карл!

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

            Rx.Observable.of('foo');
            // ---foo---|
            

             
            Rx.Observable.from(['foo', 'bar', 'baz']);
            // ---foo---bar---baz---|
            
            

             
            Rx.Observable.timer(100, 500);
            // -0---1---2---3--...
            
        

            const promise = new Promise((resolve) => { resolve('foo')});
            Rx.Observable.fromPromise(promise);
            // ----------------foo---|
            

             
            const btnElement = document.getElementById('btn');
                Rx.Observable.fromEvent(btnElement, 'click');
            // --evt-evt------evt----evt--...
            
        

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

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

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

            Rx.Observable.timer(0, 1500)
                .take(3)
                .map((value) => 'abcd'[value])
            

                .mergeMap(value => {
            

                    return Rx.Observable.timer(0, 450).take(6)
                        .map(innerValue => value + innerValue);
            
                })
                .subscribe(renderNext);
            
        

            const getPromise = (value, time) => new Promise((resolve) => {
                setTimeout(() => {
                    resolve(value);
                }, time);
            });
             
            

            Rx.Observable.fromPromise(getPromise('⏰', 1500))
                .subscribe(renderNext);
            
        

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

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

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

            const request$ = Rx.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$
                .map(...)
                .switchMap(...)
                .subscribe(...);
            

            ...
            onDestroy() {
                subscription.unsubscribe();
            }
            
        

            const destroyStream = new Subject();
            

            streamA$
                .map(...)
                .switchMap(...)
                .takeUntil(destroyStream)
                .subscribe()
            ...
            
            

            onDestroy() {
                destroyStream.next();
            }
            
        

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

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

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

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

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

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

            let cachingSubject;
             
            getResource() {
            

                if (!cachingSubject) {
            

                  cachingSubject = new ReplaySubject(1);
            
            

             
                  loadData().subscribe((result) => {
                      this.cachingSubject.next(result);
                    });
            
                }
            
            

             
                return this.cachingSubject.asObservable();
            
            }
        

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