当我将延迟运算符添加到第二个订阅时，共享可观察对象变为单播 [英] Sharing an observable becomes unicast when I add delay operator to a 2nd subscription

问题描述

以下代码按预期工作:

    const source = interval(1000).pipe(
        take(5),    
        share()
  );

  source.subscribe(x => console.log('c1', x));
  setTimeout(() => {
    source.subscribe(x => console.log('c2', x));
  }, 2000);

产生以下输出:0c1 1c1 2c2 2c1 3c2 3c1 4c2 4

producing the following output: c1 0 c1 1 c1 2 c2 2 c1 3 c2 3 c1 4 c2 4

但是当我将第二个订阅更改为使用 delay(2000) 而不是 setTimeout()我收到了一个未共享的不同流.

but when I change the second subscription to use delay(2000) instead of the setTimeout() I get a different stream that is not shared.

    const source = interval(1000).pipe(
        take(5),    
        share()
  );

  source.subscribe(x => console.log('c1', x));

  source.pipe(delay(2000)).subscribe(x => console.log('c2', x));

产生这个输出:

c1 0c1 1c1 2c2 0c1 3c2 1c1 4c2 2c2 3c2 4

c1 0 c1 1 c1 2 c2 0 c1 3 c2 1 c1 4 c2 2 c2 3 c2 4

如何让第二个订阅者使用共享流?我显然不完全理解 RX 操作员是如何在幕后工作的.

How do I get the second subscriber to use the shared stream? I obviously don't understand fully how RX operators work under the hood.

推荐答案

使用 source.pipe(delay(2000)) 与使用 setTimeout() 完全不同.delay() 运算符将延迟来自其源的每次发射，这意味着您仍然会立即进行两次订阅.

Using source.pipe(delay(2000)) is completely different to using setTimeout(). delay() operator will delay each emission from its source which means you're still making two subscriptions immediately.

你可能想做的是:

of(null)
  .pipe(
    delay(2000),
    switchMapTo(source),
  )
  .subscribe();

或者这应该做同样的事情:

Or this should do the same thing:

concat(timer(2000), source)
  .subscribe();

这篇关于当我将延迟运算符添加到第二个订阅时，共享可观察对象变为单播的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！