Co/go (run with iojs --harmony_arrow_functions --harmony_rest_parameters)

cogo.js

'use strict';

/*
 * The trampoline. (Lots of inspiration taken from github.com/tj/co.)
 */

function go(star, ...args) {
  return new Promise((resolve, reject) => {
    const gen = star.apply(null, args);
    gen.resolve = resolve;
    gen.reject = reject;
    bounce(gen);
  });
}

function bounce(gen, input) {
  let output;
  try { output = gen.next(input) }
  catch (e) { return gen.reject(e); }
  next(gen, output);
}

function toss(gen, error) {
  let output;
  try { output = gen.throw(error); }
  catch (e) { return gen.reject(e); }
  next(gen, output);
}

function next(gen, output) {
  const v = output.value;
  if ( output.done ) gen.resolve(v);
  else if ( isPromise(v) ) v.then(i => bounce(gen, i), e => toss(gen, e));
  else if ( isPut(v) ) v[0].put(gen, v[1]);
  else if ( isTake(v) ) v.take(gen);
  else gen.reject("Invalid yield in go block.");
}

/*
 * Go block operations.
 */

const put = (ch, val) => [ch, val];
const take = ch => ch;
const alts = ops => new Promise((resolve, reject) => {
  const alt = { alive: true, resolve: resolve };
  ops.forEach(op => {
    if ( isPromise(op) ) {
      op.then(v => { if ( alt.alive ) { alt.alive = false; resolve(v); } }, 
              e => { if ( alt.alive ) { alt.alive = false; reject(e); } });
    } else if ( isPut(op) ) { 
      op[0].put(alt, op[1]);
    } else if ( isTake(op) ) {
      op.take(alt);
    } else {
      throw "Unsupported alt operation."
    }
  });
});

const isPromise = x => typeof x.then === 'function';
const util = require('util');
const isChan = x => typeof x.put === 'function' && typeof x.take === 'function';
const isPut = x => util.isArray(x) && x.length == 2 && isChan(x[0]);
const isTake = isChan; 
const isAlt = x => x.alive != undefined && typeof x.resolve === 'function';
const isGen = x => typeof x.next === 'function' && typeof x.throw === 'function';

/*
 * Simple unbuffered channels.
 */

class Unbuffered {
  constructor() {
    this.putings = [];
    this.takers = [];
  }

  put(puter, val) {
    if ( isAlt(puter) && !puter.alive ) return;

    this.takers = this.takers.filter(t => isGen(t) || t.alive);
    if ( !this.takers.length ) return this.putings.push({ puter: puter, val: val });

    const taker = this.takers.shift();

    if ( isAlt(taker) && isAlt(puter) ) {
      taker.alive = puter.alive = false;
      taker.resolve({ channel: this, value: val });
      puter.resolve({ channel: this });
    } else if ( isAlt(taker) && isGen(puter) ) {
      taker.alive = false;
      taker.resolve({ channel: this, value: val });
      bounce(puter);
    } else if ( isGen(taker) && isAlt(puter) ) {
      puter.alive = false;
      bounce(taker, val);
      puter.resolve({ channel: this });
    } else {
      bounce(taker, val);
      bounce(puter);
    }
  }

  take(taker) {
    if ( isAlt(taker) && !taker.alive ) return;

    this.putings = this.putings.filter(p => isGen(p.puter) || p.puter.alive);
    if ( !this.putings.length ) return this.takers.push(taker);

    const puting = this.putings.shift(),
          puter = puting.puter;

    if ( isAlt(puter) && isAlt(taker) ) {
      puter.alive = taker.alive = false;
      puter.resolve({ channel: this });
      taker.resolve({ channel: this, value: puting.val });
    } else if ( isAlt(puter) && isGen(taker) ) {
      puter.alive = false;
      puter.resolve({ channel: this });
      bounce(taker, puting.val);
    } else if ( isGen(puter) && isAlt(taker) ) {
      taker.alive = false;
      bounce(puter);
      taker.resolve({ channel: this, value: puting.val });
    } else {
      bounce(puter);
      bounce(taker, puting.val);
    }
  }
}

module.exports = {
  go: go,
  put: put,
  take: take,
  alts: alts,
  Unbuffered: Unbuffered
};

examples.js

'use strict';

// Port of one of the js-csp examples.

var csp = require('./cogo.js');

const sleep = ms => new Promise((yep, nope) => setTimeout(yep, ms));

function boring(msg) {
  const ch = new csp.Unbuffered();
  csp.go(function*() {
    for (let i = 0;; i++) {
      yield csp.put(ch, msg + " " + i);
      yield sleep(Math.random() * 1000);
    }
  }).catch(e => console.log("wtf", e));
  return ch;
}

function fanIn(in1, in2) {
  const ch = new csp.Unbuffered();
  csp.go(function*() {
    for (;;) {
      const r = yield csp.alts([in1, in2]);
      yield csp.put(ch, r.value);
    }
  }).catch(e => console.log("wtf", e));
  return ch;
}

csp.go(function*() {
  const ch = fanIn(boring("sup"), boring("yo"));
  for (let i = 0; i < 10; i++) {
    console.log(yield ch);
  }
}).catch(e => console.log("wtf", e));

fakesearch.js

'use strict';

// Port of one of the js-csp examples.

const csp = require("./index.js"),
      go = csp.go,
      put = csp.put, 
      take = csp.take, 
      alts = csp.alts;

const sleep = ms => new Promise((yep, nope) => setTimeout(yep, ms));

function fakeSearch(kind) {
  return function*(query) {
    yield sleep(Math.random() * 200);
    return kind  + " result for query " + query;
  };
}

var web1 = fakeSearch("web1");
var web2 = fakeSearch("web2");
var image1 = fakeSearch("image1");
var image2 = fakeSearch("image2");
var video1 = fakeSearch("video1");
var video2 = fakeSearch("video2");

function* first(query, replicas) {
  const ch = new csp.Unbuffered();
  function* searchReplica(i) {
    yield put(ch, (yield* replicas[i](query)));
  }
  for (var i = 0; i < replicas.length; i++) {
    go(searchReplica, [i]).catch(e => console.log("wtf", e));
  }
  return (yield ch);
}

function* google(query) {
  var ch = new csp.Unbuffered();

  go(function*() {
    yield put(ch, (yield* first(query, [web1, web2])));
  }).catch(e => console.log("wtf", e));
  go(function*() {
    yield put(ch, (yield* first(query, [image1, image2])));
  }).catch(e => console.log("wtf", e));
  go(function*() {
    yield put(ch, (yield* first(query, [video1, video2])));
  }).catch(e => console.log("wtf", e));

  var t = sleep(80);

  var results = [];
  for (var i = 0; i < 3; i++) {
    var r = yield alts([ch, t.then(() => "zzz")]);
    if (r.channel) {
      results.push(r.value);
    } else {
      console.log("timed out");
      break;
    }
  }

  return results;
}

go(function*() {
  var start = new Date();
  var results = yield* google("PLT");
  var elapsed = new Date() - start;
  console.log(results.join("\n"));
  console.log(elapsed);
}).catch(e => console.log("wtf", e));