tiny observable implementation

Observable.ts

export interface Subscription {
  readonly closed: boolean;
  unsubscribe: () => void;
}

export interface Observer<T> {
  closed: boolean;
  error: (error: any) => void;
  complete: () => void;
  next: (value: T) => void;
}

export type SubscriberFunction<T> = (
  observer: Observer<T>,
) => { unsubscribe: () => void } | void;

export default class Observable<T> {
  private _subscribe: SubscriberFunction<T>;

  public constructor(subscribe: SubscriberFunction<T>) {
    this._subscribe = subscribe;
  }

  subscribe(
    nextOrObserver: Partial<Observer<T>> | ((value?: T) => void),
    error?: (errorValue: Error) => void,
    complete?: () => void,
  ) {
    let closed = false;

    const observer =
      typeof nextOrObserver === 'function'
        ? { next: nextOrObserver, error, complete }
        : nextOrObserver;

    // eslint-disable-next-line prefer-const
    let subscriber: void | { unsubscribe: () => void };

    const subscription: Subscription = {
      get closed() {
        return closed;
      },
      unsubscribe() {
        if (closed) return;
        closed = true;
        subscriber?.unsubscribe();
      },
    };

    subscriber = this._subscribe({
      closed,
      error(err?: any) {
        if (closed) return;
        subscription.unsubscribe();
        observer.error?.(err);
      },
      complete() {
        if (closed) return;
        try {
          subscription.unsubscribe();
          observer.complete?.();
        } catch (err) {
          subscription.unsubscribe();
          observer.error?.(err);
        }
      },
      next(value: T) {
        if (closed) return;
        try {
          observer.next?.(value);
        } catch (err) {
          subscription.unsubscribe();
          observer.error?.(err);
        }
      },
    });

    return subscription;
  }
}

Examples of use

function fromIterable<T>(source: Iterable<T>): Observable<T> {
  return new Observable<T>(observer => { 
     for (let item of source) {
       observer.next(item)
     }
     observer.complete()
  })
}

const observable = fromIterable([1, 2, 3, 4])

function *generateNumbers() {
  let i = 0
  while (true) yield ++i
}
const infiniteObservable = fromIterable(generateNumbers())