Merge pull request #608 from paulpdaniels/controlled-completed-fix

Controlled completed fix

Author: mattpodwysocki

src/core/backpressure/controlled.js

@@ -41,24 +41,29 @@
       this.error = null;
       this.hasFailed = false;
       this.hasCompleted = false;
-      this.controlledDisposable = disposableEmpty;
     }
 
     addProperties(ControlledSubject.prototype, Observer, {
       onCompleted: function () {
         this.hasCompleted = true;
-        (!this.enableQueue || this.queue.length === 0) && this.subject.onCompleted();
+        if (!this.enableQueue || this.queue.length === 0)
+          this.subject.onCompleted();
+        else
+          this.queue.push(Rx.Notification.createOnCompleted());
       },
       onError: function (error) {
         this.hasFailed = true;
         this.error = error;
-        (!this.enableQueue || this.queue.length === 0) && this.subject.onError(error);
+        if (!this.enableQueue || this.queue.length === 0)
+          this.subject.onError(error);
+        else
+          this.queue.push(Rx.Notification.createOnError(error));
       },
       onNext: function (value) {
         var hasRequested = false;
 
         if (this.requestedCount === 0) {
-          this.enableQueue && this.queue.push(value);
+          this.enableQueue && this.queue.push(Rx.Notification.createOnNext(value));
         } else {
           (this.requestedCount !== -1 && this.requestedCount-- === 0) && this.disposeCurrentRequest();
           hasRequested = true;
@@ -67,25 +72,23 @@
       },
       _processRequest: function (numberOfItems) {
         if (this.enableQueue) {
-          while (this.queue.length >= numberOfItems && numberOfItems > 0) {
-            this.subject.onNext(this.queue.shift());
-            numberOfItems--;
+          while ((this.queue.length >= numberOfItems && numberOfItems > 0) ||
+          (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 = []; }
           }
 
-          return this.queue.length !== 0 ?
-            { numberOfItems: numberOfItems, returnValue: true } :
-            { numberOfItems: numberOfItems, returnValue: false };
+          return { numberOfItems : numberOfItems, returnValue: this.queue.length !== 0};
         }
 
-        if (this.hasFailed) {
-          this.subject.onError(this.error);
-          this.controlledDisposable.dispose();
-          this.controlledDisposable = disposableEmpty;
-        } else if (this.hasCompleted) {
-          this.subject.onCompleted();
-          this.controlledDisposable.dispose();
-          this.controlledDisposable = disposableEmpty;
-        }
+        //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 };
       },
@@ -100,7 +103,7 @@
             self.requestedCount = 0;
           });
 
-          return this.requestedDisposable
+          return this.requestedDisposable;
         } else {
           return disposableEmpty;
         }

tests/observable/controlled.js

@@ -95,3 +95,125 @@ test('controlled gets two sets of values', function () {
     onCompleted(500)
   );
 });
+
+test('controlled fires on completed', function(){
+
+    var scheduler = new TestScheduler();
+    var results = scheduler.createObserver();
+
+    var source = Rx.Observable.range(1, 2).controlled();
+
+    scheduler.scheduleAbsolute(200, function(){
+        source.subscribe(results);
+    });
+
+    scheduler.scheduleAbsolute(300, function () {
+        source.request(3);
+    });
+
+    scheduler.start();
+
+    results.messages.assertEqual(
+        onNext(300, 1),
+        onNext(300, 2),
+        onCompleted(300)
+    );
+});
+
+test('controlled cancels inflight request', function(){
+
+    var scheduler = new TestScheduler();
+
+    var source = scheduler.createHotObservable(
+        onNext(400, 2),
+        onNext(410, 3),
+        onNext(420, 4)
+    ).controlled();
+
+    var results = scheduler.createObserver();
+
+    scheduler.scheduleAbsolute(200, function(){
+        source.request(3);
+    });
+
+    scheduler.scheduleAbsolute(200, function(){
+        source.request(2);
+    });
+
+    scheduler.scheduleAbsolute(300, function(){
+        source.subscribe(results);
+    });
+
+    scheduler.advanceBy(420);
+
+
+    results.messages.assertEqual(
+        onNext(400, 2),
+        onNext(410, 3)
+    );
+
+
+});
+
+test('controlled fires onError', function(){
+
+    var scheduler = new TestScheduler();
+    var results = scheduler.createObserver();
+
+    var error = new Error("expected");
+    var source = Rx.Observable.range(1, 2, scheduler)
+        .concat(Rx.Observable.throw(error, scheduler))
+        .controlled();
+
+    scheduler.scheduleAbsolute(200, function(){
+        source.subscribe(results);
+    });
+
+
+    scheduler.scheduleAbsolute(300, function () {
+        source.request(3);
+    });
+
+    scheduler.start();
+
+    results.messages.assertEqual(
+        onNext(300, 1),
+        onNext(300, 2),
+        onError(300, error)
+    );
+
+});
+
+test('controlled drops messages with queue disabled', function(){
+
+    var scheduler = new TestScheduler();
+
+    var source = scheduler.createHotObservable(
+        onNext(400, 1),
+        onNext(410, 2),
+        onNext(420, 3),
+        onNext(430, 4),
+        onCompleted(500)
+    ).controlled(false);
+
+    var results = scheduler.createObserver();
+
+
+    scheduler.scheduleAbsolute(415, function(){
+        source.request(2);
+    });
+
+    scheduler.scheduleAbsolute(200, function(){
+        source.subscribe(results);
+    });
+
+    scheduler.start();
+
+    results.messages.assertEqual(
+      onNext(420, 3),
+      onNext(430, 4),
+      onCompleted(500)
+    );
+
+
+});