refactor(throttle): remove tryCatch/errorObject

perf/micro/current-thread-scheduler/operators/throttle.js

@@ -0,0 +1,22 @@
+var RxOld = require('rx');
+var RxNew = require('../../../../index');
+
+module.exports = function (suite) {
+  // v4's `throttle()` does not accept a duration selector.
+  // So this test is just a sample for each revisions of RxNew.
+  var oldThrottleWithImmediateScheduler = RxOld.Observable.range(0, 25, RxOld.Scheduler.currentThread)
+    .throttle(1);
+  var newThrottleWithImmediateScheduler = RxNew.Observable.range(0, 25, RxNew.Scheduler.queue)
+    .throttle(function () { return RxNew.Observable.of(0); });
+
+  function _next(x) { }
+  function _error(e) { }
+  function _complete() { }
+  return suite
+    .add('old throttle() with immediate scheduler', function () {
+      oldThrottleWithImmediateScheduler.subscribe(_next, _error, _complete);
+    })
+    .add('new throttle() with immediate scheduler', function () {
+      newThrottleWithImmediateScheduler.subscribe(_next, _error, _complete);
+    });
+};

perf/micro/immediate-scheduler/operators/throttle.js

@@ -0,0 +1,22 @@
+var RxOld = require('rx');
+var RxNew = require('../../../../index');
+
+module.exports = function (suite) {
+  // v4's `throttle()` does not accept a duration selector.
+  // So this test is just a sample for each revisions of RxNew.
+  var oldThrottleWithImmediateScheduler = RxOld.Observable.range(0, 25, RxOld.Scheduler.immediate)
+    .throttle(1);
+  var newThrottleWithImmediateScheduler = RxNew.Observable.range(0, 25)
+    .throttle(function () { return RxNew.Observable.of(0); });
+
+  function _next(x) { }
+  function _error(e) { }
+  function _complete() { }
+  return suite
+    .add('old throttle() with immediate scheduler', function () {
+      oldThrottleWithImmediateScheduler.subscribe(_next, _error, _complete);
+    })
+    .add('new throttle() with immediate scheduler', function () {
+      newThrottleWithImmediateScheduler.subscribe(_next, _error, _complete);
+    });
+};

src/operator/throttle.ts

@@ -3,8 +3,6 @@ import {Observable} from '../Observable';
 import {Subscriber} from '../Subscriber';
 import {Subscription} from '../Subscription';
 
-import {tryCatch} from '../util/tryCatch';
-import {errorObject} from '../util/errorObject';
 import {OuterSubscriber} from '../OuterSubscriber';
 import {InnerSubscriber} from '../InnerSubscriber';
 import {subscribeToResult} from '../util/subscribeToResult';
@@ -25,24 +23,34 @@ class ThrottleOperator<T> implements Operator<T, T> {
 class ThrottleSubscriber<T, R> extends OuterSubscriber<T, R> {
   private throttled: Subscription;
 
-  constructor(destination: Subscriber<any>,
+  constructor(protected destination: Subscriber<T>,
               private durationSelector: (value: T) => Observable<number> | Promise<number>) {
     super(destination);
   }
 
   protected _next(value: T): void {
     if (!this.throttled) {
-      const duration = tryCatch(this.durationSelector)(value);
-      if (duration === errorObject) {
-        this.destination.error(errorObject.e);
-      } else {
-        this.add(this.throttled = subscribeToResult(this, duration));
-        this.destination.next(value);
-      }
+      this.tryDurationSelector(value);
     }
   }
 
-  _unsubscribe() {
+  private tryDurationSelector(value: T): void {
+    let duration: Observable<number> | Promise<number> = null;
+    try {
+      duration = this.durationSelector(value);
+    } catch (err) {
+      this.destination.error(err);
+      return;
+    }
+    this.emitAndThrottle(value, duration);
+  }
+
+  private emitAndThrottle(value: T, duration: Observable<number> | Promise<number>) {
+    this.add(this.throttled = subscribeToResult(this, duration));
+    this.destination.next(value);
+  }
+
+  protected _unsubscribe() {
     const throttled = this.throttled;
     if (throttled) {
       this.remove(throttled);