Adding throttleFirst for Issue #352

Author: mattpodwysocki

Gruntfile.js

@@ -292,6 +292,7 @@ var browsers = [{
               'src/core/linq/observable/skipwithtime.js',
               'src/core/linq/observable/skipuntilwithtime.js',
               'src/core/linq/observable/takeuntilwithtime.js',
+              'src/core/linq/observable/throttlefirst.js',
 
               // Transducers
               'src/core/linq/observable/transduce.js',
@@ -564,6 +565,7 @@ var browsers = [{
               'src/core/linq/observable/skipwithtime.js',
               'src/core/linq/observable/skipuntilwithtime.js',
               'src/core/linq/observable/takeuntilwithtime.js',
+              'src/core/linq/observable/throttlefirst.js',
 
               // Experimental Flattening
               'src/core/linq/observable/exclusive.js',
@@ -972,6 +974,7 @@ var browsers = [{
               'src/core/linq/observable/timestamp.js', // timeoutScheduler, select
               'src/core/linq/observable/sample.js', // AnonymousObservable, CompositeDisposable, interval, timeoutScheduler
               'src/core/linq/observable/timeout.js', // AnonymousObservable, timeoutScheduler, throw, SingleAssignmentDisposable, SerialDisposable, CompositeDisposable
+              'src/core/linq/observable/throttlefirst.js',
 
               // Backpressure operators
               'src/core/backpressure/pausable.js',
@@ -1124,6 +1127,7 @@ var browsers = [{
               'src/core/linq/observable/timestamp.js', // timeoutScheduler, select
               'src/core/linq/observable/sample.js', // AnonymousObservable, CompositeDisposable, interval, timeoutScheduler
               'src/core/linq/observable/timeout.js', // AnonymousObservable, timeoutScheduler, throw, SingleAssignmentDisposable, SerialDisposable, CompositeDisposable
+              'src/core/linq/observable/throttlefirst.js',
 
               // Backpressure operators
               'src/core/backpressure/pausable.js',
@@ -1418,6 +1422,7 @@ var browsers = [{
               'src/core/linq/observable/skipwithtime.js',
               'src/core/linq/observable/skipuntilwithtime.js',
               'src/core/linq/observable/takeuntilwithtime.js',
+              'src/core/linq/observable/throttlefirst.js',
               'src/core/suboutro.js'
             ],
             dest: 'dist/rx.time.js'

dist/rx.all.compat.js

@@ -8643,6 +8643,33 @@ if (!Array.prototype.forEach) {
     });
   };
 
+  /**
+   * Returns an Observable that emits only the first item emitted by the source Observable during sequential time windows of a specified duration.
+   * @param {Number} windowDuration time to wait before emitting another item after emitting the last item
+   * @param {Scheduler} [scheduler] the Scheduler to use internally to manage the timers that handle timeout for each item. If not provided, defaults to Scheduler.timeout.
+   * @returns {Observable} An Observable that performs the throttle operation.
+   */
+  observableProto.throttleFirst = function (windowDuration, scheduler) {
+    isScheduler(scheduler) || (scheduler = timeoutScheduler);
+    var duration = +windowDuration || 0;
+    if (duration <= 0) { throw new RangeError('windowDuration cannot be less or equal zero.'); }
+    var source = this;
+    return new AnonymousObservable(function (observer) {
+      var lastOnNext = 0;
+      return source.subscribe(
+        function (x) {
+          var now = scheduler.now();
+          if (lastOnNext === 0 || now - lastOnNext >= duration) {
+            lastOnNext = now;
+            observer.onNext(x);
+          }
+        },
+        observer.onError.bind(observer),
+        observer.onCompleted.bind(observer)
+      );
+    });
+  };
+
   /*
    * Performs a exclusive waiting for the first to finish before subscribing to another observable.
    * Observables that come in between subscriptions will be dropped on the floor.

dist/rx.all.compat.map

@@ -8424,6 +8424,33 @@
     });
   };
 
+  /**
+   * Returns an Observable that emits only the first item emitted by the source Observable during sequential time windows of a specified duration.
+   * @param {Number} windowDuration time to wait before emitting another item after emitting the last item
+   * @param {Scheduler} [scheduler] the Scheduler to use internally to manage the timers that handle timeout for each item. If not provided, defaults to Scheduler.timeout.
+   * @returns {Observable} An Observable that performs the throttle operation.
+   */
+  observableProto.throttleFirst = function (windowDuration, scheduler) {
+    isScheduler(scheduler) || (scheduler = timeoutScheduler);
+    var duration = +windowDuration || 0;
+    if (duration <= 0) { throw new RangeError('windowDuration cannot be less or equal zero.'); }
+    var source = this;
+    return new AnonymousObservable(function (observer) {
+      var lastOnNext = 0;
+      return source.subscribe(
+        function (x) {
+          var now = scheduler.now();
+          if (lastOnNext === 0 || now - lastOnNext >= duration) {
+            lastOnNext = now;
+            observer.onNext(x);
+          }
+        },
+        observer.onError.bind(observer),
+        observer.onCompleted.bind(observer)
+      );
+    });
+  };
+
   /**
    * Executes a transducer to transform the observable sequence
    * @param {Transducer} transducer A transducer to execute

dist/rx.all.compat.min.js

@@ -4217,6 +4217,33 @@ if (!Array.prototype.forEach) {
     });
   };
 
+  /**
+   * Returns an Observable that emits only the first item emitted by the source Observable during sequential time windows of a specified duration.
+   * @param {Number} windowDuration time to wait before emitting another item after emitting the last item
+   * @param {Scheduler} [scheduler] the Scheduler to use internally to manage the timers that handle timeout for each item. If not provided, defaults to Scheduler.timeout.
+   * @returns {Observable} An Observable that performs the throttle operation.
+   */
+  observableProto.throttleFirst = function (windowDuration, scheduler) {
+    isScheduler(scheduler) || (scheduler = timeoutScheduler);
+    var duration = +windowDuration || 0;
+    if (duration <= 0) { throw new RangeError('windowDuration cannot be less or equal zero.'); }
+    var source = this;
+    return new AnonymousObservable(function (observer) {
+      var lastOnNext = 0;
+      return source.subscribe(
+        function (x) {
+          var now = scheduler.now();
+          if (lastOnNext === 0 || now - lastOnNext >= duration) {
+            lastOnNext = now;
+            observer.onNext(x);
+          }
+        },
+        observer.onError.bind(observer),
+        observer.onCompleted.bind(observer)
+      );
+    });
+  };
+
   var PausableObservable = (function (_super) {
 
     inherits(PausableObservable, _super);