为什么 Subject 没有从 Observable 获得 complete() 信号？

Question

我尝试在多播中使用 Subject 来完成，但它没有显示在 complete() 方法中设置的完成消息。是什么原因？以及错误和完整的方法是如何使用的？

let source = new Observable(subscriber => {
   subscriber.next(1);
   subscriber.next(2);

   return () => { 
      subscriber.complete(); 
      console.log('Unsubscribed');
   }
});

let subject = new Subject<number>();
const observer = {
  next: (x) => console.log('Observer got a next value: ', x),
  error: (err) => console.error('Observer got an error: ', err),
  complete: () => console.log('Observer got a complete notification'),
};
subject.subsribe(observer);
const connectableObservable = connectable(source, {
  connector: () => subject,
});
const subscription = connectableObservable.connect();
subscription.unsubscribe();

Answer 1

这与主题或多播或类似的事情无关。这是关于 RxJS 将尝试强制执行的 Observable 合约。

您可以在this link 处查看 Rxjs 合同的可观察对象。

当观察者向 Observable 发出取消订阅通知时，Observable 将尝试停止向观察者发出通知。

这包括complete 通知。

---

更多背景知识

考虑以下几点：

const source = new Observable(observer => {
  observer.next(1);
  observer.next(2);
  observer.complete();
  observer.next(3);

  return { unsubscribe: () => {
    console.log('Unsubscribed'); 
  }};
});

const subscription = source.subscribe({
  next: console.log,
  complete: () => console.log("Source Complete")
});

subscription.unsubscribe();
subscription.unsubscribe();
subscription.unsubscribe();

输出是这样的：

1
2
Source Complete
Unsubscribed

1. 您会注意到在输出中看不到 3，即使上面的 observable 在完成后会发出 3。这是因为 RxJS 可观察对象保证它不会在 complete 或 error 之后发出任何内容。
   所以上面的代码已经被代理了，保证合约得到满足。

2. 您会注意到unsubscribe 代码只被调用一次。实际上，在这种情况下，您根本不需要它，因为已完成的 observable 会自动取消订阅。

考虑以下几点：

const source = new Observable(observer => {
  observer.next(1);
  observer.complete();

  return { unsubscribe: () => {
    console.log('Unsubscribed'); 
  }};
});

source.subscribe({
  next: console.log,
  complete: () => console.log("Source Complete")
});

输出：

1
Source Complete
Unsubscribed

请注意取消订阅代码是如何被调用的，即使我们自己从未调用过它？这也是 RxJS 的 Observable 合约的一部分。在 observable 终止（错误或完成）后，立即调用清理代码（取消订阅），然后不再调用。