如何使`shared` Observable 允许触发第二个 observable 一次而不是每次发射？答案

【问题标题】：How to make `shared` Observable to allow firing a second observable once and not on each emission?如何使`shared` Observable 允许触发第二个 observable 一次而不是每次发射？
【发布时间】：2021-07-02 15:20:18
【问题描述】：

我有一个可观察的源，用于 WS 连接。该连接已打开onMount。最重要的是，我触发了另一个 observable onMount 以允许 LoadingIndicator 屏幕，直到数据到达。我希望该指标一直存在，直到数据从源 observable 发出，我已经这样做了。

  const mount$ = action$.pipe(Type.ON_MOUNT);

  const source$ = mount$.pipe(
    ... // Stuff happen here..
    share()
  );

  const setSuccess$ = source$.pipe(
    map(() => ({ type: Type.SET_LOADING, payload: false }))
  );

  const setLoading$ = mount$.pipe(map(() => ({ type: Type.SET_LOADING, payload: true })));

  return merge(source$, setLoading$, setSuccess$);

然后，我有setSucccess observable，它正在监听source observable 发射，并将isLoading 设置为false。 这就是问题所在。每次排放都会发生这种情况。我希望这仅发生在第一个上。我怎样才能做到这一点？

我尝试了一些不同的方法。我在setSuccess$ 上尝试了shareReplay({ refCount: true, bufferSize: 1 })、take(1)，但没有任何效果。 我怎样才能只发出第二个共享 observable (setSuccess$) 一次？

【问题讨论】：

take(1) 应该做你想做的事。什么没试过？
这不适用于较慢的连接。因为第二个订阅者不会像它应该的那样稍后发出。虽然它确实会在我无法真正可靠地使用它时发出..
使用 martins 建议的 take(1)，但尝试将您的 merge() 替换为 concat()，以便强制执行顺序并且不允许并发。
我想我明白为什么take(1) 不起作用。当我source$.pipe(...) 时，我总是有 1 个订阅者，因此第二次订阅将不起作用。关于如何立即终止我的第二次订阅的任何想法。

标签： javascript typescript rxjs multicast

【解决方案1】：

您不需要从mount$ 开始setLoading$。它最初处于加载状态，因此您可以

return merge(source$, of({ type: Type.SET_LOADING, payload: true }), setSuccess$.pipe(take(1)));

【讨论】：