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();
 }