为什么使用 share 会阻止使用 withLatestFrom 的 observable？

Question

问题

我有以下运算符：

  const prepare = (value$: Observable<string>) =>
    value$.pipe(
      tap((x) => console.log("prepare: ", x)),
      share()
    );

  const performTaskA = (removed$: Observable<string>) =>
    removed$.pipe(tap((x) => console.log("taskA: ", x)));

  const performTaskB = (removed$: Observable<string>) =>
    removed$.pipe(
      tap((x) => console.log("taskB 1: ", x)),
      withLatestFrom(otherValue$),
      tap((x) => console.log("taskB 2: ", x))
    );

我这样称呼他们：

  const prepared$ = value$.pipe(prepare);
  const taskADone$ = prepared$.pipe(performTaskA);
  const taskBDone$ = prepared$.pipe(performTaskB);

  merge(taskADone$, taskBDone$).subscribe();

产生以下输出：

prepare:  TEST 
taskA:  TEST 
taskB 1:  TEST

请注意，taskB 2 尚未被记录 - 似乎 taskBDone 可观察对象已停止在 performTaskB 中的 withLatestFrom(otherValue$)。

如果prepare 中的share 被删除，observable 不会停止，但它（不出所料）会导致prepare 执行两次，这是我不希望的。

问题

1. 我怎样才能同时执行performTaskA 和performTaskB 但prepare 只执行一次？
2. 鉴于下面的调试说明，为什么share 会导致发射序列发生变化？

演示
分享（如上）：https://codesandbox.io/s/so-share-with-latest-from-with-share-rtyex?file=/src/index.ts:663-853
无分享：https://codesandbox.io/s/so-share-with-latest-from-no-share-p702e

转到右侧的“测试”选项卡，确保控制台可见，然后单击“播放”按钮。

部分解释
调试withLatestFrom 很明显，当源(removed$) 发射时，ready 是false here，这会阻止发射。

这是因为当 share 存在时，输入 (otherValue$) 订阅在源之后发出，因此尚未设置 ready。 （或者是share导致源提前发射？）

但是当share 被删除时，输入订阅发出在源之前发出，这意味着ready 通过here 和here 设置为真，因此withLatestFrom 发出为预计。

Answer 1

我尝试运行你的代码，但我无法让它像你的那样挂起。我怀疑这与您的测试框架如何管理可观察对象有关。我无法复制它。

我确实注意到您所写的内容存在一些内在的交错/排序问题。共享同步 observables 会带来一些问题，例如第一个观察者同步观察所有源值以及它的完成在第二个观察者可以订阅之前。即使它们应该“同时”订阅，也会发生这种情况。

使用事件循环排序

我在这里写的对我有用：

function init(
  value$: Observable<string>,
  otherValue$: Observable<string>
) {
  const prepare = pipe(
    tap(x => console.log("prepare: ", x)),
    delay(0),
    share()
  );

  const performTaskA = pipe(
    tap(x => console.log("taskA: ", x))
  );

  const performTaskB = pipe(
    tap(x => console.log("taskB 1: ", x)),
    withLatestFrom(otherValue$),
    tap(x => console.log("taskB 2: ", x))
  );

  const prepared$ = value$.pipe(prepare);

  merge(
    prepared$.pipe(performTaskA), 
    prepared$.pipe(performTaskB)
  ).subscribe();
}

init(of("a"), of("b"));

对我来说，这会将其打印到控制台：

prepare: a
taskA: a
taskB 1: a
taskB 2: ["a","b"]

您会注意到在prepare 中调用了 delay(0)。否则，由于共享的 observable 已同步完成，因此会调用两次 prepare。 Delay(0) 只是尽快将下一个调用放入事件队列中。

这不是最好的解决方案。这是一个黑客。最佳解决方案取决于如何使用它。大多数时候，shareReplay(1) 完成了这项工作。如果你在这里使用它，那就行了。

使用发布/连接进行排序

否则，您可以发布和连接以确保订单符合您的预期。

发布/连接可让您在源正式启动/连接/订阅之前设置对源的所有订阅。这可以确保在幕后，所有回调都在任何事情发生之前就位。这是确保完全同步的 observable 可以共享其值的唯一方法。

function init(
  value$: Observable<string>,
  otherValue$: Observable<string>
) {
  const prepare = pipe(
    tap(x => console.log("prepare: ", x)),
    share()
  );

  const performTaskA = pipe(
    tap(x => console.log("taskA: ", x))
  );

  const performTaskB = pipe(
    tap(x => console.log("taskB 1: ", x)),
    withLatestFrom(otherValue$),
    tap(x => console.log("taskB 2: ", x))
  );

  const prepared$ = publish()(value$.pipe(prepare));

  merge(
    prepared$.pipe(performTaskA), 
    prepared$.pipe(performTaskB)
  ).subscribe();

  prepared$.connect();
}