Наблюдаемые

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

Observable может доставлять несколько значений любого типа - литералов, сообщений или событий, в зависимости от контекста. API для получения значений одинаков независимо от того, доставляются ли значения синхронно или асинхронно. Поскольку настройка и логика разрыва обрабатываются наблюдаемым, вашему коду приложения нужно беспокоиться только о подписке на использование значений и, когда это будет сделано, отписке. Независимо от того, был ли поток нажатий клавиш, HTTP-ответ или интервальный таймер, интерфейс для прослушивания значений и остановки прослушивания одинаков.

Из-за этих преимуществ наблюдаемые широко используются в Angular и рекомендуются также для разработки приложений.

Основное использование и условия

Как разработчик, вы создаете экземпляр Observable, который определяет функцию подписчика. Это функция, которая выполняется, когда потребитель вызывает метод subscribe (). Функция подписчика определяет, как получить или сгенерировать значения или сообщения для публикации.

Чтобы выполнить созданную вами наблюдаемую область и начать получать уведомления, вы вызываете ее метод subscribe (), передавая наблюдателя. Это объект JavaScript, который определяет обработчики получаемых вами уведомлений. Вызов подписки () возвращает объект Subscription, у которого есть метод unsubscribe (), который вызывается для прекращения получения уведомлений.

Код TypeScript:

// Create an Observable that will start listening to geolocation updates
// when a consumer subscribes.
const locations = new Observable((observer) => {
  // Get the next and error callbacks. These will be passed in when
  // the consumer subscribes.
  const {next, error} = observer;
  let watchId;
  // Simple geolocation API check provides values to publish
  if ('geolocation' in navigator) {
    watchId = navigator.geolocation.watchPosition(next, error);
  } else {
    error('Geolocation not available');
  }
  // When the consumer unsubscribes, clean up data ready for next subscription.
  return {unsubscribe() { navigator.geolocation.clearWatch(watchId); }};
});
// Call subscribe() to start listening for updates.
const locationsSubscription = locations.subscribe({
  next(position) { console.log('Current Position: ', position); },
  error(msg) { console.log('Error Getting Location: ', msg); }
});
// Stop listening for location after 10 seconds
setTimeout(() => { locationsSubscription.unsubscribe(); }, 10000);

Демонстрация в реальном времени:

См. Pen observable_example.ts от w3resource ( @ w3resource ) в CodePen .

Определение наблюдателей

Обработчик для получения наблюдаемых уведомлений реализует интерфейс Observer. Это объект, который определяет методы обратного вызова для обработки трех типов уведомлений, которые может отправлять наблюдаемая:

Объект-наблюдатель может определять любую комбинацию этих обработчиков. Если вы не предоставите обработчик для типа уведомления, наблюдатель игнорирует уведомления этого типа.

Подписавшись

Наблюдаемый экземпляр начинает публиковать значения только тогда, когда кто-то подписывается на него. Вы подписываетесь, вызывая метод subscribe () экземпляра, передавая объект-наблюдатель для получения уведомлений.

Вот пример создания и подписки на простой наблюдаемый, с наблюдателем, который регистрирует полученное сообщение на консоли:

Код TypeScript:

// Create simple observable that emits three values
const myObservable = of(1, 2, 3);
// Create observer object
const myObserver = {
  next: x => console.log('Observer got a next value: ' + x),
  error: err => console.error('Observer got an error: ' + err),
  complete: () => console.log('Observer got a complete notification'),
};
// Execute with the observer object
myObservable.subscribe(myObserver);
// Logs:
// Observer got a next value: 1
// Observer got a next value: 2
// Observer got a next value: 3
// Observer got a complete notification

Демонстрация в реальном времени:

Смотрите Pen subscribingToObservables по w3resource ( @ w3resource ) на CodePen .

Создание наблюдаемых

Мы используем конструктор Observable для создания наблюдаемого потока любого типа. Конструктор принимает в качестве аргумента функцию подписчика, которая запускается при выполнении метода подписки () наблюдаемого объекта. Функция подписчика получает объект Observer и может публиковать значения в методе next () наблюдателя. Пример показан ниже:

Код TypeScript:

// This function runs when subscribe() is called
function sequenceSubscriber(observer) {
  // synchronously deliver 1, 2, and 3, then complete
  observer.next(1);
  observer.next(2);
  observer.next(3);
  observer.complete();
  // unsubscribe function doesn't need to do anything in this
  // because values are delivered synchronously
  return {unsubscribe() {}};
}
// Create a new Observable that will deliver the above sequence
const sequence = new Observable(sequenceSubscriber);
// execute the Observable and print the result of each notification
sequence.subscribe({
  next(num) { console.log(num); },
  complete() { console.log('Finished sequence'); }
});
// Logs:
// 1
// 2
// 3
// Finished sequence

