如何仅缓冲来自可观察源的选定项目集并立即发出其他项目？

Question

例如，我有一个每秒发出一个数字的 observable。我想将它们缓冲到数组中 4 秒。但是当源发出一个可以被 5 整除的数字时，我想立即发出它而不是缓冲它。// emit value in sequence every 1 second// 0,1,2,3,4,5...const source = interval(1000);// buffer values for 4 secondsconst 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 onStackblitz：https ://stackblitz.com/edit/typescript-cdvc2d ? file = index.ts

Answer 1

拥有两个具有两种不同职责的可观察对象，然后将它们合并在一起怎么样？

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));

我不确定您是否还想过滤第一个 observable 中 5 的倍数，但这很容易做到。

Answer 2

您可以使用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