CSP-style channel implementation for Javascript on top of ES6-style Promises.

channel.js

"use strict";

// Channel is a simple object that can be used for
// CSP-style concurrency in Javascript. In JS code,
// the act of taking a value from a channel looks
// like a blocking call and therefore is not appropriate
// for single-process/single-thread environments like
// NodeJS. To address that, Channel produces promises
// for values.
//
//     var ch = new Channel();
//     var vp = ch.take();
//     vp.then(function (v) {
//         console.log('Got - ' + v);
//     });
//
// Now 'v' is a promise for a value that some other
// code will place into the channel.
//
//     ch.put(42);
//
// Now the promise 'v' will be fulfilled with the value
// '42' and soon enough you'll see the message 'Got - 42'
// on the console. It is worth noting that the return value of 
// the put() call is also a promise that will be fulfilled
// when the value 42 actually gets delivered to the one
// waiting for it.
//
// This facility goes hand in hand with a new method on
// promises - `send` - which calls a method on the value
// being promised when the value becomes available, and
// returns the value that the method call would produce
// as a promise itself. The speciality about `send` is
// that the arguments can be normal values or promises,
// and if there are any promises in the arguments, the
// promise returned by `send` will take into account the
// waiting required for all those promises to be fulfilled.
//
// This approach provides a way to naturally work with
// promises as equivalent to the values they promise,
// passing them into and out of method invocations as
// though they were normal method calls.
//
// This approach will be made simpler via ES6 proxies,
// which will make an explicit `send` method unnecessary
// since we can intercept `get` accesses to forward method
// invocations to promised values. Until then, `send` 
// might be useful.
//
// Ex: If 'a' is a channel to which string values
// are being delivered and they need to be split into
// words and placed into another channel 'b', we can
// write this as -
//
//      a.take().send('split', /\s/g).then(b.put);
//
// It is also possible to setup processes that behave
// like unix pipes, in that they apply some processing
// to values received on an input channel (analogous to
// stdin) and send the result to an output channel (analogous
// to stdout). So if we wanted to make a "splitter", we could
// do it like this using asynchronous recursion -
//
//      function splitter(a, b) {
//          a.take().send('split', /\s/g).then(function (words) {
//              b.put(words); // Don't wait for the words put into 'b' to be received.
//              splitter(a, b);
//          });
//      }
//
// The nice thing is that if the regex according to which to do
// the splitting is to be taken from another channel 'r', it is
// a simple modification to splitter -
//
//     function splitter(a, r, b) {
//         a.take().send('split', r.take()).then(function (words) {
//             b.put(words);
//             splitter(a, r, b);
//         });
//     }
//
// The `send` will take care of ensuring that a regexp is available 
// from the channel `r`.

var Channel = (function () {

    // From bluebird code - global.js
    var theGlobal = (function(){
        //Not in strict mode
        if (typeof this !== "undefined") {
            return this;
        }
        //Strict mode, node
        if (typeof process !== "undefined" &&
            typeof global !== "undefined" &&
            typeof process.execPath === "string") {
                return global;
            }
        //Strict mode, browser
        if (typeof window !== "undefined" &&
            typeof document !== "undefined" &&
            typeof navigator !== "undefined" && navigator !== null &&
            typeof navigator.appName === "string") {
                //Strict mode, Firefox extension
                if(window.wrappedJSObject !== undefined){
                    return window.wrappedJSObject;
                }
                return window;
            }
    }());
    
    var Promise = theGlobal.Promise;

    function Channel() {

        var self = (this === theGlobal ? {} : this); // Don't corrupt the global.

        var queue = [], pending = [];

        // Produces a promise for a value that will be
        // placed into the channel at some point in the 
        // future.
        self.take = function take() {
            var p;
            if (queue.length === 0) {
                // Make a promise for a value.
                p = new Promise();
                pending.push(p);
            } else {
                p = queue.shift();
                p.resolve(p.$value);
            }
            return p;
        };

        // Puts the given value (which can be a promise)
        // into the channel, that will get passed to takers.
        // The result value is itself a promise that will be
        // fulfilled with the given dynValue at the point 
        // a takers gets it.
        self.put = function put(dynValue) {
            var p;
            if (pending.length === 0) {
                p = new Promise();
                p.$value = dynValue;
                queue.push(p);
            } else {
                p = pending.shift();
                p.resolve(dynValue);
            }
            return p;
        };

        // Empty the channel by rejecting all the
        // put and take operations.
        self.drain = function drain(err) {
            while (queue.length > 0) {
                queue.pop().reject(err);
            }
            while (pending.length > 0) {
                pending.pop().reject(err);
            }
            return this;
        };

        return self;
    }

    // Forward a method invocation on the value promised
    // by a Promise instance to the value itself, resolving
    // all argument promises along the way.
    Promise.prototype.send = function (methodName) {
        var argps = Promise.all([].slice.call(arguments, 1));
        var self = this;

        return argps.then(function(args) {
            return self.then(function(value) {
                return value[methodName].apply(value, args);
            });
        });
    };

    return Channel;
}());

if (typeof module !== 'undefined') {
    module.exports = Channel;
}