Method subscribe

packages/core/src/event_emitter.ts:144–175

(observerOrNext?: any, error?: any, complete?: any)

  }

  override subscribe(observerOrNext?: any, error?: any, complete?: any): Subscription {
    let nextFn = observerOrNext;
    let errorFn = error || (() => null);
    let completeFn = complete;

    if (observerOrNext && typeof observerOrNext === 'object') {
      const observer = observerOrNext as PartialObserver<unknown>;
      nextFn = observer.next?.bind(observer);
      errorFn = observer.error?.bind(observer);
      completeFn = observer.complete?.bind(observer);
    }

    if (this.__isAsync) {
      errorFn = this.wrapInTimeout(errorFn);

      if (nextFn) {
        nextFn = this.wrapInTimeout(nextFn);
      }

      if (completeFn) {
        completeFn = this.wrapInTimeout(completeFn);
      }
    }

    const sink = super.subscribe({next: nextFn, error: errorFn, complete: completeFn});

    if (observerOrNext instanceof Subscription) {
      observerOrNext.add(sink);
    }

    return sink;
  }

  private wrapInTimeout(fn: (value: unknown) => any) {
    return (value: unknown) => {