Version bump to 2.5.2

bower.json

@@ -1,6 +1,6 @@
 {
   "name": "rxjs",
-  "version": "2.5.1",
+  "version": "2.5.2",
   "main": [
     "dist/rx.all.js",
     "dist/rx.all.map",

dist/rx.all.compat.js

@@ -6694,15 +6694,14 @@
       return this.source.subscribe(observer);
     }
 
-    function ControlledObservable (source, enableQueue) {
+    function ControlledObservable (source, enableQueue, scheduler) {
       __super__.call(this, subscribe, source);
-      this.subject = new ControlledSubject(enableQueue);
+      this.subject = new ControlledSubject(enableQueue, scheduler);
       this.source = source.multicast(this.subject).refCount();
     }
 
     ControlledObservable.prototype.request = function (numberOfItems) {
-      if (numberOfItems == null) { numberOfItems = -1; }
-      return this.subject.request(numberOfItems);
+      return this.subject.request(numberOfItems == null ? -1 : numberOfItems);
     };
 
     return ControlledObservable;
@@ -6717,7 +6716,7 @@
 
     inherits(ControlledSubject, __super__);
 
-    function ControlledSubject(enableQueue) {
+    function ControlledSubject(enableQueue, scheduler) {
       enableQueue == null && (enableQueue = true);
 
       __super__.call(this, subscribe);
@@ -6729,29 +6728,32 @@
       this.error = null;
       this.hasFailed = false;
       this.hasCompleted = false;
+      this.scheduler = scheduler || currentThreadScheduler;
     }
 
     addProperties(ControlledSubject.prototype, Observer, {
       onCompleted: function () {
         this.hasCompleted = true;
-        if (!this.enableQueue || this.queue.length === 0)
+        if (!this.enableQueue || this.queue.length === 0) {
           this.subject.onCompleted();
-        else
-          this.queue.push(Rx.Notification.createOnCompleted());
+        } else {
+          this.queue.push(Notification.createOnCompleted());
+        }
       },
       onError: function (error) {
         this.hasFailed = true;
         this.error = error;
-        if (!this.enableQueue || this.queue.length === 0)
+        if (!this.enableQueue || this.queue.length === 0) {
           this.subject.onError(error);
-        else
-          this.queue.push(Rx.Notification.createOnError(error));
+        } else {
+          this.queue.push(Notification.createOnError(error));
+        }
       },
       onNext: function (value) {
         var hasRequested = false;
 
         if (this.requestedCount === 0) {
-          this.enableQueue && this.queue.push(Rx.Notification.createOnNext(value));
+          this.enableQueue && this.queue.push(Notification.createOnNext(value));
         } else {
           (this.requestedCount !== -1 && this.requestedCount-- === 0) && this.disposeCurrentRequest();
           hasRequested = true;
@@ -6764,37 +6766,35 @@
           (this.queue.length > 0 && this.queue[0].kind !== 'N')) {
             var first = this.queue.shift();
             first.accept(this.subject);
-            if (first.kind === 'N') numberOfItems--;
-            else { this.disposeCurrentRequest(); this.queue = []; }
+            if (first.kind === 'N') {
+              numberOfItems--;
+            } else {
+              this.disposeCurrentRequest();
+              this.queue = [];
+            }
           }
 
           return { numberOfItems : numberOfItems, returnValue: this.queue.length !== 0};
         }
 
-        //TODO I don't think this is ever necessary, since termination of a sequence without a queue occurs in the onCompletion or onError function
-        //if (this.hasFailed) {
-        //  this.subject.onError(this.error);
-        //} else if (this.hasCompleted) {
-        //  this.subject.onCompleted();
-        //}
-
         return { numberOfItems: numberOfItems, returnValue: false };
       },
       request: function (number) {
         this.disposeCurrentRequest();
-        var self = this, r = this._processRequest(number);
+        var self = this;
 
-        var number = r.numberOfItems;
-        if (!r.returnValue) {
-          this.requestedCount = number;
-          this.requestedDisposable = disposableCreate(function () {
-            self.requestedCount = 0;
-          });
+        this.requestedDisposable = this.scheduler.scheduleWithState(number,
+        function(s, i) {
+          var r = self._processRequest(i), remaining = r.numberOfItems;
+          if (!r.returnValue) {
+            self.requestedCount = remaining;
+            self.requestedDisposable = disposableCreate(function () {
+              self.requestedCount = 0;
+            });
+          }
+        });
 
-          return this.requestedDisposable;
-        } else {
-          return disposableEmpty;
-        }
+        return this.requestedDisposable;
       },
       disposeCurrentRequest: function () {
         this.requestedDisposable.dispose();
@@ -6810,12 +6810,19 @@
    * @example
    * var source = Rx.Observable.interval(100).controlled();
    * source.request(3); // Reads 3 values
-   * @param {Observable} pauser The observable sequence used to pause the underlying sequence.
+   * @param {bool} enableQueue truthy value to determine if values should be queued pending the next request
+   * @param {Scheduler} scheduler determines how the requests will be scheduled
    * @returns {Observable} The observable sequence which is paused based upon the pauser.
    */
-  observableProto.controlled = function (enableQueue) {
+  observableProto.controlled = function (enableQueue, scheduler) {
+
+    if (enableQueue && isScheduler(enableQueue)) {
+        scheduler = enableQueue;
+        enableQueue = true;
+    }
+
     if (enableQueue == null) {  enableQueue = true; }
-    return new ControlledObservable(this, enableQueue);
+    return new ControlledObservable(this, enableQueue, scheduler);
   };
 
   var StopAndWaitObservable = (function (__super__) {

dist/rx.all.js

@@ -6481,15 +6481,14 @@
       return this.source.subscribe(observer);
     }
 
-    function ControlledObservable (source, enableQueue) {
+    function ControlledObservable (source, enableQueue, scheduler) {
       __super__.call(this, subscribe, source);
-      this.subject = new ControlledSubject(enableQueue);
+      this.subject = new ControlledSubject(enableQueue, scheduler);
       this.source = source.multicast(this.subject).refCount();
     }
 
     ControlledObservable.prototype.request = function (numberOfItems) {
-      if (numberOfItems == null) { numberOfItems = -1; }
-      return this.subject.request(numberOfItems);
+      return this.subject.request(numberOfItems == null ? -1 : numberOfItems);
     };
 
     return ControlledObservable;
@@ -6504,7 +6503,7 @@
 
     inherits(ControlledSubject, __super__);
 
-    function ControlledSubject(enableQueue) {
+    function ControlledSubject(enableQueue, scheduler) {
       enableQueue == null && (enableQueue = true);
 
       __super__.call(this, subscribe);
@@ -6516,29 +6515,32 @@
       this.error = null;
       this.hasFailed = false;
       this.hasCompleted = false;
+      this.scheduler = scheduler || currentThreadScheduler;
     }
 
     addProperties(ControlledSubject.prototype, Observer, {
       onCompleted: function () {
         this.hasCompleted = true;
-        if (!this.enableQueue || this.queue.length === 0)
+        if (!this.enableQueue || this.queue.length === 0) {
           this.subject.onCompleted();
-        else
-          this.queue.push(Rx.Notification.createOnCompleted());
+        } else {
+          this.queue.push(Notification.createOnCompleted());
+        }
       },
       onError: function (error) {
         this.hasFailed = true;
         this.error = error;
-        if (!this.enableQueue || this.queue.length === 0)
+        if (!this.enableQueue || this.queue.length === 0) {
           this.subject.onError(error);
-        else
-          this.queue.push(Rx.Notification.createOnError(error));
+        } else {
+          this.queue.push(Notification.createOnError(error));
+        }
       },
       onNext: function (value) {
         var hasRequested = false;
 
         if (this.requestedCount === 0) {
-          this.enableQueue && this.queue.push(Rx.Notification.createOnNext(value));
+          this.enableQueue && this.queue.push(Notification.createOnNext(value));
         } else {
           (this.requestedCount !== -1 && this.requestedCount-- === 0) && this.disposeCurrentRequest();
           hasRequested = true;
@@ -6551,37 +6553,35 @@
           (this.queue.length > 0 && this.queue[0].kind !== 'N')) {
             var first = this.queue.shift();
             first.accept(this.subject);
-            if (first.kind === 'N') numberOfItems--;
-            else { this.disposeCurrentRequest(); this.queue = []; }
+            if (first.kind === 'N') {
+              numberOfItems--;
+            } else {
+              this.disposeCurrentRequest();
+              this.queue = [];
+            }
           }
 
           return { numberOfItems : numberOfItems, returnValue: this.queue.length !== 0};
         }
 
-        //TODO I don't think this is ever necessary, since termination of a sequence without a queue occurs in the onCompletion or onError function
-        //if (this.hasFailed) {
-        //  this.subject.onError(this.error);
-        //} else if (this.hasCompleted) {
-        //  this.subject.onCompleted();
-        //}
-
         return { numberOfItems: numberOfItems, returnValue: false };
       },
       request: function (number) {
         this.disposeCurrentRequest();
-        var self = this, r = this._processRequest(number);
+        var self = this;
 
-        var number = r.numberOfItems;
-        if (!r.returnValue) {
-          this.requestedCount = number;
-          this.requestedDisposable = disposableCreate(function () {
-            self.requestedCount = 0;
-          });
+        this.requestedDisposable = this.scheduler.scheduleWithState(number,
+        function(s, i) {
+          var r = self._processRequest(i), remaining = r.numberOfItems;
+          if (!r.returnValue) {
+            self.requestedCount = remaining;
+            self.requestedDisposable = disposableCreate(function () {
+              self.requestedCount = 0;
+            });
+          }
+        });
 
-          return this.requestedDisposable;
-        } else {
-          return disposableEmpty;
-        }
+        return this.requestedDisposable;
       },
       disposeCurrentRequest: function () {
         this.requestedDisposable.dispose();
@@ -6597,12 +6597,19 @@
    * @example
    * var source = Rx.Observable.interval(100).controlled();
    * source.request(3); // Reads 3 values
-   * @param {Observable} pauser The observable sequence used to pause the underlying sequence.
+   * @param {bool} enableQueue truthy value to determine if values should be queued pending the next request
+   * @param {Scheduler} scheduler determines how the requests will be scheduled
    * @returns {Observable} The observable sequence which is paused based upon the pauser.
    */
-  observableProto.controlled = function (enableQueue) {
+  observableProto.controlled = function (enableQueue, scheduler) {
+
+    if (enableQueue && isScheduler(enableQueue)) {
+        scheduler = enableQueue;
+        enableQueue = true;
+    }
+
     if (enableQueue == null) {  enableQueue = true; }
-    return new ControlledObservable(this, enableQueue);
+    return new ControlledObservable(this, enableQueue, scheduler);
   };
 
   var StopAndWaitObservable = (function (__super__) {