Merge pull request #660 from paulpdaniels/controlled-async-fix

Fixed the controlled operator recursion issue by making requests async

Author: mattpodwysocki

src/core/backpressure/controlled.js

@@ -6,9 +6,9 @@
       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();
     }
 
@@ -29,7 +29,7 @@
 
     inherits(ControlledSubject, __super__);
 
-    function ControlledSubject(enableQueue) {
+    function ControlledSubject(enableQueue, scheduler) {
       enableQueue == null && (enableQueue = true);
 
       __super__.call(this, subscribe);
@@ -41,6 +41,7 @@
       this.error = null;
       this.hasFailed = false;
       this.hasCompleted = false;
+      this.scheduler = scheduler || Rx.Scheduler['currentThread'];
     }
 
     addProperties(ControlledSubject.prototype, Observer, {
@@ -83,30 +84,28 @@
           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; //r = this._processRequest(number);
+
+        this.requestedDisposable = this.scheduler.scheduleWithState(number,
+        function(s, i){
+          var r = self._processRequest(i);
+          var remaining = r.numberOfItems;
+          if (!r.returnValue) {
+            self.requestedCount = remaining;
+            self.requestedDisposable = disposableCreate(function(){
+              self.requestedCount = 0;
+            });
+          }
 
-        var number = r.numberOfItems;
-        if (!r.returnValue) {
-          this.requestedCount = number;
-          this.requestedDisposable = disposableCreate(function () {
-            self.requestedCount = 0;
-          });
 
-          return this.requestedDisposable;
-        } else {
-          return disposableEmpty;
-        }
+        });
+
+        return this.requestedDisposable;
+
       },
       disposeCurrentRequest: function () {
         this.requestedDisposable.dispose();
@@ -122,10 +121,17 @@
    * @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);
   };

src/core/headers/backpressureheader.js

@@ -13,4 +13,6 @@
     timeoutScheduler = Rx.Scheduler.timeout,
     currentThreadScheduler = Rx.Scheduler.currentThread,
     identity = Rx.helpers.identity,
+    //TODO Get some consistency about where this is declared
+    isScheduler = Rx.helpers.isScheduler || Rx.Scheduler.isScheduler,
     checkDisposed = Rx.Disposable.checkDisposed;

tests/observable/controlled.js

@@ -217,3 +217,61 @@ test('controlled drops messages with queue disabled', function(){
 
 
 });
+
+test('controlled requests are scheduled', function() {
+    var scheduler = new TestScheduler();
+    var results = scheduler.createObserver();
+
+    var xs = scheduler
+        .createHotObservable(
+        onNext(210, 0),
+        onNext(220, 1),
+        onNext(230, 2),
+        onNext(240, 3),
+        onNext(250, 4),
+        onNext(260, 5),
+        onNext(270, 6),
+        onCompleted(280)
+
+    );
+    var source = xs.controlled();
+
+// process one event at a time
+    scheduler.scheduleAbsolute(200, function() {
+        var subscription =
+            source.subscribe(
+            function (x) {
+                // alternate between immediate and delayed request(1), causes hanging
+                if (x % 2) {
+                    //Immediate
+                    source.request(1); // request next
+                } else {
+                    //Delayed
+                    scheduler.schedule(function () {
+                        source.request(1); // request next
+                    });
+                }
+                results.onNext(x);
+            },
+            results.onError.bind(results),
+            results.onCompleted.bind(results)
+        );
+    });
+
+    scheduler.scheduleAbsolute(300, function() {
+        source.request(1); // start by requesting first item
+    });
+
+    scheduler.start();
+
+    results.messages.assertEqual(
+        onNext(300, 0),
+        onNext(301, 1),
+        onNext(301, 2),
+        onNext(302, 3),
+        onNext(302, 4),
+        onNext(303, 5),
+        onNext(303, 6),
+        onCompleted(303)
+    );
+});