123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133
import {  SubscriberFunction,  Subscription,  SubscriptionLike} from "./Subscription.ts";import { Observer } from "./Observer.ts";import {  isAsyncIterable,  isIterable,  isObservableLike} from "./symbol.ts";import { isObserver, assertIsObserver } from "./utils.ts";
/** * Interface that aligns with signatures to an object that is an Observable */export interface ObservableLike<Value = unknown> {  [Symbol.observable](): Observable<Value>;}
/** * An Observable represents a sequence of values which may be observed. */export class Observable<Value = unknown> implements ObservableLike<Value> {  private _subscriber: SubscriberFunction<Value>;  constructor(subscriber: SubscriberFunction<Value>) {    if (!(this instanceof Observable)) {      throw new TypeError("Observable cannot be called as a function");    }    if (typeof subscriber !== "function") {      throw new TypeError("Observable initializer must be a function");    }    this._subscriber = subscriber;  }
  /**   * Subscribes to the sequence with an observer   * @param observer the observer to be subscribed   */  subscribe(observer: Observer<Value>): SubscriptionLike;  /**   * Subscribes to the sequence with callbacks   * @param onNext callback to be invoked on the next value of the sequence   * @param onError callback to be invoked on the next error of the sequence   * @param onComplete callback to be invoked when the sequence is completed   */  subscribe(    onNext: (value: Value) => void,    onError?: (error: unknown) => void,    onComplete?: () => void,  ): SubscriptionLike;  subscribe(    observerOrOnNext: Observer<Value> | ((value: Value) => void),    onError?: (error: unknown) => void,    onComplete?: () => void,  ): SubscriptionLike {    const observer = isObserver(observerOrOnNext) ? observerOrOnNext : {      next: observerOrOnNext,      error: onError,      complete: onComplete,    };    assertIsObserver(observer);    return new Subscription(observer, {subscriber: this._subscriber});  }
  /** Returns itself */  [Symbol.observable](): Observable<Value> {    return this;  }
  /**   * Converts arguments to an Observable   * @param items an iterablet   */  static of<Value = unknown>(...items: Value[]): Observable<Value> {    return new Observable<Value>((observer) => {      for (const iterator of items) observer.next(iterator);      observer.complete();    });  }
  /**   * Converts an observable  Observable   * @param observable the observableLike to be converted   */  static from<Value = unknown>(    observable: ObservableLike<Value>,  ): Observable<Value>;  /**   * Converts an iterable to an Observable   * @param iterable the iterable to be converted   */  static from<Value = unknown>(    iterable: Iterable<Value>,  ): Observable<Value>;  /**   * Converts an async iterable to an Observable   * @param iterable the iterable to be converted   */  static from<Value = unknown>(    iterable: AsyncIterable<Value>,  ): Observable<Value>;  static from<Value = unknown>(    argument:      | Iterable<Value>      | AsyncIterable<Value>      | ObservableLike<Value>,  ): Observable<Value> {    if (isObservableLike(argument)) {      const observable = argument[Symbol.observable]();      return new Observable<Value>((observer) =>        observable.subscribe(observer)      );    }    if (isIterable(argument)) {      return new Observable<Value>((observer) => {        for (const value of argument) observer.next(value);        observer.complete();      });    }    if (isAsyncIterable(argument)) {      return new Observable<Value>((observe) => {        (async () => {          for await (const value of argument) observe.next(value);        })();        observe.complete();      });    }    throw new TypeError(      "Observable.from expects to receive an ObservableLike, Iterable or AsyncIterable",    );  }}