observe

observe.ts

import { Observable } from 'rxjs/Observable';
import { Observer } from 'rxjs/Observer';
import { combineLatest } from 'rxjs/observable/combineLatest';

export function observe<T>(variables?: Record<string, any>): Observable<T> {
  const keys = Object.keys(variables);

  return Observable.create((observer: Observer<any>) => {
    combineLatest(mapVariablesToObservables(variables)).subscribe(values => {
      const resultVariables: Record<string, T> = {};

      values.forEach((value, i) => {
        const key = keys[i];
        resultVariables[key] = value;
      });

      observer.next(resultVariables);
    });
  });
}

function mapVariablesToObservables<T>(
  variables?: Record<string, T>,
): Observable<T>[] {
  return Object.keys(variables).map(key =>
    getVariableToObservable(variables[key]),
  );
}

function getVariableToObservable<T>(
  variable: T | Observable<T>,
): Observable<T> {
  if (variable instanceof Observable) {
    return variable;
  } else if (typeof variable !== 'undefined') {
    return new Observable(subscriber => {
      subscriber.next(variable);
    });
  } else {
    return new Observable(subscriber => {
      subscriber.next(null);
    });
  }
}