simple-backpressure.js

Rx.Observable.prototype.pauseAndResume = function (subject) {
  var source = this;
  return Observable.create(function (observer) {
    
    var q = [], previous = true;
    
    var subscription =  
      source.combineLatest(
        subject.distinctUntilChanged(), 
        function (data, shouldFire) {
          return { data: data, shouldFire: shouldFire };      
        })
        .subscribe(
          function (results) {
            if (results.shouldFire && previous) {
              observer.onNext(results.data);
            }
            if (results.shouldFire && !previous) {
              while (q.length > 0) {
                observer.onNext(q.shift());
              }
              previous = true;
            } else if (!results.shouldFire && !previous) {
              q.push(results.data);
            } else if (!results.shouldFire && previous) {
              previous = false;
            }

          }, 
          observer.onError.bind(observer),
          observer.onCompleted.bind(observer)
        );

    subject.onNext(false);

    return subscription;
  });
};

usage.js

var Rx = require('rx');

var subject = new Rx.Subject();

var subscription = Rx.Observable
  .interval(100)
  .pauseAndResume(subject)
  .timeInterval()
  .subscribe(function (ts) { console.log('value: ' + ts.value + '@' + ts.interval);  });
  
var shouldFire = true;
subject.onNext(shouldFire);

setInterval(function () {
  subject.onNext(shouldFire = !shouldFire);
}, 5000);

/* Results
value: 0@110
value: 1@158
value: 2@107
value: 3@107
value: 4@546
value: 5@0
value: 6@0
value: 7@0
value: 8@1
value: 9@112
value: 10@106
value: 11@112
value: 12@107
value: 13@592
value: 14@0
value: 15@0
value: 16@1
value: 17@0
value: 18@67
value: 19@106
value: 20@106
*/