当最后一个观察者取消订阅时如何防止 AsyncSubject 完成

The AsyncSubject当最后一个主题观察者取消订阅该主题时，该主题变得可观察。这是报价单 https://medium.com/@benlesh/hot-vs-cold-observables-f8094ed53339#2991:

当它完成时，它就完成了。对象在使用后无法重复使用 取消订阅、已完成或有错误。

这是演示：

const ofObservable = Rx.Observable.of(1, 2, 3);
const subject = new Rx.AsyncSubject();

ofObservable.subscribe(subject);

subject.subscribe((v) => {
    console.log(v);
});

subject.unsubscribe((v) => {
    console.log(v);
});

// here I'll get the error "object unsubscribed"
subject.subscribe((v) => {
    console.log(v);
});

如何阻止主题完成？

有一个share操作员：

在 RxJS 5 中，运算符share()制作一个热的、refCounted 的 observable 失败时可以重试，成功时可以重复。因为 一旦出现错误、完成或其他情况，主题就无法重复使用 取消订阅，则share() 操作员将把死去的受试者回收到 启用重新订阅到结果可观察值。

这就是我正在寻找的。但share创建一个主题，我需要AsyncSubject.

问题归结为这一行：

subject.unsubscribe((v) => {
    console.log(v);
});

Subject 实施ISubscription https://github.com/ReactiveX/rxjs/blob/5.4.2/src/Subject.ts#L22;这意味着它有一个unsubscribe方法和一个closed财产。它是实施unsubscribe https://github.com/ReactiveX/rxjs/blob/5.4.2/src/Subject.ts#L96-L100如下：

unsubscribe() {
  this.isStopped = true;
  this.closed = true;
  this.observers = null;
}

这有点残酷。本质上，它会切断与该主题的任何订阅者的所有通信，而无需取消订阅。同样，它不会从它可能碰巧订阅的任何可观察对象中取消订阅主题本身。 （它还将主题标记为关闭/停止，这就是您错误的原因。）

鉴于它实际上并不执行任何取消订阅，因此尚不清楚它应该如何使用。的描述这次测试 https://github.com/ReactiveX/rxjs/blob/5.4.2/spec/Subject-spec.ts#L235-L276:

it('should disallow new subscriber once subject has been disposed', () => {

表明这可能是 RxJS 4 的某种后遗症 - 其中取消订阅被称为处置。

无论其存在的原因是什么，我建议永远不要调用它。举例来说，看看这个片段：

const source = Rx.Observable
  .interval(200)
  .take(5)
  .do(value => console.log(`source: ${value}`));

const subject = new Rx.Subject();
source.subscribe(subject);

const subscription = subject
  .switchMap(() => Rx.Observable
    .interval(200)
    .take(5)
    .delay(500))
  .subscribe(value => console.log(`subscription: ${value}`));

.as-console-wrapper { max-height: 100% !important; top: 0; }

<script src="https://unpkg.com/rxjs@5/bundles/Rx.min.js"></script>

它将主题订阅到源可观察量，然后订阅由该主题组成的可观察量。

If unsubscribe当这个主题被调用时，一些问题就变得明显了：

• 主题对源的订阅并未取消订阅，并且当源尝试调用主题的next方法;和
• 对由主题组成的可观察量的订阅不会取消订阅，因此interval内可观察到switchMap后继续发光unsubscribe已拨打电话。

试试看：

const source = Rx.Observable
  .interval(200)
  .take(5)
  .do(value => console.log(`source: ${value}`));

const subject = new Rx.Subject();
source.subscribe(subject);

const subscription = subject
  .switchMap(() => Rx.Observable
    .interval(200)
    .take(5)
    .delay(500))
  .subscribe(value => console.log(`subscription: ${value}`));

setTimeout(() => {
  console.log("subject.unsubscribe()");
  subject.unsubscribe();
}, 700);

.as-console-wrapper { max-height: 100% !important; top: 0; }

<script src="https://unpkg.com/rxjs@5/bundles/Rx.min.js"></script>

这些似乎都不是理想的行为，所以打电话unsubscribe on a Subject是应该避免的事情。

相反，您的代码片段中的代码应该使用以下方式取消订阅Subscription由返回subscribe call:

const subscription = subject.subscribe((v) => {
  console.log(v);
});
subscription.unsubscribe();

写完这个答案后，我发现了以下评论 https://medium.com/@benlesh/on-the-subject-of-subjects-in-rxjs-2b08b7198b93 from Ben Lesh http://stackoverflow.com/users/135786，这符合我的理论，即它与主题的处理有关：

如果你想让对象在你说话时大声而愤怒地犯错next当它有用后，你可以调用unsubscribe直接作用于主题实例本身。