Снижение RxJS не продолжается

Question

Почему flatMap не вызывает срабатывания сокращений вниз по течению?

Я получил код как:

handleFiles.flatMap(files =>
  Rx.Observable.from(files).
  flatMap((file, i) => fileReader(file, i)).
  reduce((form, file, i) => {
    form.append('file[' + i + ']', result);
    console.log('reduce step', file);
    return form;
  }, new FormData()).
  tap(console.log.bind(console, 'after reduce'))
).
subscribe(console.log.bind(console, 'response'));

И проблема в том, что нажатие "после уменьшения" никогда не срабатывает. Зачем?

Журнал похож на:

reduce step [data]
reduce step [data]

Скриншот:

Answer 1

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

Answer 2

Проблема не в flatMap; это в пути reduce работает.

reduce считывает весь поток и сводит его к одному значению, генерируемому только тогда, когда исходный поток закрыт. Если твой from(files) поток не заканчивается reduce никогда не будет выводить его значение.

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

Answer 3

Вот пример использования reduce с наблюдаемым без завершения;

используя windowTime:

import { fromEvent, interval, timer } from 'rxjs';
import { reduce, filter, windowTime, map, mergeMap } from 'rxjs/operators';

const interval$ = interval(1000);
const observable = interval$.pipe(
  windowTime(2000), // each window is 2s
  mergeMap(window$ => window$.pipe(
    reduce((a,x) => { // reduce to array
      return [...a,x];
    }, []),
    filter(x => !!x.length) // in the background timer is still running so suppress empty events
  )), // flatten the Observable-of-Observables
);
const subscription = observable.subscribe(x => console.log(x));
setTimeout(() => subscription.unsubscribe(), 10000);

используя bufferTime:

import { fromEvent } from 'rxjs';
import { bufferTime, filter, map } from 'rxjs/operators';

let count = 1;
const clicks = fromEvent(document, 'click');
const observable = clicks.pipe(
  bufferTime(1000), // batch into array every 1s
  filter(x => !!x.length), // ignore events without clicks
  map(x => x.reduce((a,y) => ({...a, [count++]: y}), {})),
);
observable.subscribe(x => console.log(x));

используя auditTime:

import { fromEvent } from 'rxjs';
import { tap, auditTime, map } from 'rxjs/operators';

let buffer = [];
const clicks = fromEvent(document, 'click');
const observable = clicks.pipe(
  tap((event) => buffer.push(event)),
  auditTime(1000), // buffer every 1s after 1st click is detected
  map((_lastEvent) => { // ignore last event
    const events = buffer; // save off buffer
    buffer = []; // clear buffer
    return events.reduce((a,e,i) => ({...a, [i]: e}),{});
  }),
);
observable.subscribe((events) => console.log(events));

используя takeUntil и повторите:

ПРИМЕЧАНИЕ: выполнение / повтор будет сбрасывать наблюдаемое (т. Е. Счетчик интервалов остается на 0, и события могут быть потеряны)

import { fromEvent, timer, interval } from 'rxjs';
import { takeUntil, reduce, repeat, filter } from 'rxjs/operators';

const interval$ = interval(1000);
const timer$ = timer(2000);

const observable = interval$.pipe(
  takeUntil(timer$), // unsubscribe from stream every 2s so reduce terminates
  reduce((acc, event) => [...acc, event], []), // reduce to array of events
  filter(x => !!x.length), // suppress emission of empty stream
  repeat(), // resubscribe to stream
);
// console will only show array of [0] since takeUntil stops right when interval emits
const subscription = observable.subscribe(x => console.log(x));
setTimeout(() => subscription.unsubscribe(), 10000);