问流与其他流的Bacon.js控制缓冲

EN

Bacon.EventStream.prototype.bufferUntilValue = function(valve) {
var valve_ = valve.startWith(false);

  return this.filter(false).merge(valve_.flatMapConcat((function(_this) {
    return function() {
        return _this.scan([], (function(xs, x) {
            return xs.concat(x);
        }), {
            eager: true
        }).sampledBy(valve).take(1);
    };
  })(this)));
};

// source streams
var sourceObservable = Bacon.interval(1000);
var closingSelector = new Bacon.Bus();

// Constructing a new Observable where we're going to keep our state.
// 
// We need to keep track of two things: 
//   - the buffer that is currently being filled, and
//   -  a previous buffer that is being flushed.
// The state will then look like this:
//   [ buffer, flushed]
// where both buffer and flushed is an array of events from the source observable.

// empty initial state
var initialState = {buffer: [], flushed: []}

// There are two operations on the state: appending a new element to the buffer 
// and flushing the current buffer:

// append each event from the source observable to the buffer,
// keeping flushed unchanged
var appends = sourceObservable.map(function(e) {
   return function(state) {
       state.buffer.push(e); return state; 
   } 
});

// each event from the closingSelector replaces the `flushed` with 
// the `buffer`'s contents, inserting an empty buffer.
var flushes = closingSelector.map(function(_) {
   return function(state) { return {buffer: [], flushed: state.buffer} }
})

// merge appends and flushes into a single stream and apply them to the initial state
var ops = appends.merge(flushes)
var state = ops.scan(initialState, function(acc, f) { return f(acc) });

// resulting stream of flushed events
var flushed = state.sampledBy(closingSelector).map(function(state) { return state.flushed })

// triggered with `closingSelector.push({})`
flushed.onValue(function(x) { console.log("flushed", x) })

var result = sourceStream.holdWhen(closingSelector.flatMap(function(){
  return Bacon.fromArray([false, true]);
}).toProperty(true));