SwitchMap 和 PublishReplay 组合

Question

我正在处理一个 Angular 项目，但我不完全理解为什么这段代码 sn-p 没有按预期工作：

const httpObs = Rx.Observable.of("")
  .do(() => console.log("trigger http call")) //only triggered once
  .publishReplay(1).refCount();

let observable = Rx.Observable.timer(0,1000)
  .do((e) => console.log("new event: " + e))
  .switchMap(() => this.httpObs);

const c1 = observable.subscribe(() => console.log("subscriber 1"));
const c2 = observable.subscribe(() => console.log("subscriber 2"));

JS Bin

我不明白为什么 Rx.Observable.timer 发出的每个事件都不会触发“触发 http 调用”。我的期望是，只要一个订阅者订阅了 observable，每个发出的事件都会触发 http 调用。但是，每个订阅者都会收到相同的排放，这意味着每个事件只有一个 http 调用。

我知道我可以通过在 observable 上使用 publishReplay + refcount 来修复它（见下文），但这仅适用于这个简单的示例。

我真正尝试实现的是两个组件使用相同的 http 调用响应，并且每个组件都会发出一个 onInit 事件，然后将其 switchMapped 到 http-response。我不希望每个新订户都触发呼叫。但是，我希望在发出另一个事件时触发调用。 onInit observable 与 click-observable 合并，只要用户点击“更新”按钮，它就会在 switchMap 操作符用于转换之前发出新事件。现在的问题是，更新按钮不再触发 http 调用。

这个例子展示了它应该如何表现：

const httpObs = Rx.Observable.of("")
  .do(() => console.log("trigger http call"));

let observable = Rx.Observable.timer(0,1000)
  .do((e) => console.log("new event: " + e))
  .switchMap(() => this.httpObs)
  .publishReplay(1);

observable.subscribe(() => console.log("subscriber 1"));
observable.subscribe(() => console.log("subscriber 2"))

observable.connect();

JS Bin

Answer 1

您遇到的问题是所有 Subject 实例都有其内部状态，当它们收到 complete 或 error 通知时，它们会将自己标记为“已停止”并且永远不会发出任何东西（这是所谓的可观察合同）。你可以看看我前段时间写的这篇文章https://medium.com/@martin.sikora/rxjs-subjects-and-their-internal-state-7cfdee905156。它涵盖了非常相似的主题。

在您的代码中，您多次使用相同的 httpObs，但源 Observable 是 of()，它发出一个值，然后发送 complete 通知。

看到它真的叫：http://jsbin.com/fuvuxax/2/edit?js,console

const httpObs = Rx.Observable.of("")
  .do(() => console.log("trigger call"), undefined, () => console.log('complete'))
  .publishReplay(1).refCount();

这意味着publishReplay 中的 Subject 收到了 complete 通知，并且不会再次订阅其源 Observable。但是，由于publishReplay 在内部使用ReplaySubject，它仍然会刷新其缓冲区，然后发送complete 通知。但这对您来说不是很有用，因为我认为您想再次执行 HTTP 调用，而不仅仅是重播过时的调用。

您可以做的是将httpObs 转换为每次switchMap 收到值时创建链的方法：

const httpObs = () => Rx.Observable.of("")
  .do(() => console.log("trigger call"), undefined, () => console.log('complete'));

let observable = Rx.Observable.timer(0,1000)
  .do((e) => console.log("new event: " + e))
  .take(3)
  .switchMap(() => this.httpObs())
  .publishReplay(1).refCount();

observable.subscribe(() => console.log("subscriber 1"));
observable.subscribe(() => console.log("subscriber 2"));

您可以看到，"trigger call" 只打印了 3 次，如果我正确理解您的描述，我认为这就是您想要的：

"trigger call"
"subscriber 1"
"subscriber 2"
"complete"
"trigger call"
"subscriber 1"
"subscriber 2"
"complete"
"trigger call"
"subscriber 1"
"subscriber 2"
"complete"