throttleTime: fixes and tests

spec/operators/throttleTime-spec.js

@@ -0,0 +1,114 @@
+/* globals describe, it, expect, expectObservable, expectSubscription, hot, cold */
+var Rx = require('../../dist/cjs/Rx');
+var Observable = Rx.Observable;
+var Scheduler = Rx.Scheduler;
+
+describe('Observable.prototype.throttleTime()', function () {
+  it('should throttle events by 50 time units', function (done) {
+    Observable.of(1, 2, 3).throttleTime(50)
+      .subscribe(function (x) {
+        expect(x).toBe(1);
+      }, null, done);
+  });
+
+  it('should throttle events multiple times', function (done) {
+    var expected = ['1-0', '2-0'];
+    Observable.concat(
+      Observable.timer(0, 10).take(3).map(function (x) { return '1-' + x; }),
+      Observable.timer(80, 10).take(5).map(function (x) { return '2-' + x; })
+      )
+      .throttleTime(50)
+      .subscribe(function (x) {
+        expect(x).toBe(expected.shift());
+      }, null, done);
+  });
+
+  it('should simply mirror the source if values are not emitted often enough', function () {
+    var e1 =   hot('-a--------b-----c----|');
+    var subs =     '^                    !';
+    var expected = '-a--------b-----c----|';
+
+    expectObservable(e1.throttleTime(50, rxTestScheduler)).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(subs);
+  });
+
+  it('should immediately emit the first value in each time window', function () {
+    var e1 =   hot('-a-x-y----b---x-cx---|');
+    var subs =     '^                    !';
+    var expected = '-a--------b-----c----|';
+
+    expectObservable(e1.throttleTime(50, rxTestScheduler)).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(subs);
+  });
+
+  it('should handle a busy producer emitting a regular repeating sequence', function () {
+    var e1 =   hot('abcdefabcdefabcdefabcdefa|');
+    var subs =     '^                        !';
+    var expected = 'a-----a-----a-----a-----a|';
+
+    expectObservable(e1.throttleTime(50, rxTestScheduler)).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(subs);
+  });
+
+  it('should complete when source does not emit', function () {
+    var e1 =   hot('-----|');
+    var subs =     '^    !';
+    var expected = '-----|';
+
+    expectObservable(e1.throttleTime(50, rxTestScheduler)).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(subs);
+  });
+
+  it('should raise error when source does not emit and raises error', function () {
+    var e1 =   hot('-----#');
+    var expected = '-----#';
+
+    expectObservable(e1.throttleTime(10, rxTestScheduler)).toBe(expected);
+  });
+
+  it('should handle an empty source', function () {
+    var e1 =  cold('|');
+    var subs =     '(^!)';
+    var expected = '|';
+
+    expectObservable(e1.throttleTime(30, rxTestScheduler)).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(subs);
+  });
+
+  it('should handle a never source', function () {
+    var e1 =  cold('-');
+    var subs =     '^';
+    var expected = '-';
+
+    expectObservable(e1.throttleTime(30, rxTestScheduler)).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(subs);
+  });
+
+  it('should handle a throw source', function () {
+    var e1 =  cold('#');
+    var subs =     '(^!)';
+    var expected = '#';
+
+    expectObservable(e1.throttleTime(30, rxTestScheduler)).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(subs);
+  });
+
+  it('should throttle and does not complete when source does not completes', function () {
+    var e1 =   hot('-a--(bc)-------d----------------');
+    var unsub =    '                               !';
+    var subs =     '^                              !';
+    var expected = '-a-------------d----------------';
+
+    expectObservable(e1.throttleTime(50, rxTestScheduler), unsub).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(subs);
+  });
+
+  it('should throttle values until source raises error', function () {
+    var e1 =   hot('-a--(bc)-------d---------------#');
+    var subs =     '^                              !';
+    var expected = '-a-------------d---------------#';
+
+    expectObservable(e1.throttleTime(50, rxTestScheduler)).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(subs);
+  });
+});

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