使用 RxJ 有条件地合并两个流答案

【问题标题】：Conditionally merge two streams with RxJs使用 RxJ 有条件地合并两个流
【发布时间】：2018-11-20 18:19:13
【问题描述】：

我有两个服务器调用要在我的流中获得一些响应。但如果第一次调用返回一些数据，我不想执行第二次调用。在第一个请求成功时，我找不到合适的运算符或运算符组合来取消（我不想抛出错误）第二个请求。

我也尝试了 defaultIfEmpty，但看起来像个 hack，我不喜欢它。

我创建了一个 StackBlitz 示例，您可以查看它here，但我不确定这是否是一个好习惯。所以基本上我需要一个条件 concatMap。

如果您知道如何在不创建新运算符的情况下使用可用的运算符，我真的很想听听您的方法。

提前致谢。

【问题讨论】：

【解决方案1】：

如果您的第一个异步请求是 Observable<T> 类型，并且仅在成功时返回值 T，否则只是完成而不发出值，您可以改用简单的 .concat()：

firstAsyncCall()
  .concat(secondAsyncCall())
  .take(1)

.take(1) 将确保secondAsyncCall() 在firstAsyncCall 返回（至少）1 个值时不会被调用。

【讨论】：

绚丽优雅
哇，是的，这真的很简单。这实际上是我一直在寻找的。谢谢！不过，这对我来说并不直观。 take(1) 在 concat() 之后并且可以取消它。我应该更多地了解管道的执行。再次感谢。

【解决方案2】：

当值不适合您时，使用filter 停止流：

const interval$ = interval(100);

interval$.pipe(
  filter(val => val % 10 === 0),
  take(10),
).subscribe(val => console.log(val));

【讨论】：

感谢您的回答。但我的第一个来源只是一个异步 API 调用。如果我过滤不存在的响应，如果第一个调用返回 null，它不会触发第二个请求，对吗？
如果你设置正确的条件是，它不会发送第二个请求。如果不满足过滤器中的条件，则不会运行任何跟随它（或订阅）的运算符。
@UgurKorfali（这就是为什么你顺便看到0到90，因为take运算符在它之后，它只会在满足条件时触发）
const subscribe = source1$.pipe( filter(x => x !== null), concatMap(() => source2$ ) ).subscribe(console.log); 这样我就无法得到firstSource$的响应，如果我得到正确的话
在这种情况下，如果source1$ 不返回 null，您将看到控制台日志。否则，您将看不到控制台日志。所以你确实得到了第一个请求（发出）的结果，它只是在concatMap 之前停止。