javascript - SwitchMap 和 PublishReplay 结合使用

Question

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

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 调用”。我的期望是，只要有一个订阅者订阅了可观察对象，每个发出的事件都会立即触发 http 调用。但是，每个订阅者都会收到相同的发射，这意味着每个事件只有一个 http 调用。

我知道我可以通过在可观察对象上使用 publishReplay + refcount 来修复它(见下文)，但这只适用于这个简单的示例。

我真正想要实现的是，两个组件使用相同的 http 调用响应，并且每个组件都发出一个 onInit 事件，然后将其切换映射到 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 通知时，它们将自己标记为“已停止”并且永远不会发出任何东西(这是所谓的 Observable 合约的要求)。可以看看这篇文章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" 如果我正确理解了您的描述，这就是您想要的:

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