pausable[Buffered]: remember pause state

...across multiple subscriptions.

Author: rhendric

src/core/backpressure/pausable.js

@@ -3,6 +3,7 @@
     function PausableObservable(source, pauser) {
       this.source = source;
       this.controller = new Subject();
+      this.paused = true;
 
       if (pauser && pauser.subscribe) {
         this.pauser = this.controller.merge(pauser);
@@ -18,7 +19,7 @@
         subscription = conn.subscribe(o),
         connection = disposableEmpty;
 
-      var pausable = this.pauser.distinctUntilChanged().subscribe(function (b) {
+      var pausable = this.pauser.startWith(!this.paused).distinctUntilChanged().subscribe(function (b) {
         if (b) {
           connection = conn.connect();
         } else {
@@ -31,10 +32,12 @@
     };
 
     PausableObservable.prototype.pause = function () {
+      this.paused = true;
       this.controller.onNext(false);
     };
 
     PausableObservable.prototype.resume = function () {
+      this.paused = false;
       this.controller.onNext(true);
     };
 

src/core/backpressure/pausablebuffered.js

@@ -52,6 +52,7 @@
     function PausableBufferedObservable(source, pauser) {
       this.source = source;
       this.controller = new Subject();
+      this.paused = true;
 
       if (pauser && pauser.subscribe) {
         this.pauser = this.controller.merge(pauser);
@@ -70,7 +71,7 @@
       var subscription =
         combineLatestSource(
           this.source,
-          this.pauser.startWith(false).distinctUntilChanged(),
+          this.pauser.startWith(!this.paused).distinctUntilChanged(),
           function (data, shouldFire) {
             return { data: data, shouldFire: shouldFire };
           })
@@ -103,10 +104,12 @@
     };
 
     PausableBufferedObservable.prototype.pause = function () {
+      this.paused = true;
       this.controller.onNext(false);
     };
 
     PausableBufferedObservable.prototype.resume = function () {
+      this.paused = false;
       this.controller.onNext(true);
     };
 

tests/observable/pausable.js

@@ -201,4 +201,98 @@
     );
   });
 
+  test('paused with default controller and multiple subscriptions', function () {
+    var paused, subscription, subscription2;
+
+    var scheduler = new TestScheduler();
+
+    var results = scheduler.createObserver();
+    var results2 = scheduler.createObserver();
+
+    var xs = scheduler.createHotObservable(
+      onNext(150, 1),
+      onNext(210, 2),
+      onNext(230, 3),
+      onNext(301, 4),
+      onNext(350, 5),
+      onNext(399, 6),
+      onCompleted(500)
+    );
+
+    scheduler.scheduleAbsolute(null, 200, function () {
+      paused = xs.pausable();
+      subscription = paused.subscribe(results);
+      paused.resume();
+    });
+
+    scheduler.scheduleAbsolute(null, 240, function () {
+      subscription2 = paused.subscribe(results2);
+    });
+
+    scheduler.scheduleAbsolute(null, 1000, function () {
+      subscription.dispose();
+      subscription2.dispose();
+    });
+
+    scheduler.start();
+
+    results.messages.assertEqual(
+      onNext(210, 2),
+      onNext(230, 3),
+      onNext(301, 4),
+      onNext(350, 5),
+      onNext(399, 6),
+      onCompleted(500)
+    );
+
+    results2.messages.assertEqual(
+      onNext(301, 4),
+      onNext(350, 5),
+      onNext(399, 6),
+      onCompleted(500)
+    );
+  });
+
+  test('pausable is unaffected by currentThread scheduler', function () {
+    var subscription;
+
+    var scheduler = new TestScheduler();
+
+    var controller = new Subject();
+
+    var results = scheduler.createObserver();
+
+    var xs = scheduler.createHotObservable(
+      onNext(150, 1),
+      onNext(210, 2),
+      onNext(230, 3),
+      onNext(301, 4),
+      onNext(350, 5),
+      onNext(399, 6),
+      onCompleted(500)
+    );
+
+    scheduler.scheduleAbsolute(null, 200, function () {
+      Rx.Scheduler.currentThread.schedule(null, function () {
+        subscription = xs.pausable(controller).subscribe(results);
+        controller.onNext(true);
+      });
+    });
+
+    scheduler.scheduleAbsolute(null, 1000, function () {
+      subscription.dispose();
+    });
+
+    scheduler.start();
+
+    results.messages.assertEqual(
+      onNext(210, 2),
+      onNext(230, 3),
+      onNext(301, 4),
+      onNext(350, 5),
+      onNext(399, 6),
+      onCompleted(500)
+    );
+  });
+  
 }());

tests/observable/pausablebuffered.js

@@ -461,4 +461,98 @@
     QUnit.deepEqual(results, [1]);
   });
 
+  test('paused with default controller and multiple subscriptions', function () {
+    var paused, subscription, subscription2;
+
+    var scheduler = new TestScheduler();
+
+    var results = scheduler.createObserver();
+    var results2 = scheduler.createObserver();
+
+    var xs = scheduler.createHotObservable(
+      onNext(150, 1),
+      onNext(210, 2),
+      onNext(230, 3),
+      onNext(301, 4),
+      onNext(350, 5),
+      onNext(399, 6),
+      onCompleted(500)
+    );
+
+    scheduler.scheduleAbsolute(null, 200, function () {
+      paused = xs.pausableBuffered();
+      subscription = paused.subscribe(results);
+      paused.resume();
+    });
+
+    scheduler.scheduleAbsolute(null, 240, function () {
+      subscription2 = paused.subscribe(results2);
+    });
+
+    scheduler.scheduleAbsolute(null, 1000, function () {
+      subscription.dispose();
+      subscription2.dispose();
+    });
+
+    scheduler.start();
+
+    results.messages.assertEqual(
+      onNext(210, 2),
+      onNext(230, 3),
+      onNext(301, 4),
+      onNext(350, 5),
+      onNext(399, 6),
+      onCompleted(500)
+    );
+
+    results2.messages.assertEqual(
+      onNext(301, 4),
+      onNext(350, 5),
+      onNext(399, 6),
+      onCompleted(500)
+    );
+  });
+
+  test('pausableBuffered is unaffected by currentThread scheduler', function () {
+    var subscription;
+
+    var scheduler = new TestScheduler();
+
+    var controller = new Subject();
+
+    var results = scheduler.createObserver();
+
+    var xs = scheduler.createHotObservable(
+      onNext(150, 1),
+      onNext(210, 2),
+      onNext(230, 3),
+      onNext(301, 4),
+      onNext(350, 5),
+      onNext(399, 6),
+      onCompleted(500)
+    );
+
+    scheduler.scheduleAbsolute(null, 200, function () {
+      Rx.Scheduler.currentThread.schedule(null, function () {
+        subscription = xs.pausableBuffered(controller).subscribe(results);
+        controller.onNext(true);
+      });
+    });
+
+    scheduler.scheduleAbsolute(null, 1000, function () {
+      subscription.dispose();
+    });
+
+    scheduler.start();
+
+    results.messages.assertEqual(
+      onNext(210, 2),
+      onNext(230, 3),
+      onNext(301, 4),
+      onNext(350, 5),
+      onNext(399, 6),
+      onCompleted(500)
+    );
+  });
+
 }());