Typing fixes

src/Observable.ts

@@ -91,7 +91,7 @@ export class Observable<T> implements CoreOperators<T>  {
    * @description creates a new Observable, with this Observable as the source, and the passed
    * operator defined as the new observable's operator.
    */
-  lift<T, R>(operator: Operator<T, R>): Observable<R> {
+  lift<R>(operator: Operator<T, R>): Observable<R> {
     const observable = new Observable<R>();
     observable.source = this;
     observable.operator = operator;

src/Operator.ts

@@ -1,7 +1,7 @@
 import {Subscriber} from './Subscriber';
 
 export class Operator<T, R> {
-  call<T, R>(subscriber: Subscriber<R>): Subscriber<T> {
+  call(subscriber: Subscriber<R>): Subscriber<T> {
     return new Subscriber<T>(subscriber);
   }
 }

src/observable/dom/WebSocketSubject.ts

@@ -61,7 +61,7 @@ export class WebSocketSubject<T> extends Subject<T> {
 
   lift<R>(operator: Operator<T, R>) {
     const sock: WebSocketSubject<T> = new WebSocketSubject(this, this.destination);
-    sock.operator = operator;
+    sock.operator = <any>operator;
     return sock;
   }
 

src/operator/combineLatest.ts

@@ -35,7 +35,7 @@ export function combineLatest<T, R>(...observables: Array<ObservableInput<any> |
 
   observables.unshift(this);
 
-  return new ArrayObservable(observables).lift<T, R>(new CombineLatestOperator<T, R>(project));
+  return new ArrayObservable(observables).lift(new CombineLatestOperator(project));
 }
 
 /* tslint:disable:max-line-length */
@@ -107,7 +107,7 @@ export function combineLatestStatic<T, R>(...observables: Array<any | Observable
     observables = <Array<Observable<any>>>observables[0];
   }
 
-  return new ArrayObservable(observables, scheduler).lift<T, R>(new CombineLatestOperator<T, R>(project));
+  return new ArrayObservable(observables, scheduler).lift(new CombineLatestOperator<T, R>(project));
 }
 
 export class CombineLatestOperator<T, R> implements Operator<T, R> {

src/operator/concat.ts

@@ -30,5 +30,5 @@ export function concatStatic<T, R>(...observables: Array<Observable<any> | Sched
     scheduler = args.pop();
   }
 
-  return new ArrayObservable(observables, scheduler).lift<Observable<T>, T | R>(new MergeAllOperator<T | R>(1));
+  return new ArrayObservable(observables, scheduler).lift(new MergeAllOperator<T | R>(1));
 }

src/operator/defaultIfEmpty.ts

@@ -11,20 +11,20 @@ export function defaultIfEmpty<T, R>(defaultValue: R = null): Observable<T | R>
   return this.lift(new DefaultIfEmptyOperator(defaultValue));
 }
 
-class DefaultIfEmptyOperator<T, R> implements Operator<T, R> {
+class DefaultIfEmptyOperator<T, R> implements Operator<T, T | R> {
 
   constructor(private defaultValue: R) {
   }
 
-  call(subscriber: Subscriber<T>): Subscriber<T> {
+  call(subscriber: Subscriber<T | R>): Subscriber<T> {
     return new DefaultIfEmptySubscriber(subscriber, this.defaultValue);
   }
 }
 
 class DefaultIfEmptySubscriber<T, R> extends Subscriber<T> {
   private isEmpty: boolean = true;
 
-  constructor(destination: Subscriber<T>, private defaultValue: R) {
+  constructor(destination: Subscriber<T | R>, private defaultValue: R) {
     super(destination);
   }
 

src/operator/distinct.ts

@@ -13,11 +13,11 @@ import {Subscription} from '../Subscription';
  * @param {Observable} [flushes] optional Observable for flushing the internal HashSet of the operator.
  * @returns {Observable} an Observable that emits items from the source Observable with distinct values.
  */
-export function distinct<T>(compare?: (x: T, y: T) => boolean, flushes?: Observable<any>) {
+export function distinct<T>(compare?: (x: T, y: T) => boolean, flushes?: Observable<any>): Observable<T> {
   return this.lift(new DistinctOperator(compare, flushes));
 }
 
-class DistinctOperator<T, R> implements Operator<T, R> {
+class DistinctOperator<T> implements Operator<T, T> {
   constructor(private compare: (x: T, y: T) => boolean, private flushes: Observable<any>) {
   }
 

src/operator/every.ts

@@ -39,7 +39,7 @@ export function every<T>(predicate: (value: T, index: number, source: Observable
   return source.lift(new EveryOperator(predicate, thisArg, source));
 }
 
-class EveryOperator<T, R> implements Operator<T, R> {
+class EveryOperator<T, R> implements Operator<T, boolean> {
   constructor(private predicate: (value: T, index: number, source: Observable<T>) => boolean,
               private thisArg?: any,
               private source?: Observable<T>) {

src/operator/groupBy.ts

@@ -31,7 +31,7 @@ export interface RefCountSubscription {
   attemptedToUnsubscribe: boolean;
 }
 
-class GroupByOperator<T, K, R> extends Operator<T, R> {
+class GroupByOperator<T, K, R> extends Operator<T, GroupedObservable<K, R>> {
   constructor(public source: Observable<T>,
               private keySelector: (value: T) => K,
               private elementSelector?: (value: T) => R,

src/operator/inspect.ts

@@ -12,7 +12,7 @@ export function inspect<T>(durationSelector: (value: T) => Observable<any> | Pro
   return this.lift(new InspectOperator(durationSelector));
 }
 
-class InspectOperator<T, R> implements Operator<T, R> {
+class InspectOperator<T> implements Operator<T, T> {
   constructor(private durationSelector: (value: T) => Observable<any> | Promise<any>) {
   }
 

src/operator/inspectTime.ts

@@ -9,7 +9,7 @@ export function inspectTime<T>(delay: number, scheduler: Scheduler = asap): Obse
   return this.lift(new InspectTimeOperator(delay, scheduler));
 }
 
-class InspectTimeOperator<T, R> implements Operator<T, R> {
+class InspectTimeOperator<T> implements Operator<T, T> {
   constructor(private delay: number, private scheduler: Scheduler) {
   }
 

src/operator/isEmpty.ts

@@ -13,15 +13,14 @@ export function isEmpty(): Observable<boolean> {
   return this.lift(new IsEmptyOperator());
 }
 
-class IsEmptyOperator<T> implements Operator<boolean, boolean> {
-  call (observer: Subscriber<T>): Subscriber<boolean> {
+class IsEmptyOperator implements Operator<any, boolean> {
+  call (observer: Subscriber<boolean>): Subscriber<any> {
     return new IsEmptySubscriber(observer);
   }
 }
 
-class IsEmptySubscriber extends Subscriber<boolean> {
-
-  constructor(destination: Subscriber<any>) {
+class IsEmptySubscriber extends Subscriber<any> {
+  constructor(destination: Subscriber<boolean>) {
     super(destination);
   }
 

src/operator/merge.ts

@@ -34,5 +34,5 @@ export function mergeStatic<T, R>(...observables: Array<Observable<any> | Schedu
     return <Observable<R>>observables[0];
   }
 
-  return new ArrayObservable<Observable<T>>(<any>observables, scheduler).lift<Observable<T>, R>(new MergeAllOperator<R>(concurrent));
+  return new ArrayObservable(<any>observables, scheduler).lift(new MergeAllOperator<R>(concurrent));
 }

src/operator/mergeMapTo.ts

@@ -13,13 +13,15 @@ export function mergeMapTo<T, R, R2>(observable: Observable<R>,
   return this.lift(new MergeMapToOperator(observable, <any>resultSelector, concurrent));
 }
 
+// TODO: Figure out correct signature here: an Operator<Observable<T>, R2>
+//       needs to implement call(observer: Subscriber<R2>): Subscriber<Observable<T>>
 export class MergeMapToOperator<T, R, R2> implements Operator<Observable<T>, R2> {
   constructor(private ish: Observable<R> | Promise<R>,
               private resultSelector?: (outerValue: T, innerValue: R, outerIndex: number, innerIndex: number) => R2,
               private concurrent: number = Number.POSITIVE_INFINITY) {
-    }
+  }
 
-  call(observer: Subscriber<R2>): Subscriber<T> {
+  call(observer: Subscriber<R2>): Subscriber<any> {
     return new MergeMapToSubscriber(observer, this.ish, this.resultSelector, this.concurrent);
   }
 }

src/operator/pairwise.ts

@@ -11,12 +11,12 @@ import {Subscriber} from '../Subscriber';
  *
  * @returns {Observable<R>} an observable of pairs of values.
  */
-export function pairwise<T>(): Observable<T> {
+export function pairwise<T>(): Observable<[T, T]> {
   return this.lift(new PairwiseOperator());
 }
 
-class PairwiseOperator<T, R> implements Operator<T, R> {
-  call(subscriber: Subscriber<T>): Subscriber<T> {
+class PairwiseOperator<T> implements Operator<T, [T, T]> {
+  call(subscriber: Subscriber<[T, T]>): Subscriber<T> {
     return new PairwiseSubscriber(subscriber);
   }
 }
@@ -25,7 +25,7 @@ class PairwiseSubscriber<T> extends Subscriber<T> {
   private prev: T;
   private hasPrev: boolean = false;
 
-  constructor(destination: Subscriber<T>) {
+  constructor(destination: Subscriber<[T, T]>) {
     super(destination);
   }
 

src/operator/race.ts

@@ -44,7 +44,7 @@ export function raceStatic<T>(...observables: Array<Observable<T> | Array<Observ
   return new ArrayObservable(observables).lift(new RaceOperator());
 }
 
-export class RaceOperator<T, R> implements Operator<T, R> {
+export class RaceOperator<T> implements Operator<T, T> {
   call(subscriber: Subscriber<T>): Subscriber<T> {
     return new RaceSubscriber(subscriber);
   }

src/operator/reduce.ts

@@ -27,7 +27,7 @@ export class ReduceOperator<T, R> implements Operator<T, R> {
   constructor(private project: (acc: R, value: T) => R, private seed?: R) {
   }
 
-  call(subscriber: Subscriber<T>): Subscriber<T> {
+  call(subscriber: Subscriber<R>): Subscriber<T> {
     return new ReduceSubscriber(subscriber, this.project, this.seed);
   }
 }
@@ -39,7 +39,7 @@ export class ReduceSubscriber<T, R> extends Subscriber<T> {
   hasValue: boolean = false;
   project: (acc: R, value: T) => R;
 
-  constructor(destination: Subscriber<T>, project: (acc: R, value: T) => R, seed?: R) {
+  constructor(destination: Subscriber<R>, project: (acc: R, value: T) => R, seed?: R) {
     super(destination);
     this.acc = seed;
     this.project = project;

src/operator/scan.ts

@@ -21,7 +21,7 @@ class ScanOperator<T, R> implements Operator<T, R> {
   constructor(private accumulator: (acc: R, x: T) => R, private seed?: T | R) {
   }
 
-  call(subscriber: Subscriber<T>): Subscriber<T> {
+  call(subscriber: Subscriber<R>): Subscriber<T> {
     return new ScanSubscriber(subscriber, this.accumulator, this.seed);
   }
 }
@@ -40,7 +40,7 @@ class ScanSubscriber<T, R> extends Subscriber<T> {
 
   private accumulatorSet: boolean = false;
 
-  constructor(destination: Subscriber<T>, private accumulator: (acc: R, x: T) => R, seed?: T|R) {
+  constructor(destination: Subscriber<R>, private accumulator: (acc: R, x: T) => R, seed?: T|R) {
     super(destination);
     this.seed = seed;
     this.accumulator = accumulator;

src/operator/skipWhile.ts

@@ -16,7 +16,7 @@ export function skipWhile<T>(predicate: (value: T, index: number) => boolean): O
   return this.lift(new SkipWhileOperator(predicate));
 }
 
-class SkipWhileOperator<T, R> implements Operator<T, R> {
+class SkipWhileOperator<T> implements Operator<T, T> {
   constructor(private predicate: (value: T, index: number) => boolean) {
   }
 

src/operator/withLatestFrom.ts

@@ -58,7 +58,7 @@ class WithLatestFromOperator<T, R> implements Operator<T, R> {
               private project?: (...values: any[]) => Observable<R>) {
   }
 
-  call(subscriber: Subscriber<T>): Subscriber<T> {
+  call(subscriber: Subscriber<R>): Subscriber<T> {
     return new WithLatestFromSubscriber(subscriber, this.observables, this.project);
   }
 }
@@ -67,7 +67,7 @@ class WithLatestFromSubscriber<T, R> extends OuterSubscriber<T, R> {
   private values: any[];
   private toRespond: number[] = [];
 
-  constructor(destination: Subscriber<T>,
+  constructor(destination: Subscriber<R>,
               private observables: Observable<any>[],
               private project?: (...values: any[]) => Observable<R>) {
     super(destination);

src/operator/zip.ts

@@ -57,7 +57,7 @@ export function zipStatic<T, R>(...observables: Array<ObservableInput<any> | ((.
   if (typeof project === 'function') {
     observables.pop();
   }
-  return new ArrayObservable(observables).lift<T, R>(new ZipOperator<T, R>(project));
+  return new ArrayObservable(observables).lift(new ZipOperator(project));
 }
 
 export class ZipOperator<T, R> implements Operator<T, R> {