backpressure

Author: mattpodwysocki

src/core/backpressure/stopandwait.js

@@ -6,7 +6,7 @@
     }
 
     function scheduleMethod(s, self) {
-      self.source.request(1);
+      return self.source.request(1);
     }
 
     StopAndWaitObservable.prototype._subscribe = function (o) {
@@ -38,15 +38,15 @@
       };
 
       function innerScheduleMethod(s, self) {
-        self.observable.source.request(1);
+        return self.observable.source.request(1);
       }
 
       StopAndWaitObserver.prototype.next = function (value) {
         this.observer.onNext(value);
         this.scheduleDisposable = defaultScheduler.schedule(this, innerScheduleMethod);
       };
 
-      StopAndWaitObservable.dispose = function () {
+      StopAndWaitObserver.dispose = function () {
         this.observer = null;
         if (this.cancel) {
           this.cancel.dispose();

src/core/backpressure/windowed.js

@@ -7,7 +7,7 @@
     }
 
     function scheduleMethod(s, self) {
-      self.source.request(self.windowSize);
+      return self.source.request(self.windowSize);
     }
 
     WindowedObservable.prototype._subscribe = function (o) {
@@ -40,7 +40,7 @@
       };
 
       function innerScheduleMethod(s, self) {
-        self.observable.source.request(self.observable.windowSize);
+        return self.observable.source.request(self.observable.windowSize);
       }
 
       WindowedObserver.prototype.next = function (value) {

src/modular/index.js

@@ -45,6 +45,7 @@ Observable.addToObject({
   timer: require('./observable/timer'),
   toAsync: require('./observable/toasync'),
   when: require('./observable/when'),
+  using: require('./observable/using'),
   zip: require('./observable/zip')
 });
 

src/modular/observable/controlled.js

@@ -0,0 +1,152 @@
+'use strict';
+
+var Observable = require('../observable');
+var StopAndWaitObservable = require('./stopandwait');
+var WindowedObservable = require('./windowed');
+var multicast = require('./multicast');
+var Notification = require('../notification');
+var Observer = require('../observer');
+var Subject = require('../subject');
+var Disposable = require('../disposable');
+var Scheduler = require('../scheduler');
+var addProperties = require('../internal/addproperties');
+var inherits = require('inherits');
+
+global._Rx || (global._Rx = {});
+if (!global._Rx.currentThreadScheduler) {
+  require('../scheduler/currentthreadscheduler');
+}
+
+if (!global._Rx.defaultScheduler) {
+  require('../scheduler/defaultscheduler');
+}
+
+function ControlledSubject(enableQueue, scheduler) {
+  enableQueue == null && (enableQueue = true);
+  this.subject = new Subject();
+  this.enableQueue = enableQueue;
+  this.queue = enableQueue ? [] : null;
+  this.requestedCount = 0;
+  this.requestedDisposable = null;
+  this.error = null;
+  this.hasFailed = false;
+  this.hasCompleted = false;
+  this.scheduler = scheduler || global._Rx.currentThreadScheduler;
+  Observable.call(this);
+}
+
+inherits(ControlledSubject, Observable);
+
+addProperties(ControlledSubject.prototype, Observer, {
+  _subscribe: function (o) {
+    return this.subject.subscribe(o);
+  },
+  onCompleted: function () {
+    this.hasCompleted = true;
+    if (!this.enableQueue || this.queue.length === 0) {
+      this.subject.onCompleted();
+      this.disposeCurrentRequest();
+    } else {
+      this.queue.push(Notification.createOnCompleted());
+    }
+  },
+  onError: function (error) {
+    this.hasFailed = true;
+    this.error = error;
+    if (!this.enableQueue || this.queue.length === 0) {
+      this.subject.onError(error);
+      this.disposeCurrentRequest();
+    } else {
+      this.queue.push(Notification.createOnError(error));
+    }
+  },
+  onNext: function (value) {
+    if (this.requestedCount <= 0) {
+      this.enableQueue && this.queue.push(Notification.createOnNext(value));
+    } else {
+      (this.requestedCount-- === 0) && this.disposeCurrentRequest();
+      this.subject.onNext(value);
+    }
+  },
+  _processRequest: function (numberOfItems) {
+    if (this.enableQueue) {
+      while (this.queue.length > 0 && (numberOfItems > 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 numberOfItems;
+  },
+  request: function (number) {
+    this.disposeCurrentRequest();
+    var self = this;
+
+    this.requestedDisposable = this.scheduler.schedule(number,
+    function(s, i) {
+      var remaining = self._processRequest(i);
+      var stopped = self.hasCompleted || self.hasFailed;
+      if (!stopped && remaining > 0) {
+        self.requestedCount = remaining;
+
+        return Disposable.create(function () {
+          self.requestedCount = 0;
+        });
+          // Scheduled item is still in progress. Return a new
+          // disposable to allow the request to be interrupted
+          // via dispose.
+      }
+    });
+
+    return this.requestedDisposable;
+  },
+  disposeCurrentRequest: function () {
+    if (this.requestedDisposable) {
+      this.requestedDisposable.dispose();
+      this.requestedDisposable = null;
+    }
+  }
+});
+
+function ControlledObservable (source, enableQueue, scheduler) {
+  this.subject = new ControlledSubject(enableQueue, scheduler);
+  this.source = multicast(source, this.subject).refCount();
+  Observable.call(this);
+}
+
+inherits(ControlledObservable, Observable);
+
+ControlledObservable.prototype._subscribe = function (o) {
+  return this.source.subscribe(o);
+};
+
+ControlledObservable.prototype.request = function (numberOfItems) {
+  return this.subject.request(numberOfItems == null ? -1 : numberOfItems);
+};
+
+ControlledObservable.prototype.stopAndWait = function (scheduler) {
+  Scheduler.isScheduler(scheduler) || (scheduler = global._Rx.defaultScheduler);
+  return new StopAndWaitObservable(this, scheduler);
+};
+
+ControlledObservable.prototype.windowed = function (windowSize, scheduler) {
+  Scheduler.isScheduler(scheduler) || (scheduler = global._Rx.defaultScheduler);
+  return new WindowedObservable(this, windowSize, scheduler);
+};
+
+module.exports = function controlled (source, enableQueue, scheduler) {
+
+  if (enableQueue && Scheduler.isScheduler(enableQueue)) {
+    scheduler = enableQueue;
+    enableQueue = true;
+  }
+
+  if (enableQueue == null) {  enableQueue = true; }
+  return new ControlledObservable(source, enableQueue, scheduler);
+};

src/modular/observable/pausable.js

@@ -0,0 +1,54 @@
+'use strict';
+
+var Observable = require('../observable');
+var distinctUntilChanged = require('./distinctuntilchanged');
+var merge = require('./merge');
+var publish = require('./publish');
+var Subject = require('../subject');
+var Disposable = require('../disposable');
+var NAryDisposable = require('../narydisposable');
+var inherits = require('inherits');
+
+function PausableObservable(source, pauser) {
+  this.source = source;
+  this.controller = new Subject();
+
+  if (pauser && pauser.subscribe) {
+    this.pauser = merge(this.controller, pauser);
+  } else {
+    this.pauser = this.controller;
+  }
+
+  Observable.call(this);
+}
+
+inherits(PausableObservable, Observable);
+
+PausableObservable.prototype._subscribe = function (o) {
+  var conn = publish(this.source),
+    subscription = conn.subscribe(o),
+    connection = Disposable.empty;
+
+  var pausable = distinctUntilChanged(this.pauser).subscribe(function (b) {
+    if (b) {
+      connection = conn.connect();
+    } else {
+      connection.dispose();
+      connection = Disposable.empty;
+    }
+  });
+
+  return new NAryDisposable([subscription, connection, pausable]);
+};
+
+PausableObservable.prototype.pause = function () {
+  this.controller.onNext(false);
+};
+
+PausableObservable.prototype.resume = function () {
+  this.controller.onNext(true);
+};
+
+module.exports = function pausable (source, pauser) {
+  return new PausableObservable(source, pauser);
+};