Pipe is a small function that pipes functionality to resemble RxJS

how-to-v2.0.js

// HOW TO
pipe.pipeline(run => {
  var sensorData = [];
  document.addEventListener("mousemove", _ => {
    sensorData.push(_.x.toFixed(0));
  }, false);
  requestAnimationFrame(function loop(){
    if(sensorData.length > 0) {
      sensorData.forEach(run);
      sensorData = [];
    }
    requestAnimationFrame(loop);
  });
})
.pipe(pipe.continuousBuffer(5))
.pipe(_ => console.log(_))

how-to.js

// Small Request Animation Frame loop
var raf = window.requestAnimationFrame;
function rAFLoop(fn){
  raf(_ => {
    fn();
    rAFLoop(fn);
  });
}

// Example of Sensors
var sensorQueue = [];
document.addEventListener("mousemove", e => sensorQueue.push(Math.abs(e.screenX)));


function add(data){
  data = data.length > 0 ? data : [0];
  return data.reduce((a,b) => { return a+b; });
}
function filterOutPairs(v) {
  return v.filter(_ => v%2 !== 0);
}
function sensorControlSwitch(data) {
  if(data > 600) return pipe(_ => data + " sensor fusion");
  return pipe(_ => data + " raw data");
}


pipe(_ => 5)
.pipe(res => console.log(res));

rAFLoop(_ => {
  if(sensorQueue.length === 0) return;
  
  // How to use
  pipe(_ => sensorQueue.splice(0))
  .pipe(filterOutPairs)
  .pipe(add)
  .pipe(sensorControlSwitch)
  .pipe(res => console.log(res));
  
});

pipe-v2.0.js

// HELPERS
function uniqueID() {
  return Object.freeze({});
}

// PIPE
function pipe(callback=null, _pipes={cbs:[]}){
  callback && _pipes.cbs.push(callback);
  return {
    pipe: cb => pipe(cb, _pipes),
    run: res => {
      var cb, i=0;
      while(cb = _pipes.cbs[i++]){
        temp = cb(res);
        if(temp && temp.pipe) temp.pipe(_ => temp = _).run();
        if(temp === pipe.skip) continue;
        res = temp;
        if(res === pipe.stop) break;
      }
      return res;
    }
  };
}

// COMMANDS
pipe.stop = uniqueID();
pipe.skip = uniqueID();

// EXTRA FUNCTIONALITY
pipe.pipeline = function (callback){
  var _pipe = pipe();
  callback(_pipe.run);
  return _pipe;
};
pipe.preciseBuffer = function(exactly=0){
  var buffer = [];
  return _ => {
    buffer.push(_);
    if(buffer.length < exactly) return pipe.stop;
    return buffer.splice(0);
  };
};
pipe.continuousBuffer = function(upperBoundLimit=100){
  var buffer = [];
  return _ => {
    buffer.push(_);
    if(buffer.length > upperBoundLimit) buffer.splice(0,buffer.length - upperBoundLimit);
    return buffer;
  };
};
pipe.latest = function(n) {
  return _ => {
    return _.slice(-n);
  };
};
pipe.log = function(msg, process = _ => _) {
  return _ => {
    let log = [process(_)]
    msg && log.unshift(msg);
    return console.log.apply(console, log);
  };
};

pipe.js

// Secuential Pipe
// see: https://jsbin.com/turucam/9/edit?js,console,output
function pipe(callback, res){
  var _pipe = pipe;
  if(res && res.pipe) {
    res.pipe(_ => { res = _; })
  }
  res = callback(res);
  return {
    pipe: cb => _pipe(cb, res)
  };
}