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类中。在订阅时，我们还可支持多个订阅，增加了复用和可扩展性。