Блог

Я пытаюсь создать observable, который выдает значения из ряда асинхронных действий (http-запросы с сервера Jenkins), которые позволят подписчику узнать, как только все действия будут завершены. Я чувствую, что, должно быть, что-то недопонимаю, потому что это не дает того, чего я ожидаю.

 'use strict';

let Rx = require('rx');
let _ = require('lodash');
let values = [
    {'id': 1, 'status': true},
    {'id': 2, 'status': true},
    {'id': 3, 'status': true}
];

function valuesObservable() {

    return Rx.Observable.create(function(observer) {
        _.map(values, function(value) {
            var millisecondsToWait = 1000;
            setTimeout(function() { // just using setTimeout here to construct the example
                console.log("Sending value: ", value);
                observer.onNext(value)
            }, millisecondsToWait);
        });
        console.log("valuesObservable Sending onCompleted");
        observer.onCompleted()
    });
}

let observer = Rx.Observer.create((data) => {
    console.log("Received Data: ", data);
    // do something with the info
}, (error) => {
    console.log("Error: ", error);
}, () => {
    console.log("DONE!");
    // do something else once done
});

valuesObservable().subscribe(observer);
  

Запустив это, я получаю вывод:

 valuesObservable Sending onCompleted
DONE!
Sending value:  { id: 1, status: true }
Sending value:  { id: 2, status: true }
Sending value:  { id: 3, status: true }
  

Хотя я хотел бы видеть что-то более похожее:

 Sending value:  { id: 1, status: true }
Received Data:  { id: 1, status: true }
Sending value:  { id: 2, status: true }
Received Data:  { id: 2, status: true }
Sending value:  { id: 3, status: true }
Received Data:  { id: 3, status: true }
valuesObservable Sending onCompleted
DONE!
  

На самом деле меня не волнует порядок элементов в списке, я бы просто хотел, чтобы наблюдатель их получил.

Я полагаю, что происходит то, что Javascript асинхронно запускает функцию тайм-аута и сразу переходит к observer.onCompleted() строке. Как только подписавшийся наблюдатель получает событие OnCompleted (это правильное слово?), Он решает, что это сделано, и избавляется от себя. Затем, когда асинхронные действия завершаются и наблюдаемый срабатывает onNext , наблюдатель больше не существует, чтобы предпринимать с ними какие-либо действия.

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

Редактировать:

Поскольку я использовал setTimeout для построения своего примера, я понял, что могу использовать его для частичного решения своей проблемы, предоставив наблюдаемому тайм-аут.

 function valuesObservable() {

    return Rx.Observable.create(function(observer) {
        let observableTimeout = 10000;
        setTimeout(function() {
            console.log("valuesObservable Sending onCompleted");
            observer.onCompleted();
        }, observableTimeout);
        _.map(values, function(value) {
            let millisecondsToWait = 1000;
            setTimeout(function() {
                console.log("Sending value: ", value);
                observer.onNext(value)
            }, millisecondsToWait);
        });
    });
}
  

Это дает мне всю информацию из наблюдаемого в нужном мне порядке (данные, затем завершение), но в зависимости от выбора времени ожидания я либо могу пропустить некоторые данные, либо мне придется долго ждать события завершения. Это просто неотъемлемая проблема асинхронного программирования, с которой мне приходится жить?

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

Первый шаг — отказаться от прямого использования setTimeout . Вместо этого используйте timer

 Rx.Observable.timer(waitTime);
  

Затем вы можете преобразовать массив значений в Observable таким образом, чтобы каждое значение выдавалось как событие, выполнив:

 Rx.Observable.from(values);
  

 Rx.Observable.from(values)
  .flatMap(
    // Map the value into a stream
    value => Rx.Observable.timer(waitTime),
    // This function maps the value returned from the timer Observable
    // back into the original value you wanted to emit
    value => value
  )
  

 function valuesObservable(values) {
  return Rx.Observable.from(values)
    .flatMap(
      value => Rx.Observable.timer(waitTime),
      value => value
    )
    .do(
      x => console.log(`Sending value: ${value}`),
      null,
      () => console.log('Sending values completed')
    );
}
  

 Rx.Observable.from(streams)
    .flatMap(stream => stream);

// OR
Rx.Observable.from(streams).merge();

// Or simply
Rx.Observable.mergeAll(streams);