Подождите, пока будет выполняться операция async в onNext из RxJS Observable

Ответ 1

Каждая операция, которую вы хотите выполнить, может быть смоделирована как наблюдаемая. Даже синхронную операцию можно моделировать таким образом. Затем вы можете использовать map чтобы преобразовать свою последовательность в последовательность последовательностей, а затем использовать concatAll чтобы сгладить последовательность.

someObservable
    .map(function (item) {
        if (item === "do-something-async") {
            // create an Observable that will do the async action when it is subscribed
            // return Rx.Observable.timer(5000);

            // or maybe an ajax call?  Use 'defer' so that the call does not
            // start until concatAll() actually subscribes.
            return Rx.Observable.defer(function () { return Rx.Observable.ajaxAsObservable(...); });
        }
        else {
            // do something synchronous but model it as an async operation (using Observable.return)
            // Use defer so that the sync operation is not carried out until
            // concatAll() reaches this item.
            return Rx.Observable.defer(function () {
                return Rx.Observable.return(someSyncAction(item));
            });
        }
    })
    .concatAll() // consume each inner observable in sequence
    .subscribe(function (result) {
    }, function (error) {
        console.log("error", error);
    }, function () {
        console.log("complete");
    });

Чтобы ответить на некоторые ваши комментарии... в какой-то момент вам нужно заставить некоторые ожидания от потока функций. В большинстве языков, когда речь идет о функциях, которые возможно асинхронны, сигнатуры функций являются асинхронными, и фактический асинхронный и синхронный характер функции скрывается как деталь реализации функции. Это верно, используете ли вы javaScript-обещания, Rx-наблюдаемые, С# Tasks, c++ Futures и т.д. Функции возвращают обещание/наблюдаемое /task/future/etc и если функция фактически синхронна, тогда объект it возврат уже завершен.

Сказав это, поскольку это JavaScript, вы можете обмануть:

var makeObservable = function (func) {
    return Rx.Observable.defer(function () {
        // execute the function and then examine the returned value.
        // if the returned value is *not* an Rx.Observable, then
        // wrap it using Observable.return
        var result = func();
        return result instanceof Rx.Observable ? result: Rx.Observable.return(result);
    });
}

someObservable
    .map(makeObservable)
    .concatAll()
    .subscribe(function (result) {
    }, function (error) {
        console.log("error", error);
    }, function () {
        console.log("complete");
    });

Ответ 2

Прежде всего, удалите асинхронные операции из subscribe, это не сделано для асинхронных операций.

Вы можете использовать mergeMap (псевдоним flatMap) или concatMap. (Я упоминаю оба из них, но concatMap на самом деле является mergeMap с параметром concurrent mergeMap установленным в 1.) Установка другого параметра параллельной работы полезна, так как иногда вам может потребоваться ограничить количество одновременных запросов, но при этом все же выполнять пару одновременных.

source.concatMap(item => {
  if (item == 'do-something-async-and-wait-for-completion') {
    return Rx.Observable.timer(5000)
      .mapTo(item)
      .do(e => console.log('okay, we can continue'));
    } else {
      // do something synchronously and keep on going immediately
      return Rx.Observable.of(item)
        .do(e => console.log('ready to go!!!'));
    }
}).subscribe();

Я также покажу, как вы можете оценить ограничение ваших звонков. Совет: ограничение скорости только в том месте, где оно действительно необходимо, например, при вызове внешнего API, который разрешает только определенное количество запросов в секунду или минуты. В противном случае лучше просто ограничить количество одновременных операций и позволить системе двигаться с максимальной скоростью.

Начнем со следующего фрагмента:

const concurrent;
const delay;
source.mergeMap(item =>
  selector(item, delay)
, concurrent)

Далее нам нужно выбрать значения для concurrent, delay и selector реализации. concurrent и delay тесно связаны. Например, если мы хотим запустить 10 элементов в секунду, мы можем использовать concurrent = 10 и delay = 1000 (миллисекунда), но также concurrent = 5 и delay = 500 или concurrent = 4 и delay = 400. Количество элементов в секунду всегда будет concurrent/(delay/1000).

Теперь давайте реализуем selector. У нас есть пара вариантов. Мы можем установить минимальное время выполнения для selector, мы можем добавить к нему постоянную задержку, мы можем выдавать результаты, как только они будут доступны, мы можем выдавать результат только после того, как минимальная задержка прошла и т.д. Это даже возможно. добавить время ожидания с помощью операторов времени timeout. Удобство.

Установите минимальное время, отправьте результат раньше:

function selector(item, delay) {
   return Rx.Observable.of(item)
     .delay(1000) // replace this with your actual call.
     .merge(Rx.Observable.timer(delay).ignoreElements())
}

Установить минимальное время, отправить результат поздно:

function selector(item, delay) {
   return Rx.Observable.of(item)
     .delay(1000) // replace this with your actual call.
     .zip(Rx.Observable.timer(delay), (item, _))
}

Добавьте время, отправьте результат раньше:

function selector(item, delay) {
   return Rx.Observable.of(item)
     .delay(1000) // replace this with your actual call.
     .concat(Rx.Observable.timer(delay).ignoreElements())
}

Добавить время, отправить результат поздно:

function selector(item, delay) {
   return Rx.Observable.of(item)
     .delay(1000) // replace this with your actual call.
     .delay(delay)
}

Ответ 3

Еще один простой пример для ручных операций async.

Имейте в виду, что это не хорошая реактивная практика! Если вы хотите только подождать 1000 мс, используйте оператор Rx.Observable.timer или delay.

someObservable.flatMap(response => {
  return Rx.Observable.create(observer => {
    setTimeout(() => {
      observer.next('the returned value')
      observer.complete()
    }, 1000)
  })
}).subscribe()

Теперь замените setTimeout вашей асинхронной функцией, например Image.onload или fileReader.onload...