javascript - RxJS - 将最后发出的值与以前的值连接起来

Question

我有一个点击事件流 - 我们称它为 stream$。在事件处理程序中，我递增一个计数器(从 0 开始)并调用 observer.next 将计数器作为参数传递。当计数器等于 3 时，我正在完成流。

假设我的 Observable 看起来像这样:

--a---b--c->

其中a、b、c是点击事件

现在，我将获取之前的所有元素(计数器不等于 3):

const a$ = stream$.pipe(filter(x => x !== 3))

最后一个元素(计数器等于 3)

const b$ = stream$.pipe(filter(x => x === 3))

好吧，让我们把它连接起来——a$ 和 b$

const c$ = a$.pipe(concat(b$))

现在，订阅 c$

c$.subscribe(console.log)

不幸的是，它不会工作。我在控制台中只得到 1、2，但没有得到 3(最后一个)。谁能向我解释为什么会这样？

演示:https://stackblitz.com/edit/typescript-iucduj?file=index.ts

代码:

let clickCounter = 0

const stream$ = Observable.create((observer) => {
  document.addEventListener('click', () => {
    clickCounter++;
    observer.next(clickCounter)

    if (clickCounter === 3) {
      observer.complete()
    }
  })
})

const a$ = stream$.pipe(
  filter(x => x !== 3)
)

const b$ = stream$.pipe(
  filter(x => x === 3)
)

const c$ = a$.pipe(concat(b$))

c$.subscribe(console.log)

Answer 1

我认为有两点需要注意:

1. 您正在创建两个 'click' 事件处理程序，因为您对 $stream 进行了两次订阅。这意味着每次点击都会使 clickCounter 增加两次，我认为这是您不想要的。

2. concat 运算符仅在 $a 完成后订阅第二个 Observable b$。然而，在 $a 完成时，值 3 已经发送到 b$ 但没有人在听。

所以我认为最好的方法是为 $stream 使用 share() 然后使用 merge 而不是 concat 因为无论如何 b$ 都不会发出任何东西，这要归功于 filter。

查看更新后的演示:https://stackblitz.com/edit/typescript-t3i7mj?file=index.ts

const stream$ = Observable.create(
  (observer) => {
    document.addEventListener('click', () => {
      ...
    })
  })
  .pipe(
    share()
  );

...

const c$ = a$.pipe(merge(b$));