Демонстрация в реальном времени:

Посмотрите перо, создающее_обследуемые w3resource ( @ w3resource ) на CodePen .

Multicasting

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

Иногда вместо того, чтобы запускать независимое выполнение для каждого подписчика, вы хотите, чтобы каждая подписка получала одинаковые значения, даже если значения уже начали излучаться. Это может иметь место с чем-то вроде наблюдаемых кликов на объекте документа.

Многоадресная передача - это практика трансляции списка нескольких подписчиков в одном исполнении. При наличии многоадресной рассылки вы не регистрируете несколько прослушивателей в документе, а вместо этого повторно используете первый прослушиватель и отправляете значения каждому подписчику.

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

В приведенном ниже примере отсчитывается от 1 до 3 с задержкой в одну секунду после каждого испущенного числа.

Код TypeScript:

function sequenceSubscriber(observer) {
  const seq = [1, 2, 3];
  let timeoutId;
  // Will run through an array of numbers, emitting one value
  // per second until it gets to the end of the array.
  function doSequence(arr, idx) {
    timeoutId = setTimeout(() => {
      observer.next(arr[idx]);
      if (idx === arr.length - 1) {
        observer.complete();
      } else {
        doSequence(arr, ++idx);
      }
    }, 1000);
  }
  doSequence(seq, 0);
  // Unsubscribe should clear the timeout to stop execution
  return {unsubscribe() {
    clearTimeout(timeoutId);
  }};
}
// Create a new Observable that will deliver the above sequence
const sequence = new Observable(sequenceSubscriber);
sequence.subscribe({
  next(num) { console.log(num); },
  complete() { console.log('Finished sequence'); }
});
// Logs:
// (at 1 second): 1
// (at 2 seconds): 2
// (at 3 seconds): 3
// (at 3 seconds): Finished sequence

Демонстрация в реальном времени:

См. Pen delayed_sequence от w3resource ( @ w3resource ) в CodePen .

Многоадресная реализация вышеуказанного будет

Код TypeScript:

function multicastSequenceSubscriber() {
  const seq = [1, 2, 3];
  // Keep track of each observer (one for every active subscription)
  const observers = [];
  // Still a single timeoutId because there will only ever be one
  // set of values being generated, multicasted to each subscriber
  let timeoutId;
  // Return the subscriber function (runs when subscribe()
  // function is invoked)
  return (observer) => {
    observers.push(observer);
    // When this is the first subscription, start the sequence
    if (observers.length === 1) {
      timeoutId = doSequence({
        next(val) {
          // Iterate through observers and notify all subscriptions
          observers.forEach(obs => obs.next(val));
        },
        complete() {
          // Notify all complete callbacks
          observers.slice(0).forEach(obs => obs.complete());
        }
      }, seq, 0);
    }
    return {
      unsubscribe() {
        // Remove from the observers array so it's no longer notified
        observers.splice(observers.indexOf(observer), 1);
        // If there's no more listeners, do cleanup
        if (observers.length === 0) {
          clearTimeout(timeoutId);
        }
      }
    };
  };
}
// Run through an array of numbers, emitting one value
// per second until it gets to the end of the array.
function doSequence(observer, arr, idx) {
  return setTimeout(() => {
    observer.next(arr[idx]);
    if (idx === arr.length - 1) {
      observer.complete();
    } else {
      doSequence(observer, arr, ++idx);
    }
  }, 1000);
}
// Create a new Observable that will deliver the above sequence
const multicastSequence = new Observable(multicastSequenceSubscriber());
// Subscribe starts the clock, and begins to emit after 1 second
multicastSequence.subscribe({
  next(num) { console.log('1st subscribe: ' + num); },
  complete() { console.log('1st sequence finished.'); }
});
// After 1 1/2 seconds, subscribe again (should "miss" the first value).
setTimeout(() => {
  multicastSequence.subscribe({
    next(num) { console.log('2nd subscribe: ' + num); },
    complete() { console.log('2nd sequence finished.'); }
  });
}, 1500);
// Logs:
// (at 1 second): 1st subscribe: 1
// (at 2 seconds): 1st subscribe: 2
// (at 2 seconds): 2nd subscribe: 2
// (at 3 seconds): 1st subscribe: 3
// (at 3 seconds): 1st sequence finished
// (at 3 seconds): 2nd subscribe: 3
// (at 3 seconds): 2nd sequence finished

Демонстрация в реальном времени:

Посмотрите многоадресную рассылку Pen от w3resource ( @ w3resource ) на CodePen .

Предыдущая: Динамические формы
Далее: Библиотека RxJS