Как буферизовать только выбранный набор элементов из наблюдаемого источника и немедленно создать другие элементы?

Question

Например, у меня есть наблюдаемое, которое испускает число каждую секунду. Я хочу буферизовать их в массивы в течение 4 секунд. Но когда источник испускает число, которое делится на 5, я хочу немедленно его испустить, а не буферизовать.

// emit value in sequence every 1 second
// 0,1,2,3,4,5...
const source = interval(1000);

// buffer values for 4 seconds
const buffered = source.pipe(
  buffer(interval(4000)),
  filter(x => x.length > 0)  
);

const subscribe = buffered.subscribe(val => console.log(val));

Мой ожидаемый результат:

[0,1,2,3] // buffered and emitted after 4 seconds
[5] // emitted immediately
[4,6,7,8] // buffered and emitted after 4 seconds
[10] // emitted immediately
[9,11,12,13]
// ... and so on

Stackblitz: https://stackblitz.com/edit/typescript-cdvc2d?file=index.ts

Answer 1

Вы можете использовать multicast, чтобы разделить исходный поток на два, а затем объединить их снова с различными filter операторами.

import { of, interval, Subject, merge } from 'rxjs';
import { bufferTime, multicast, filter } from 'rxjs/operators';

const buffered = source.pipe(
  multicast(new Subject(), s => merge(
    s.pipe(
      filter((value: number) => value % 5 === 0),
    ),
    s.pipe(
      filter((value: number) => value % 5 !== 0),
      bufferTime(4000),
    ),
  ))
);

Обновленная демоверсия: https://stackblitz.com/edit/typescript-sbhn8s?file=index.ts

Answer 2

Как насчет наличия двух наблюдаемых с двумя разными обязанностями и последующего их слияния?

const source = interval(1000);

// buffer values for 4 seconds
const buffered = source.pipe(
  buffer(interval(4000)),
  filter(x => x.length > 0)  
);

const multipleOfFive = source.pipe(
  filter(x => x % 5 === 0)
);

const subscribe = merge(buffered, multipleOfFive).subscribe(val => console.log(val));

Я не уверен, хотели бы вы также отфильтровать кратные 5 в первой наблюдаемой, но этобыло бы легко сделать.