RxJS Angular：监听一个 observable 并使用另外两个的最新值

Question

我正在使用 Redux 和 RxJS Angular (4) 实现在我的应用中重新管理状态和操作。

我正在收听一个事件（产品列表中所选产品 ID 的更改），因此我们称之为 selectedProductId$。

当相关的发射器发出一个新的 selectedId 值时，我想从这个 Observable 值中执行一个switchMap，以及来自其他两个 Observable 的最新发射值（比如说：productDetails$ 和 productComments$）。

因此，“触发更改”应该是 selectedProductID，我们将使用 productDetails 和 productComments 的最新值创建 switchMap。

我试过这个：

// subscribed to the 3 observable passed in combineLatest
 fetchProduct$: Observable<any>;
 this.fetchProduct$ = Observable.combineLatest(
       this.selectedProductId$,
       this.productComments$,
       this.productDetails$
 ).switchMap(
       result => Observable.of(
         {
           'uuid': result[0],
           'comments': result[1],
           'details': result[2]
         }
      )
 );

然后我订阅这个 observable 并在传递的值上放置一个断点：

this.fetchProductSubscription = this.fetchProduct$.subscribe(result => this.someFn(value));

它可以工作，但每次三个 Observable 之一检测到状态变化时都会调用它。我也尝试了 Observable.forkJoin，但是从文档中，它等待三个 Observable 发出新值，这让我在从另一个 Web 服务获取 cmets 时出现奇怪的状态，并且它可以在选择下一个产品 Id 时检索它。

Observable.merge 也不是一个解决方案，因为它只会在一个“通道”上一个接一个地发出值。

我还没有完全习惯反应式编程，也许我没有清楚地看到一个明显的解决方案，但我知道我很想得到一个提示或一些帮助来解决这个问题:( 非常感谢

Answer 1

首先，让我们合并您不想触发任何更改的“额外”可观察对象。

const extras$ = Observable.combineLatest(this.productComments$, this.productDetails$);

现在，我们在“有趣”的 observable 中观察排放，并使用我们额外的最新值。

cosnt allTogether$ = this.selectedProductId$.withLatestFrom(extras$,
    ([uuid, [comments, details]]) => ({uuid, comments, details}) );

现在欢迎订阅allTogether$。

Answer 2

尝试withLatestFrom，它的工作方式类似于combineLatest()，但每次触发源可观察对象时都会触发。示例（对于 rxjs 5.5）：

this.fetchProduct$ = 
  this.selectedProductId$.pipe(
   withLatestFrom(
       this.productComments$,
       this.productDetails$
   ));

或者对于早期版本：

this.fetchProduct$ = 
  this.selectedProductId$.withLatestFrom(
       this.productComments$,
       this.productDetails$
   );