RxJS：延迟可观察项的生成

Question

假设我有一个 Observable，它生成从 0 到 9 的元素。我想要一个新的 Observable，它将仅在 7 之后发出 3。我相信它应该看起来像这样：

// INPUT:  0 - 1 - 2 - 3 - 4 - 5 - 6 - 7 - 8 - 9
// OUTPUT: 0 - 1 - 2 ----- 4 - 5 - 6 - 7 - 3 - 8 - 9 

let source = Rx.Observable.range(0, 10).XXX([
    item => item === 3,
    item => item === 7
]);

RxJS 是否包含这样的 XXX 方法？

Answer 1

我不知道有什么方法存在，但是，我们可以尝试解决一些问题。让我们将我们的底层可观察对象分成两个流；一个满足我们的谓词（即 x => x === 3）和一个值不满足该谓词的值：

const item$ = Rx.Observable.range(0, 10);
const predicate = x => x === 3;

// Split our underlying stream into two Observables; one where each value
// will equal three, and the other where each value is not three.
const [ three$, notThree$ ] = item$.partition(predicate);

我们可以创建一个在数字 7 出现时发出的流：

const seven$ = notThree$.filter(x => x === 7);

现在，让我们创建两个流来表示发出 7 之前和之后的基础数据

const beforeSeven$ = notThree$.takeUntil(seven$);
const afterSeven$ = notThree$.skipUntil(seven$);

最后，我们可以组合这些流以获得所需的效果。我们基本上在 beforeSeven$ 流结束后立即连接 afterSeven$ 和 three$ 流：

const result$ = beforeSeven$.concat(three$.merge(afterSeven$);

这是一个 console.log 结果的 jsfiddle：https://jsfiddle.net/cbelden/ppjmxww1/

RxJS：延迟可观察项的生成

RxJS: defer item generation of an observable

javascript

reactive-programming

rxjs