RxJs操作符，其行为类似于withLatestFrom，但会等待第二个流的值。

Posted 2023-01-21

嗨，我正在寻找一个RxJs操作符，它的行为类似于withLatestFrom，例外的是它会等待第二个流发出一个值，而不是跳过它。更确切地说：我只想在第一个流发出一个新的值时进行排放。

因此，我想用以下行为代替。

---A-----B----C-----D-|
------1----------2----|
   withLatestFrom
---------B1---C1----D2|

我想要这样的行为

---A-----B----C-----D-|
------1----------2----|
   ?????????????
------A1-B1---C1----D2|

有没有一个操作符可以实现这个功能？

答案

我不认为有一个操作符能完全做到这一点，但你可以通过结合高阶映射操作符和高阶映射操作符来实现这些结果。Subject:

second$ = second$.pipe(shareReplay( bufferSize: 1, refCount: false ));

first$.pipe(
  concatMap(
    firstVal => second$.pipe(
      map(secondVal =>  `$firstVal$secondVal`),
      take(1),
    )
  )
)

shareReplay 场所a ReplaySubject 在数据生产者的前面。这意味着它将回复最新的 N(bufferSize)值的每个 新用户. refCount 确保如果没有更多的活跃用户，则会在 ReplaySubject 使用中 不灭.

我决定用 concatMap 我看 更安全 对于 ReplaySubject 只有 一个 主动订阅者。

考虑到这个方案。

---A-----B----C-----D-| first$
------1----------2----| second$

当... A 来了 ReplaySubject(自 shareReplay)将收到一个新的用户和 A 等到 second$ 发出。当这种情况发生时，你会得到 A1 而内部的观测值将完成(这意味着它的订阅者将被从 ReplaySubject的订阅者列表）。) 1 将被 ReplaySubject.

那么 B 进来，新创建的内部订阅者将订阅给 second$ 并将收到 1 立即导致 B1. 同理 C.

现在是一个重要的部分： ReplaySubject 当它从源头接收到一个新的值时，应该没有活跃的订阅者，所以这就是为什么我选择用 take(1). 当 2 将来 ReplaySubject 将没有活跃的订阅者，所以什么都不会发生。

那么 D 到达后将收到最新的存储值。2导致 D2.

另一答案

对，就是这样 加入. 从你的例子来看，不清楚你是在用左右连接还是内连接。

哦，对不起。确实很清楚。你需要内联接的语义（如果两个联接对象中的数据都存在，则发出iff，否则等待）。