条件 RxJS 流

【问题标题】：Conditional RxJS Stream条件 RxJS 流
【发布时间】：2017-11-06 19:28:19
【问题描述】：

resource1$ = hash1$.map( (renew: boolean) => renew ? http1$ : Observable.empty() );
resource2$ = hash2$.map( (renew: boolean) => renew ? http2$ : Observable.empty() );

sync$ = Observable.forkJoin(resource1$, resource2$);

sync$.subscribe( () => console.log('Sync done!), (err) => console.log('Sync failed!') );

你好，当我的应用程序启动时，我有多个资源要从 API 同步。我想并行同步它们，并检查是否需要在之前使用 HEAD 请求同步它们，并将 X-HASH 标头与旧存储的标头进行比较。

所以 hash1$ 做一个 HEAD 请求，比较哈希值并返回 true 或 false。

我被卡住了，因为如果 resource1$ 返回 Observable.empty，sync$ 取消所有流...我不明白为什么。

【问题讨论】：

标签： angular typescript rxjs

【解决方案1】：

forkJoin 要求所有源 Observable 至少发出一项并完成。如果您使用Observable.empty()，您只会发送complete 通知，这就是forkJoin 永远不会发出的原因。

你可以这样做：

resource1$ = hash1$.map((renew: boolean) => renew ? http1$ : Observable.of(false));
resource2$ = hash2$.map((renew: boolean) => renew ? http2$ : Observable.of(false));

sync$ = Observable.forkJoin(resource1$, resource2$)
  .filter(results => results[0] && results[1]); // Or whatever condition you want

【讨论】：

谢谢马丁。现在没有失败。但我需要使用 .flatMap 运算符代替 .map hash1$.flatMap((renew: boolean) => renew ? http1$ : Observable.of(false)); 可能是因为 http1$ 是 Observable.fromPromise 吗？
这取决于您以后要如何使用sync$。