Demystifying RxJS, Part I: from()

04-demystifying-rxjs-from.ts

function from<T>(convertible: Promise<T> | ArrayLike<T>): Observable<T> {
  if (typeof (convertible as any).length === "number") {
    const arrayLike = convertible as ArrayLike<T>;
    return new Observable(obs => {
      for (let i = 0; i < arrayLike.length; i++) {
        obs.next(arrayLike[i]);
      }
      obs.complete();
    });
  }
  
  // Otherwise, it's a promise
  const promise = convertible as Promise<T>;
  return new Observable(obs => {
    promise
      .then(x => {
        obs.next(x);
        obs.complete();
      })
      .catch(err => obs.error(err));
  });
}

// Logs:
// fromArray: 4
// fromArray: 5
// fromArray: 6
from([4, 5, 6]).subscribe(x => console.log("fromArray:", x));

// Waits 1.1 seconds, then logs:
// fromPromise: 42
from(new Promise(resolve => setTimeout(() => resolve(42), 1100))).subscribe(
  x => {
    console.log("fromPromise:", x);
  }
);