Backpressure mechanisms in RxJS

buffering.js

var source = Rx.Observable.interval(100).bufferWithTimeOrCount(500 /*ms*/, 50 /* items */);

source.subscribe(function (arr) {
  // Have chunked array based upon time or count, whichever hits first
})

pausable-stream.js

// Create pauser
var pauser = new Rx.Subject();

// Create paused stream state
var source = Rx.Observable.interval(100).pausable(pauser);

// Subscribe
controlledSource.subscribe(function (next) {
  // Do whatever with next, or even pause
  subject.onNext(false);
});

// Start the stream
pauser.onNext(true);

pull-stream.js

// Attach controller to the observable
var source = Rx.Observable.interval(100).attachController(true);

// The controller of the Stream
var controller = new Rx.Subject();

// Get the controlled source bound by the subject
var controlledSource = source.controlledBy(controller);

// Subscribe
controlledSource.subscribe(function (next) {
  // Do whatever with next, such as get 3 more
  subject.onNext(3);
});

// Get three values
subject.onNext(3);