javascript - angular rxjs 在没有订阅的情况下发送所有可观察到的东西之后做一些事情

Question

我将使用异步管道，不会运行订阅方法。

比如将 1 连接到 10 并将每 3 个数字切片到一个新数组。

import { timer, concat, interval, range } from 'rxjs';
import { delay, map, tap, take, startWith } from 'rxjs/operators';

const sliceEveryN = (a,b) => console.log('Method runs')
const arr = []
const source = range(1, 5); // [1..5]
const source2 = range(6, 5); // [6..10]
const final = concat(source, source2).pipe(
	map(v => arr.push(v))
	).pipe(
	// I want to this at last, not every observable sent
	map(v => sliceEveryN(3, v)) // [[1,2,3], [4,5,6], [7,8,9], [10]]
	)

这将运行 sliceEveryN 方法十次，期望一次。

Answer 1

使用 bufferCount(N) 将缓冲 N 次发射并发射和聚合数组:

Collect emitted values until provided number is fulfilled, emit as array

const final = concat(source, source2).pipe(
    tap(v => arr.push(v)),
    bufferCount(3),
    tap(v => console.log(v))
);

这将发出每 3 个值的数组并将它们打印到控制台。