制作一个只执行一次源的惰性缓存可观察对象答案

【问题标题】：Making a lazy, cached observable that only execute the source once制作一个只执行一次源的惰性缓存可观察对象
【发布时间】：2017-05-30 01:50:44
【问题描述】：

我正在尝试使用 rxjs observable 在应用程序的整个生命周期中委托但共享一项昂贵的工作。

基本上是这样的：

var work$ = Observable.create((o) => {
  const expensive = doSomethingExpensive();
  o.next(expensive);
  observer.complete();
})
.publishReplay(1)
.refCount();

现在，这工作正常，完全符合我的要求，除了一件事：如果所有订阅者都取消订阅，那么当下一个订阅者订阅时，我的昂贵工作又会发生。我想保留它。

现在，我可以使用主题，或者我可以删除 refCount() 并手动使用连接（并且永远不会断开连接）。但这会使昂贵的工作在我连接的那一刻发生，而不是订阅者第一次尝试消耗 work$。

基本上，我想要类似于 refCount 的东西，它只查看要连接的第一个订阅，并且永远不会断开连接。 “惰性连接”。

这样的事情有可能吗？

【问题讨论】：

是否可以简单地创建一个 startWith() 然后将映射操作链接到它（使用标识函数）并让消费者订阅映射的输出 Observable 而不是实际的源？
这是记录在案的行为，至少。这有点可怕，但也许您可以添加一个订阅者，这基本上会使 observable 保持 1 或更多的引用计数，因此它不会刷新缓存的工作。我浏览了 API，但找不到另一种在没有订阅者的情况下保留缓存的方法。
所以，我多玩了一点，这并没有达到我的预期（但它正在做我想要的！）。 See this fiddle我不确定我是否理解。每次订阅者达到 0 时，我都希望 refCount 和 publishReplay 操作从源数据中再次发生（并创建一个新的可连接 observable）。但这并没有发生。

标签： javascript rxjs rxjs5

【解决方案1】：

`publishReplay()` 是如何工作的

它在内部创建一个ReplaySubject 并使其与multicast 兼容。 ReplaySubject 的最小重播值为 1 个发射。结果如下：

第一次订阅将触发publishReplay(1) 在内部订阅源流并通过ReplaySubject 管道所有排放，有效缓存最后n(=1) 排放
如果在源仍处于活动状态时开始第二次订阅，multicast() 会将我们连接到同一个 replaySubject，并且我们将接收所有下一个发射，直到源流完成。
如果订阅是在源完成后开始的，replaySubject 已经缓存了最后的 n 个发射，它只会在完成之前接收这些发射。

const source = Rx.Observable.from([1,2])
  .mergeMap(i => Rx.Observable.of('emission:'+i).delay(i * 100))
  .do(null,null,() => console.log('source stream completed'))
  .publishReplay(1)
  .refCount();

// two subscriptions which are both in time before the stream completes
source.subscribe(val => console.log(`sub1:${val}`), null, () => console.log('sub1 completed'));
source.subscribe(val => console.log(`sub2:${val}`), null, () => console.log('sub2 completed'));

// new subscription after the stream has completed already
setTimeout(() => {
  source.subscribe(val => console.log(`sub_late-to-the-party:${val}`), null, () => console.log('sub_late-to-the-party completed'));
}, 500);

&lt;script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.0.3/Rx.js"&gt;&lt;/script&gt;

【讨论】：

publishReplay() 是如何工作的

`publishReplay()` 是如何工作的