Demystifying RxJS Part I: Observable Implementation

01-demystifying-rxjs-observable.ts

interface Observer<T> {
  next(value: T): void;
  error(err: any): void;
  complete(): void;
}

interface Subscription {
  unsubscribe(): void;
}
  
class Observable<T> {
  constructor(
    private observe: (observer: Observer<T>) => (() => void) | void
  ) {}

  subscribe(
    next?: (value: T) => void,
    error?: (err: any) => void,
    complete?: () => void
  ): Subscription {
    const observer: Observer<T> = {
      next: (value: T) => {
        if (next) next(value)
      },
      error: (err: any) => {
        if (error) error(err);
      },
      complete: () => {
        if (complete) complete();
      }
    };

    const teardown = this.observe(observer);
    return {
      unsubscribe: () => {
        if (teardown) teardown();
      }
    };
  }
}