CSP with JavaScript (ES6 only)

example.js

// node --harmony example.js

var go = require('./go');

var even = go.channel();
var odd = go.channel();

go(function* () {
    for (var i = 0; i < 10; i += 2) {
        yield even.put(i);
        yield go.sleep(rand());
    }
});

go(function* () {
    for (var i = 1; i < 10; i += 2) {
        yield odd.put(i);
        yield go.sleep(rand());
    }
});

go(function* () {
    while (true) {
        var ch = yield go.select(even, odd);

        switch (ch) {
            case even:
                var val = yield even.take();
                console.log('even:', val);
                break;
            case odd:
                var val = yield odd.take();
                console.log('odd:', val);
                break;
            default:
                // All channels parked
                break;
        }
    }
});

function rand() {
    return Math.floor(Math.random() * 1000);
}

go.js

module.exports = go;

function go(block) {
    var gen = block();

    function run(gen, step) {
        if (step.done) return;

        var fn = step.value();

        if (!fn) {
            defer(gen, step);
        } else {
            fn(function (err, result) {
                if (err) gen.throw(err);
                defer(gen, gen.next(result));
            });
        }
    }

    function defer(gen, step) {
        setImmediate(function () {
            run(gen, step);
        });
    }

    run(gen, gen.next());
}

go.async = wrap(function (fn) {
    return fn;
});

go.sleep = wrap(function (ms) {
    return function (callback) {
        setTimeout(callback, ms);
    };
});

go.select = wrap(function () {
    var chans = [].slice.call(arguments);

    var candidates = chans.filter(function (chan) {
        return !chan.empty();
    });

    if (candidates.length > 0) {
        var i = rand(candidates.length);
        return cont(null, candidates[i]);
    } else {
        return cont();
    }
});

go.channel = function (size) {
    return new Channel(size);
};

// Channel
// ---------------

function Channel(size) {
    this.size = size || 1;
    this.buffer = [];
}

Channel.prototype.empty = function () {
    return this.buffer.length === 0;
};

Channel.prototype.put = wrap(function (value) {
    if (this.buffer.length >= this.size) return null;

    this.buffer.push(value);
    return cont();
});

Channel.prototype.take = wrap(function () {
    if (this.empty()) return null;

    var value = this.buffer.shift();
    return cont(null, value);
});

// Helpers
// ---------------

function wrap(fn) {
    return function () {
        var args = [].slice.call(arguments);

        return function () {
            return fn.apply(this, args);
        }.bind(this);
    };
}

function cont(err, result) {
    return function (callback) {
        callback(err, result);
    };
}

function rand(n) {
    return Math.floor(Math.random() * n);
}