observable_operators.ts

type Operator<TSource, TTarget> = (source: Observable<TSource>) => Observable<TTarget>;

function pipe(this: Observable<any>, ...operators: Operator<any, any>[]): Observable<any> {
  return operators.reduce((acc, curr) => curr(acc), this);
}

// replace native pipe() method
Observable.prototype.pipe = pipe;

function map<TSource, TTarget>(transform: (TSource) => TTarget): Operator<TSource, TTarget> {
  
  return function(upstream) {
    return new Observable(subscriber => {
      return upstream.subscribe({
        next(value) { subscriber.next(transform(value)); },
        complete() { subscriber.complete(); },
        error(err) { subscriber.error(err); }
      });
    });
  };
}

function filter<TSource>(predicate: (TSource) => boolean): Operator<TSource, TSource> {
  
  return function(upstream) {
    return new Observable(subscriber => {
      return upstream.subscribe({
        next(value) {
          if (predicate(value)) {
            subscriber.next(value);
          }
        },
        complete() { subscriber.complete(); },
        error(err) { subscriber.error(err); }
      });
    });
  };
}