BFE.devAI解答 - JavaScript 代码题
57. 实现Observable

实现一个基本的Observable，满足以下要求：

1. 实现Observable构造函数，它应该接受一个参数subscribe，这个参数是一个函数，这个函数接受一个observer作为参数，这个observer是一个对象，它有三个属性：next、error、complete，这些属性都是函数。
2. next函数会传递一个值给Observer。
3. error函数会传递一个错误给Observer。
4. complete函数表示Observable结束，不再传递任何值给Observer。
5. 当一个Observer通过调用subscribe方法订阅Observable时，subscribe方法应该返回一个Subscription对象。
6. Subscription对象有一个unsubscribe函数，当它被调用时，它会取消Observer对Observable的订阅，不再接收任何值。
7. error和complete只能触发一次。其后的next/error/complete需要被忽略。
8. 在订阅的时候next/error/complete需要都不是必须。如果传入的是一个函数，这个函数需要被默认为next。
9. 需要支持多个订阅。

具体实现如下：

class Observable {  constructor(subscribe) {    this._subscribe = subscribe;  }    subscribe(observerOrNext, error, complete) {    let observer;    if (typeof observerOrNext === 'function') {      observer = {        next: observerOrNext,        error: error || (() => {}),        complete: complete || (() => {}),      };    } else {      observer = observerOrNext;    }    const subscription = new Subscription();    const safeObserver = new SafeObserver(observer);    subscription.add(safeObserver);    const teardown = this._subscribe(safeObserver);    if (typeof teardown === 'function') {      subscription.add(new Subscription(teardown));    }    return subscription;  }}class SafeObserver {  constructor(observer) {    this._observer = observer;    this._closed = false;  }    next(value) {    if (!this._closed && this._observer.next) {      try {        this._observer.next(value);      } catch (e) {        this.error(e);      }    }  }    error(error) {    if (!this._closed) {      this._closed = true;      if (this._observer.error) {        try {          this._observer.error(error);        } catch (e) {          console.error(e);        }      }      this.unsubscribe();    }  }    complete() {    if (!this._closed) {      this._closed = true;      if (this._observer.complete) {        try {          this._observer.complete();        } catch (e) {          console.error(e);        }      }      this.unsubscribe();    }  }    get closed() {    return this._closed;  }    unsubscribe() {    this._closed = true;    if (this._observer.unsubscribe) {      this._observer.unsubscribe();    }  }}class Subscription {  constructor(teardown) {    this._closed = false;    if (teardown) {      this._teardown = teardown;    }  }    add(teardown) {    if (teardown && teardown !== this) {      if (this._closed) {        teardown.unsubscribe();      } else {        if (!(teardown instanceof Subscription)) {          teardown = new Subscription(teardown);        }        if (!this._childSubscriptions) {          this._childSubscriptions = [];        }        this._childSubscriptions.push(teardown);      }    }  }    unsubscribe() {    this._closed = true;    if (this._teardown) {      try {        this._teardown();      } catch (e) {        console.error(e);      }    }    if (this._childSubscriptions) {      while (this._childSubscriptions.length) {        const childSubscription = this._childSubscriptions.shift();        childSubscription.unsubscribe();      }    }  }}

上述代码中，我们使用SafeObserver类包装原始的Observer，以处理错误和取消订阅等问题。Subscription类表示一个订阅，它可以包含一个或多个子订阅，以便在取消订阅时一起取消。在subscribe方法中，我们先判断传入的observerOrNext参数类型，如果是函数，则将其视为next回调函数，并创建一个新的Observer对象，否则直接使用传入的observerOrNext作为Observer对象。接着创建一个新的Subscription对象，并用它来存储订阅的相关信息。使用SafeObserver类包装Observer，并向Observable的subscribe方法传入它，从而将订阅的处理逻辑安全地封装在SafeObserver类中。在订阅时，我们还可支持多个订阅，增加了复用和可扩展性。