用另一个 observable 更改 observable 的订阅状态答案

【问题标题】：Change the subscription state of an observable with another observable用另一个 observable 更改 observable 的订阅状态
【发布时间】：2017-07-08 11:26:01
【问题描述】：

如果您运行下面的代码，您将在控制台中看到无论我是 subscribed 还是 subject，都会发送 xhr 请求。当我没有被替换时，我不想提出这些请求。

// npm install rxjs
const Rx = require('rxjs/Rx');

let subject = new Rx.BehaviorSubject(null)
Rx.Observable.timer(0, 1000).subscribe(i => someApiCall(i));
//at this point xhr request will be sent every second

function someApiCall(i){
    // retrieve some data
    console.log("xhr request sent")
    subject.next(i);
}

//so here we are gonna subscribe to the subject, xhr made prior 
//to this subscription are useless
let subscription;
setTimeout(() => subscription = subject.subscribe(i => console.log(i)),2500);
setTimeout(() => subscription.unsubscribe(),6000);
// now we are unsubscribing but the xhr req keep going

我使用behaviorSubject而不是直接订阅observable的原因是因为我想在我重新订阅时立即获得最后一个xhr请求的最后一个值。

【问题讨论】：

我有点困惑，Rx.Observable.timer(0, 1000).subscribe(i => someApiCall(i)); 永远不会取消订阅，而是会立即被调用。所以，对我来说，“xhr request sent”消息将在页面加载时每秒记录一次。然后在 2 xhr 消息之后，您应该开始看到正在记录的数字，并且您应该只看到记录的 4 个数字。这不是正在发生的事情吗？这就是问题所在吗？
@joe_coolish 是的，就是这样，我希望只有在我能看到数字时才提出请求，这有意义吗？

标签： javascript angular rxjs rxjs5

【解决方案1】：

您应该使用.ShareReplay(1) 而不是BehaviourSubject。这样你就可以让它保持惰性并缓存你的 xhr 调用的最后一个值。

const source = Rx.Observable.interval(1000)
  .mergeMap(I => doXhr())
  .shareReplay(1);

source.subscribe(console.log)

【讨论】：

【解决方案2】：

https://acutmore.jsbin.com/bepiho/2/edit?js,console

const { Observable } = Rx;

function someApiCall(i){
  return Observable.create(observer => {
    console.log("xhr request sent")
    observer.next(i);
    observer.complete();
  });
}

const data = Rx.Observable.timer(0, 1000)
  // map each value to the values of someApiCall
  . mergeMap(i => someApiCall(i))
  // share the values through a replaySubject
  .publishReplay(1)
  // Only connect to the source when there is at least one subscriber
  .refCount();

data
  .take(5)
  .subscribe(v => console.log(v));

data
  .take(1)
  .subscribe(v => console.log(v));

值得注意的是，如果每个人在订阅时都使用.take(1)，这将无法按预期工作，因为每个人都会在ReplaySubject 中获得值，然后在发出新的 xhrRequest 之前立即取消订阅。

即有些东西需要让它存活足够长的时间，以便间隔继续触发。

【讨论】：