Observable 主题事件监听器

Question

我正在研究 Observables 及其与 EventEmitter 的差异，然后偶然发现了 Subjects （我可以看到 Angulars EventEmitter 基于关闭）。

似乎 Observable 是单播 vs 多播的主题（然后 EE 只是一个主题，它将 .next 包装在发射中以提供正确的接口）。

Observables 似乎很容易实现

class Observable {
    constructor(subscribe) {
        this._subscribe = subscribe;
    }

    subscribe(next, complete, error) {
        const observer = new Observer(next, complete, error);

        // return way to unsubscribe
        return this._subscribe(observer);
    }

}

哪里Observer只是一个添加一些尝试捕获和监控的包装器 isComplete 以便它可以清理并停止观察。

对于我想出的一个主题：

class Subject {
    subscribers = new Set();

    constructor() {
        this.observable = new Observable(observer => {
            this.observer = observer;
        });

        this.observable.subscribe((...args) => {
            this.subscribers.forEach(sub => sub(...args))
        });
    }

    subscribe(subscriber) {
        this.subscribers.add(subscriber);
    }

    emit(...args) {
        this.observer.next(...args);
    }
}

哪种合并到一个 EventEmitter 中，它用 .next 包装 .next 和发射 - 但捕获observeObservable的参数似乎是错误的 - 就像我刚刚解决了一个解决方案。从 Observable（单播）生成主题（多播）的更好方法是什么？

我尝试查看 RXJS，但我看不到它的subscribers数组是如何填充的：/

Answer 1

我认为您也可以通过使用调试器更好地理解。打开一个 StackBlitz RxJS 项目，创建最简单的例子（取决于你试图理解的内容），然后放置一些断点。AFAIK，使用 StackBlitz，您可以调试 TypeScript 文件，这看起来很棒。

首先，Subject该类扩展了Observable：

export class Subject<T> extends Observable<T> implements SubscriptionLike { /* ... */ }

现在让我们检查一下这个Observable类。

它有一个众所周知的pipe方法：

pipe(...operations: OperatorFunction<any, any>[]): Observable<any> {
  return operations.length ? pipeFromArray(operations)(this) : this;
}

其中pipeFromArray定义如下：

export function pipeFromArray<T, R>(fns: Array<UnaryFunction<T, R>>): UnaryFunction<T, R> {
  if (fns.length === 0) {
    return identity as UnaryFunction<any, any>;
  }

  if (fns.length === 1) {
    return fns[0];
  }

  return function piped(input: T): R {
    return fns.reduce((prev: any, fn: UnaryFunction<T, R>) => fn(prev), input as any);
  };
}

在澄清上述代码段中发生的事情之前，了解运算符是很重要的。运算符是一个函数，它返回另一个函数，该函数的单个参数为 an Observable<T>，返回类型为 an Observable<R>。有时，T并且R可以是相同的（例如，当使用filter(), debounceTime()... 时）。

例如，map被定义如下：

export function map<T, R>(project: (value: T, index: number) => R, thisArg?: any): OperatorFunction<T, R> {
  return operate((source, subscriber) => {
    // The index of the value from the source. Used with projection.
    let index = 0;
    // Subscribe to the source, all errors and completions are sent along
    // to the consumer.
    source.subscribe(
      new OperatorSubscriber(subscriber, (value: T) => {
        // Call the projection function with the appropriate this context,
        // and send the resulting value to the consumer.
        subscriber.next(project.call(thisArg, value, index++));
      })
    );
  });
}

export function operate<T, R>(
  init: (liftedSource: Observable<T>, subscriber: Subscriber<R>) => (() => void) | void
): OperatorFunction<T, R> {
  return (source: Observable<T>) => {
    if (hasLift(source)) {
      return source.lift(function (this: Subscriber<R>, liftedSource: Observable<T>) {
        try {
          return init(liftedSource, this);
        } catch (err) {
          this.error(err);
        }
      });
    }
    throw new TypeError('Unable to lift unknown Observable type');
  };
}

所以，operate会返回一个函数。注意它的参数：source: Observable<T>。返回类型派生自Subscriber<R>.

Observable.lift只是创建一个新的Observable. 这就像在喜欢的列表中创建节点。

protected lift<R>(operator?: Operator<T, R>): Observable<R> {
  const observable = new Observable<R>();
  
  // it's important to keep track of the source !
  observable.source = this;
  observable.operator = operator;
  return observable;
}

因此，运算符（如map）将返回一个函数。调用该函数的是pipeFromArray函数：

export function pipeFromArray<T, R>(fns: Array<UnaryFunction<T, R>>): UnaryFunction<T, R> {
  if (fns.length === 0) {
    return identity as UnaryFunction<any, any>;
  }

  if (fns.length === 1) {
    return fns[0];
  }

  return function piped(input: T): R {
    // here the functions returned by the operators are being called
    return fns.reduce((prev: any, fn: UnaryFunction<T, R>) => fn(prev), input as any);
  };
}

