Adding pausableBuffered

Author: mattpodwysocki

src/modular/index.js

@@ -9,7 +9,7 @@ Observer.addToObject({
 var Observable = require('./observable');
 
 Observable.addToObject({
-  amb: require('./observable/amb'),
+  race: require('./observable/race'),
   bindCallback: require('./observable/bindcallback'),
   bindNodeCallback: require('./observable/bindnodecallback'),
   case: require('./observable/case'),
@@ -50,7 +50,7 @@ Observable.addToObject({
 });
 
 Observable.addToPrototype({
-  amb: require('./observable/amb'),
+  race: require('./observable/race'),
   and: require('./observable/and'),
   asObservable: require('./observable/asobservable'),
   average: require('./observable/average'),
@@ -104,6 +104,8 @@ Observable.addToPrototype({
   onErrorResumeNext: require('./observable/onerrorresumenext'),
   pairwise: require('./observable/pairwise'),
   partition: require('./observable/partition'),
+  pausable: require('./observable/pausable'),
+  pausableBuffered: require('./observable/pausablebuffered'),
   pluck: require('./observable/pluck'),
   publish: require('./observable/publish'),
   publishLast: require('./observable/publishlast'),
@@ -164,22 +166,32 @@ Subject.addToObject({
 });
 
 var Rx = {
+  // Disposables
   BinaryDisposable: require('./binarydisposable'),
   CompositeDisposable: require('./compositedisposable'),
   Disposable: require('./disposable'),
   NAryDisposable: require('./narydisposable'),
   SerialDisposable: require('./serialdisposable'),
   SingleAssignmentDisposable: require('./singleassignmentdisposable'),
 
+  // Schedulers
   Scheduler: require('./scheduler'),
+  VirtualTimeScheduler: require('./scheduler/virtualtimescheduler'),
+  HistoricalScheduler: require('./scheduler/historicalscheduler'),
 
+  // Core
   Observer: Observer,
   Observable: Observable,
 
+  // Subjects
   AsyncSubject: require('./asyncsubject'),
   BehaviorSubject: require('./behaviorsubject'),
   ReplaySubject: require('./replaysubject'),
-  Subject: Subject
+  Subject: Subject,
+
+  // Testing
+  reactiveTest: require('./testing/reactivetest'),
+  TestScheduler: require('./testing/testscheduler')
 };
 
 module.exports = Rx;

src/modular/observable/pausable.js

@@ -4,6 +4,7 @@ var Observable = require('../observable');
 var distinctUntilChanged = require('./distinctuntilchanged');
 var merge = require('./merge');
 var publish = require('./publish');
+var startWith = require('./startwith');
 var Subject = require('../subject');
 var Disposable = require('../disposable');
 var NAryDisposable = require('../narydisposable');
@@ -29,7 +30,7 @@ PausableObservable.prototype._subscribe = function (o) {
     subscription = conn.subscribe(o),
     connection = Disposable.empty;
 
-  var pausable = distinctUntilChanged(this.pauser).subscribe(function (b) {
+  var pausable = startWith(distinctUntilChanged(this.pauser), !this.paused).subscribe(function (b) {
     if (b) {
       connection = conn.connect();
     } else {

src/modular/observable/pausablebuffered.js

@@ -131,6 +131,7 @@ PausableBufferedObserver.prototype.completed = function () {
 function PausableBufferedObservable(source, pauser) {
   this.source = source;
   this.controller = new Subject();
+  this.paused = true;
 
   if (pauser && pauser.subscribe) {
     this.pauser = merge(this.controller, pauser);
@@ -151,17 +152,19 @@ PausableBufferedObservable.prototype._subscribe = function (o) {
 
   return combineLatestSource(
       this.source,
-      distinctUntilChanged(startWith(this.pauser, false)),
+      distinctUntilChanged(startWith(this.pauser, !this.paused)),
       selectorFn)
     .subscribe(new PausableBufferedObserver(o));
 };
 
 PausableBufferedObservable.prototype.pause = function () {
   this.controller.onNext(false);
+  this.paused = true;
 };
 
 PausableBufferedObservable.prototype.resume = function () {
   this.controller.onNext(true);
+  this.paused = false;
 };
 
 module.exports = function pausableBuffered (source, pauser) {

src/modular/observable/amb.js src/modular/observable/race.jssrc/modular/observable/amb.js renamed to src/modular/observable/race.js

@@ -0,0 +1,47 @@
+'use strict';
+
+var VirtualTimeScheduler = require('./virtualtimescheduler');
+var inherits = require('inherits');
+
+function baseComparer(x, y) { return x > y ? 1 : (x < y ? -1 : 0); }
+
+/**
+ * Provides a virtual time scheduler that uses Date for absolute time and number for relative time.
+ * Creates a new historical scheduler with the specified initial clock value.
+ * @constructor
+ * @param {Number} initialClock Initial value for the clock.
+ * @param {Function} comparer Comparer to determine causality of events based on absolute time.
+ */
+function HistoricalScheduler(initialClock, comparer) {
+  var clock = initialClock == null ? 0 : initialClock;
+  var cmp = comparer || baseComparer;
+  VirtualTimeScheduler.call(this, clock, cmp);
+}
+
+inherits(HistoricalScheduler, VirtualTimeScheduler);
+
+/**
+ * Adds a relative time value to an absolute time value.
+ * @param {Number} absolute Absolute virtual time value.
+ * @param {Number} relative Relative virtual time value to add.
+ * @return {Number} Resulting absolute virtual time sum value.
+ */
+HistoricalScheduler.prototype.add = function (absolute, relative) {
+  return absolute + relative;
+};
+
+HistoricalScheduler.prototype.toAbsoluteTime = function (absolute) {
+  return new Date(absolute).getTime();
+};
+
+/**
+ * Converts the TimeSpan value to a relative virtual time value.
+ * @memberOf HistoricalScheduler
+ * @param {Number} timeSpan TimeSpan value to convert.
+ * @return {Number} Corresponding relative virtual time value.
+ */
+HistoricalScheduler.prototype.toRelativeTime = function (timeSpan) {
+  return timeSpan;
+};
+
+module.exports = HistoricalScheduler;