RxJs операторы mergeScan не могут накапливаться потоком flatMap из Promise?

Question

Я хочу накапливать поток из обещания flatMap, я обнаружил, что имена создателей «mergeScan» могут сгруппировать поток и накапливать их。 Но когда я попробовал это, я обнаружил, что это не работает, когда обещание в потоке, но это работает правильно без обещания.

    const Rx = require('rxjs/Rx');
    const {
      of ,
    } = require('rxjs');

    const click$ = new Rx.Subject();
    const seed = 0;
    const testPromise = new Promise((resolve) => {
      resolve(1);
    })
    const count$ = click$.mergeScan((acc, promise) => of (promise)
      .flatMap(promise => promise)
      .map((one) => {
        console.log('acc', acc);
        return acc + one
      }), seed);
    count$.subscribe(x => console.log('value',x)); 

    click$.next(testPromise);
    click$.next(testPromise);
    click$.next(testPromise);

Этот вывод ниже, а аккумулятор не увеличился:

acc, 0
value, 1
acc, 0
value, 1
acc, 0
value, 1

но когда я пишу код ниже:

const Rx = require('rxjs/Rx');
const {
  of ,
} = require('rxjs');

const click$ = new Rx.Subject();
const seed = 0;
const count$ = click$.mergeScan((acc, one) => of (one)
  .map((one) => {
    console.log('acc', acc);
    return acc + one
  }), seed);
count$.subscribe(x => console.log('value', x));

click$.next(1);
click$.next(1);
click$.next(1);

Вывод:

acc, 0
value, 1
acc, 1
value, 2
acc, 2
value, 3

Это работа правильно!

Не знаю почему и как я могу это исправить?

Answer 1

Вся проблема в том, что Обещания разрешаются асинхронно, но когда вы вызываете click$.next(testPromise) несколько раз, все это строго синхронно, поэтому, когда функция проекции для mergeScan вызывается acc, все равно 0 (потому что ни одно обещание не имеетеще не решен).

Вы можете увидеть, как это будет работать, как вы ожидаете, сделав next() вызовы асинхронными:

setTimeout(() => click$.next(testPromise));