在上面的代码片段中，fn是operate函数返回的内容：

return (source: Observable<T>) => {
  if (hasLift(source)) { // has `lift` method
    return source.lift(function (this: Subscriber<R>, liftedSource: Observable<T>) {
      try {
        return init(liftedSource, this);
      } catch (err) {
        this.error(err);
      }
    });
  }
  throw new TypeError('Unable to lift unknown Observable type');
};

也许最好也看一个例子。我建议您自己使用调试器尝试一下。

const src$ = new Observable(subscriber => {subscriber.next(1), subscriber.complete()});

该subscriber => {}回调FN将被分配到的Observable._subscribe财产。

constructor(subscribe?: (this: Observable<T>, subscriber: Subscriber<T>) => TeardownLogic) {
  if (subscribe) {
    this._subscribe = subscribe;
  }
}

接下来，让我们尝试添加一个运算符：

const src2$ = src$.pipe(map(num => num ** 2))

在这种情况下，它将从pipeFromArray以下位置调用此块：

// `pipeFromArray`
if (fns.length === 1) {
  return fns[0];
}

// `Observable.pipe`
pipe(...operations: OperatorFunction<any, any>[]): Observable<any> {
  return operations.length ? pipeFromArray(operations)(this) : this;
}

因此，Observable.pipe将调用(source: Observable<T>) => { ... }，这里source是src$ Observable。通过调用该函数（其结果存储在 中src2$），它还将调用该Observable.lift方法。

return source.lift(function (this: Subscriber<R>, liftedSource: Observable<T>) {
  try {
    return init(liftedSource, this);
  } catch (err) {
    this.error(err);
  }
});

/* ... */

protected lift<R>(operator?: Operator<T, R>): Observable<R> {
  const observable = new Observable<R>();
  observable.source = this;
  observable.operator = operator;
  return observable;
}

在这一点上，src$是一个Observable实例，它具有source集合 tosrc$和operator集合 to function (this: Subscriber<R>, liftedSource: Observable<T>) ...。

从我的角度来看，这都是关于链表的。创建Observable链时（通过添加运算符），列表是从上到下创建的。
当尾节点subscribe调用其方法时，将创建另一个列表，这次是从下到上。我喜欢称第一个为Observable list，第二个为Subscribers list。

src2$.subscribe(console.log)

这subscribe是调用方法时发生的情况：

const subscriber = isSubscriber(observerOrNext) ? observerOrNext : new SafeSubscriber(observerOrNext, error, complete);
  
  const { operator, source } = this;
  subscriber.add(
    operator
      ? operator.call(subscriber, source)
      : source || config.useDeprecatedSynchronousErrorHandling
      ? this._subscribe(subscriber)
      : this._trySubscribe(subscriber)
  );

  return subscriber;

在这种情况下src2$有一个operator，所以它会调用它。operator定义为：

function (this: Subscriber<R>, liftedSource: Observable<T>) {
  try {
    return init(liftedSource, this);
  } catch (err) {
    this.error(err);
  }
}

其中init取决于使用的运算符。再次，这里map的init

export function map<T, R>(project: (value: T, index: number) => R, thisArg?: any): OperatorFunction<T, R> {
  return operate( /* THIS IS `init()` */(source, subscriber) => {
    
    // The index of the value from the source. Used with projection.
    let index = 0;
    // Subscribe to the source, all errors and completions are sent along
    // to the consumer.
    source.subscribe(
      new OperatorSubscriber(subscriber, (value: T) => {
        // Call the projection function with the appropriate this context,
        // and send the resulting value to the consumer.
        subscriber.next(project.call(thisArg, value, index++));
      })
    );
  });
}

source事实上是src$。当source.subscribe()被调用时，它将最终调用提供给 的回调new Observable(subscriber => { ... })。调用将从上面subscriber.next(1)调用(value: T) => { ... }，这将调用subscriber.next(project.call(thisArg, value, index++));( project- 提供给 的回调map)。最后，subscriber.next参考console.log。

回到Subject，这_subscribe是调用该方法时发生的情况：

protected _subscribe(subscriber: Subscriber<T>): Subscription {
  this._throwIfClosed(); // if unsubscribed
  this._checkFinalizedStatuses(subscriber); // `error` or `complete` notifications
  return this._innerSubscribe(subscriber);
}

protected _innerSubscribe(subscriber: Subscriber<any>) {
  const { hasError, isStopped, observers } = this;
  return hasError || isStopped
    ? EMPTY_SUBSCRIPTION
    : (observers.push(subscriber), new Subscription(() => arrRemove(this.observers, subscriber)));
}

因此，这就是Subject's订阅者列表的填充方式。通过返回new Subscription(() => arrRemove(this.observers, subscriber))，它确保订阅者取消订阅（由于complete/error通知或只是subscriber.unsubscribe()），不活动的订阅者将从Subject的列表中删除。