fix(throttleTime): fix and rename throttleTime operator

Rename throttle to throttleTime. Fix this operator to correctly
take in the first 'next' value but
ignore subsequent 'next' values for a specified time period.

Related to issue #666.

src/CoreOperators.ts

@@ -75,7 +75,7 @@ export interface CoreOperators<T> {
   switchMapTo?: <R>(observable: Observable<any>, projectResult?: (x: T, y: any, ix: number, iy: number) => R) => Observable<R>;
   take?: (count: number) => Observable<T>;
   takeUntil?: (notifier: Observable<any>) => Observable<T>;
-  throttle?: (delay: number, scheduler?: Scheduler) => Observable<T>;
+  throttleTime?: (delay: number, scheduler?: Scheduler) => Observable<T>;
   timeout?: (due: number | Date, errorToSend?: any, scheduler?: Scheduler) => Observable<T>;
   timeoutWith?: <R>(due: number | Date, withObservable: Observable<R>, scheduler?: Scheduler) => Observable<T> | Observable<R>;
   toArray?: () => Observable<T[]>;

src/Observable.ts

@@ -228,7 +228,7 @@ export class Observable<T> implements CoreOperators<T>  {
   switchMapTo: <R>(observable: Observable<any>, projectResult?: (x: T, y: any, ix: number, iy: number) => R) => Observable<R>;
   take: (count: number) => Observable<T>;
   takeUntil: (notifier: Observable<any>) => Observable<T>;
-  throttle: (delay: number, scheduler?: Scheduler) => Observable<T>;
+  throttleTime: (delay: number, scheduler?: Scheduler) => Observable<T>;
   timeout: (due: number | Date, errorToSend?: any, scheduler?: Scheduler) => Observable<T>;
   timeoutWith: <R>(due: number | Date, withObservable: Observable<R>, scheduler?: Scheduler) => Observable<T> | Observable<R>;
   toArray: () => Observable<T[]>;

src/Rx.KitchenSink.ts

@@ -285,8 +285,8 @@ observableProto.take = take;
 import {takeUntil} from './operators/takeUntil';
 observableProto.takeUntil = takeUntil;
 
-import {throttle} from './operators/throttle';
-observableProto.throttle = throttle;
+import {throttleTime} from './operators/throttleTime';
+observableProto.throttleTime = throttleTime;
 
 import {timeInterval} from './operators/extended/timeInterval';
 observableProto.timeInterval = timeInterval;

src/Rx.ts

@@ -239,8 +239,8 @@ observableProto.take = take;
 import {takeUntil} from './operators/takeUntil';
 observableProto.takeUntil = takeUntil;
 
-import {throttle} from './operators/throttle';
-observableProto.throttle = throttle;
+import {throttleTime} from './operators/throttleTime';
+observableProto.throttleTime = throttleTime;
 
 import {timeout} from './operators/timeout';
 observableProto.timeout = timeout;

src/operators/throttle.ts → src/operators/throttleTime.ts

@@ -4,20 +4,20 @@ import {Scheduler} from '../Scheduler';
 import {Subscription} from '../Subscription';
 import {nextTick} from '../schedulers/nextTick';
 
-export function throttle<T>(delay: number, scheduler: Scheduler = nextTick) {
-  return this.lift(new ThrottleOperator(delay, scheduler));
+export function throttleTime<T>(delay: number, scheduler: Scheduler = nextTick) {
+  return this.lift(new ThrottleTimeOperator(delay, scheduler));
 }
 
-class ThrottleOperator<T, R> implements Operator<T, R> {
+class ThrottleTimeOperator<T, R> implements Operator<T, R> {
   constructor(private delay: number, private scheduler: Scheduler) {
   }
 
   call(subscriber: Subscriber<T>): Subscriber<T> {
-    return new ThrottleSubscriber(subscriber, this.delay, this.scheduler);
+    return new ThrottleTimeSubscriber(subscriber, this.delay, this.scheduler);
   }
 }
 
-class ThrottleSubscriber<T> extends Subscriber<T> {
+class ThrottleTimeSubscriber<T> extends Subscriber<T> {
   private throttled: Subscription<any>;
 
   constructor(destination: Subscriber<T>,
@@ -28,24 +28,21 @@ class ThrottleSubscriber<T> extends Subscriber<T> {
 
   _next(value: T) {
     if (!this.throttled) {
-      this.add(this.throttled = this.scheduler.schedule(dispatchNext, this.delay, { value, subscriber: this }));
+      this.add(this.throttled = this.scheduler.schedule(dispatchNext, this.delay, { subscriber: this }));
+      this.destination.next(value);
     }
   }
 
-  throttledNext(value: T) {
-    this.clearThrottle();
-    this.destination.next(value);
-  }
-
   clearThrottle() {
     const throttled = this.throttled;
     if (throttled) {
       throttled.unsubscribe();
       this.remove(throttled);
+      this.throttled = null;
     }
   }
 }
 
-function dispatchNext<T>({ value, subscriber }) {
-  subscriber.throttledNext(value);
+function dispatchNext<T>({ subscriber }) {
+  subscriber.clearThrottle();
 }