feat(operator): add throttle

closes #191

Author: benlesh

Diff for: spec/operators/throttle-spec.js

@@ -0,0 +1,19 @@
+/* globals describe, it, expect */
+var Rx = require('../../dist/cjs/Rx');
+var Observable = Rx.Observable;
+var Scheduler = Rx.Scheduler;
+
+describe('Observable.prototype.throttle()', function () {
+  it('should delay calls by the specified amount', function (done) {
+    var expected = [3, 4];
+    var source = Observable.concat(Observable.value(1),
+      Observable.timer(10).mapTo(2),
+      Observable.timer(10).mapTo(3),
+      Observable.timer(100).mapTo(4)
+      )
+      .throttle(50)
+      .subscribe(function (x) {
+        expect(x).toBe(expected.shift());
+      }, null, done);
+  });
+});

Diff for: spec/scheduler-spec.js

@@ -17,4 +17,20 @@ describe('Scheduler.immediate', function() {
     expect(call1).toBe(true);
     expect(call2).toBe(true);
   });
+  
+  it('should schedule things in the future too', function (done) {
+    var called = false;
+    Scheduler.immediate.schedule(500, null, function () {
+      called = true;
+    });
+    
+    setTimeout(function () {
+      expect(called).toBe(false);
+    }, 400);
+    
+    setTimeout(function() {
+      expect(called).toBe(true);
+      done();
+    }, 700);
+  })
 });

Diff for: src/Observable.ts

@@ -147,6 +147,7 @@ export default class Observable<T> {
   // HACK: this should be Observable<Notification<T>>, but the build process didn't like it. :(
   //   this will be fixed when we can move everything to the TypeScript compiler I suspect.
   materialize: () => Observable<any>;
+  throttle: (delay: number, scheduler?: Scheduler) => Observable<T>;
 
   observeOn: (scheduler: Scheduler, delay?: number) => Observable<T>;
   subscribeOn: (scheduler: Scheduler, delay?: number) => Observable<T>;

Diff for: src/Rx.ts

@@ -162,9 +162,13 @@ import groupBy from './operators/groupBy';
 observableProto.groupBy = groupBy;
 
 import delay from './operators/delay';
+import throttle from './operators/throttle';
+
 observableProto.delay = delay;
+observableProto.throttle = throttle;
 
 export {
+
     Subject,
     Scheduler,
     Observable,

Diff for: src/operators/throttle.ts

@@ -0,0 +1,55 @@
+import Operator from '../Operator';
+import Observer from '../Observer';
+import Subscriber from '../Subscriber';
+import Observable from '../Observable';
+import Scheduler from '../Scheduler';
+import Subscription from '../Subscription';
+
+import tryCatch from '../util/tryCatch';
+import {errorObject} from '../util/errorObject';
+import bindCallback from '../util/bindCallback';
+
+export default function throttle<T>(delay: number, scheduler: Scheduler = Scheduler.nextTick) {
+  return this.lift(new ThrottleOperator(delay, scheduler));
+}
+
+export class ThrottleOperator<T, R> extends Operator<T, R> {
+  constructor(private delay:number, private scheduler:Scheduler) {
+    super();
+  }
+  
+  call(observer: Observer<R>): Observer<T> {
+    return new ThrottleSubscriber(observer, this.delay, this.scheduler);
+  }
+}
+
+export class ThrottleSubscriber<T, R> extends Subscriber<T> {
+  private throttled: Subscription<any>;
+  
+  constructor(destination:Observer<T>, private delay:number, private scheduler:Scheduler) {
+    super(destination);
+  }
+
+  _next(x) {
+    this.clearThrottle();
+    this.add(this.throttled = this.scheduler.schedule(this.delay, { value: x, subscriber: this }, dispatchNext));
+  }
+  
+  throttledNext(x) {
+    this.clearThrottle();  
+    this.destination.next(x);
+  }
+  
+  clearThrottle() {
+    const throttled = this.throttled;
+    if (throttled) {
+      this.remove(throttled);
+      throttled.unsubscribe();
+      this.throttled = null;  
+    }
+  }
+}
+
+function dispatchNext({ value, subscriber }) {
+  subscriber.throttledNext(value);
+}