juliangruber

var Readable = require('stream').Readable;
var Writable = require('stream').Writable;

var readable = Readable();
readable._read = function(){
  setTimeout(function(){
    readable.push(''+Date.now());
  }, 100);
}

juliangruber

var http = require('http');
var Readable = require('stream').Readable;

http.createServer(function(req, res){
  
  var stream = Readable();
  
  stream._read = function(){
    console.log('_read', Date.now());
    if (stream.closed) return stream.push(null);

juliangruber

function producer(){
  return function*(){
    return yield io.read();
  }; 
}

function prepend(str, fn){
  return function*(){
    return str + yield fn();

juliangruber

function pipe(){
  var streams = [].slice.call(arguments);
  return streams.slice(1).reduce(function(acc, stream){
    return stream(acc);
  }, streams[0]);
}

// usage:

var read = pipe(a(), b(), c(), d());

juliangruber

function readable(){
  function*(end){
    if (end) return; // clean up etc
    var data = createData();
    return data != null
      ? data
      : end;
  }
}

juliangruber

function values (ary) {
  var i = 0;
  return function (end, cb) {
    if (i < ary.length) cb(true);
    else cb(null, ary[i++])
  }
}

function sync (read) {
  read(null, function next (err, data) {

juliangruber

function values (ary) {
  var i = 0;
  return function*(end) {
    if (end || i < ary.length) return null;
    return ary[i++];
  }
}
 
function sink (read) {
  return function*(end){

juliangruber

{"foo":"bar","num":3,"nes":{"ted":["is",1]}}

juliangruber

var genBody = require('koa-gen-body');
var fromStream = require('co-from-stream');
var sse = require('co-sse');
var fs = require('fs');

app.use(genBody());
app.use(function*(){
  this.body = this.genBody(sse(fromStream(fs.createReadStream('path'))));
});

juliangruber

var genBody = require('koa-gen-body');
var fs = require('co-fs');
var sse = require('co-sse');
 
app.use(genBody());
app.use(function*(){
  this.body = this.genBody(sse(fs.createReadStream('path')));
});