Типичное использование Observable объектов в Angular 4

Представляю вашему вниманию типичные варианты использования Observable объектов в компонентах и сервисах Angular 4.

Подписка на параметр роутера и мапинг на другой Observable

Задача: При открытии страницы example.com/#/users/42, по userId получить данные пользователя.

Решение: При инициализации компоненты UserDetailsComponent мы подписываемся на параметры роутера. То есть если userId будет меняться — будер срабатывать наша подписка. Используя полученный userId, мы из сервиса userService получаем Observable с данными пользователя.

// UserDetailsComponent

ngOnInit() {
  this.route.params
    .pluck('userId') // получаем userId из параметров
    .switchMap(userId => this.userService.getData(userId))
    .subscribe(user => this.user = user);
}

Подписка на параметр роутера и строку запроса

Задача: При открытии страницы example.com/#/users/42?regionId=13 нужно выполнить функцию load(userId, regionId). Где userId мы получаем из роутера, а regionId — из параметров запроса.

Решение: У нас два источника событий, поэтому воспользуемся функцией Observable.combineLatest, которая будет срабатывать, когда каждый из источников генерирует событие.

ngOnInit() {
  Observable.combineLatest(this.route.params, this.route.queryParams)
    .subscribe(([params, queryParams]) => { // полученный массив деструктурируем
      const userId = params['userId'];
      const regionId = queryParams['regionId'];
      this.load(userId, regionId);
    });
}

Обратите внимание, что созданные подписки на роутер, при разрушении объекта удалятся, за этим следит ангуляр, поэтому отписываться от параметров роутера не нужно:

The Router manages the observables it provides and localizes the subscriptions. The subscriptions are cleaned up when the component is destroyed, protecting against memory leaks, so we don't need to unsubscribe from the route params Observable. Mark Rajcok

Остановка анимации загрузки после окончания выполнения подписки

Задача: Показать значок загрузки после начала сохранения данных и скрыть его, когда данные сохранятся или произойдет ошибка.

Решение: За отображение загрузчика у нас отвечает переменная loading, после нажатия на кнопку, установим ее в true. А для установки ее в false воспользуемся Observable.finally функций, которая выполняется после завершения подписки или если произошла ошибка.

save() {
  this.loading = true;
  this.userService.save(params)
    .finally(() => this.loading = false)
    .subscribe(user => {
      // Успешно сохранили
    }, error => {
      // Ошибка сохранения
    });
}

Создание собственного источника событий

Задача: Создать переменную lang$ в configService, на которую другие компоненты будут подписываться и реагировать, когда язык будет меняться.

Решение: Воспользуемся классом BehaviorSubject для создания переменной lang$;

Отличия BehaviorSubject от Subject:

1. BehaviorSubject должен инициализироваться с начальным значением;
2. Подписка возвращает последнее значение Subjectа;
3. Можно получить последнее значение напрямую через функцию getValue().

Создаём переменную lang$ и сразу инициализируем. Так же добавляем функцию setLang для установки языка.

// configService
lang$: BehaviorSubject<Language> = new BehaviorSubject<Language>(DEFAULT_LANG);
setLang(lang: Language) {
  this.lang$.next(this.currentLang); // тут мы поставим
}

Подписываеся на изменение языка в компоненте. Переменная lang$ является "горячим" Observable объектом, то есть подписка требует отписки при разрушении объекта.

private subscriptions: Subscription[] = [];
ngOnInit() {
  const langSub = this.configService.lang$
    .subscribe(() => {
      // ...
    });
  this.subscriptions.push(langSub);
}
ngOnDestroy() {
  this.subscriptions
    .forEach(s => s.unsubscribe());
}

Использование takeUntil для отписки

Отписываться можно и более изящным вариантом, особенно если в компоненте присутствует больше двух подписок:

private ngUnsubscribe: Subject<void> = new Subject<void>();

ngOnInit() {
  this.configService.lang$
    .takeUntil(this.ngUnsubscribe) // отписка по условию
    .subscribe(() => {
      // ...
    });
}

ngOnDestroy() {
  this.ngUnsubscribe.next();
  this.ngUnsubscribe.complete();
}

То есть, чтобы не терять память на горячих подписках, компонента будет работать до тех пор, пока значение ngUnsubscribe не изменится. А изменится оно, когда вызовется ngOnDestroy. Плюсы этого варианта в том, что в каждую из подписок достаточно добавить всего одну строчку, чтобы отписка сработала вовремя.

Использование Observable для автокомплита или поиска

Задача: Показывать предложения страниц при вводе данных на форме

Решение: Подпишемся на изменение данных формы, возьмём только меняющиеся данные инпута, поставим небольшую задержку, чтобы событий не было слишком много и отправим запрос в википедию. Результат выведем в консоль. Интересный момент в том, что switchMap отменит предыдущий запрос, если пришли новые данные. Это очень полезно, для избегания нежалательных эффектов от медленных запросов, если, к например, предпоследний запрос выполнялся 2 секунды, а последий 0.2 секунды, то в консоль выведется результат именно последнего запроса.

ngOnInit() {
  this.form.valueChanges
    .takeUntil(this.ngUnsubscribe)      // отписаться после разрушения
    .map(form => form['search-input'])  // данные инпута
    .distinctUntilChanged()             // брать измененные данные
    .debounceTime(300)                  // реагировать не сразу
    .switchMap(this.wikipediaSearch)    // переключить Observable на запрос в Вики
    .subscribe(data => console.log(data));
}

wikipediaSearch = (text: string) => {
  return Observable
    .ajax('https://api.github.com/search/repositories?q=' + text)
    .map(e => e.response);
}

Кеширование запроса

Задача: Необходимо закешировать Observable запрос

Решение: Воспользуемся связкой publishReplay и refCount. Первая функция закеширует одно значение функции на 2 секунды, а вторая будет считать созданные подписки. То есть, Observable завершится, когда все подписки будут выполнены. Тут можно прочитать подробнее.

// tagService

private tagsCache$ = this.getTags()
  .publishReplay(1, 2000) // кешируем одно значение на 2 секунды
  .refCount()             // считаем ссылки
  .take(1);               // берем 1 значение

getCachedTags() {
  return tagsCache$;
}

Последовательный combineLatest

Задача: Критическая ситуация на сервере! Backend команда сообщила, что для корректного обновления продукта нужно выполнять строго последовательно:

1. Обновление данных продукта (заголовок и описание);
2. Обновление списка тегов продукта;
3. Обновление списка категорий продукта;

Решение: У нас есть 3 Observable, полученных из productService. Воспользуемся concatMap:

const updateProduct$ = this.productService.update(product);
const updateTags$ = this.productService.updateTags(productId, tagList);
const updateCategories$ = this.productService.updateCategories(productId, categoryList);

Observable
  .from([updateProduct$, updateTags$, updateCategories$])
  .concatMap(a => a)  // выполняем обновление последовательно
  .toArray()          // Возвращает массив из последовательности
  .subscribe(res => console.log(res)); // res содержит массив результатов запросов

Загадка на посошок

Если у вас есть желание немного поупражняться, решите предыдущую задачу, но для создания продукта. То есть сначала создаём продукт, потом обновляем теги, а только потом — категории.

Полезные ссылки

• Заворожённо посмотреть на шарики: rxviz.com
• Потаскать шарики мышкой: rxmarbles.com