oxjs/source/Ox/js/Async.js

(function() {

    function asyncMap(forEach, col, iterator, that, callback) {
        var type = Ox.typeOf(col),
            results = type == 'object' ? {} : [];
        callback = Ox.last(arguments);
        that = arguments.length == 5 ? that : null;
        forEach(col, function(val, key, obj, callback) {
            iterator(val, key, obj, function(v) {
                results[key] = v;
                callback();
            });
        }, that, function() {
            callback(type == 'string' ? results.join('') : results);
        });
    };

    Ox.asyncMap = function(arr, iterator, that, callback) {
        callback = Ox.last(arguments);
        that = arguments.length == 4 ? that : null;
        if (arr.some(Ox.isArray)) {
            Ox.serialMap(arr, function(val, key, obj, callback) {
                Ox.parallelMap(Ox.makeArray(val), iterator, callback);
            }, callback);
        } else {
            Ox.parallelMap(Ox.makeArray(val), iterator, callback);
        }
    };

    /*@
    Ox.nonblockingForEach <f> Non-blocking forEach
        (col, iterator[, that], callback[, ms]) -> <u> undefined
    @*/
    Ox.nonblockingForEach = function(col, iterator, that, callback, ms) {
        var i, keys, last = Ox.last(arguments), n, time = +new Date();
        callback = Ox.isFunction(last) ? last : arguments[arguments.length - 2];
        col = type == 'array' || type == 'object' ? col : Ox.toArray(col);
        keys = type == 'object' ? Object.keys(col) : Ox.range(col.length);
        ms = ms || 1000;
        n = Ox.len(col);
        that = arguments.length == 5 || (
            arguments.length == 4 && Ox.isFunction(last)
        ) ? that : null;
        iterate();
        function iterate() {
            keys[i] in col && iterator.call(that, col[keys[i]], keys[i], col, function() {            
                if (++i < n) {
                    if (+new Date() < time + ms) {
                        iterate();
                    } else {
                        setTimeout(function() {
                            time += new Date();
                            iterate();
                        });
                    }
                } else {
                    callback();
                }
            });
        }
    };

    Ox.nonblockingMap = function() {
        asyncMap.apply(null, [Ox.nonblockingForEach].concat(Ox.toArray(arguments)));
    };

    /*@
    Ox.parallelForEach <f> forEach with asynchronous iterator, running in parallel
        (col, iterator[, that], callback) -> <u> undefined
        col <a|o|s> Collection
        iterator <f> Iterator function
            val <*> Value
            key <n|s> Key
            obj <a|o> The collection
            callback <f> Callback function
        that <o> The iterator's this binding
        callback <f> Callback function
    @*/
    Ox.parallelForEach = function(col, iterator, that, callback) {
        var i = 0, n, type = Ox.typeOf(col);
        callback = Ox.last(arguments);
        col = type == 'array' || type == 'object' ? col : Ox.toArray(col);
        n = Ox.len(col);
        that = arguments.length == 4 ? that : null;
        Ox.forEach(col, function(val, key, obj) {
            iterator.call(that, val, key, obj, function() {
                ++i == n && callback();
            });
        });
    };

    /*@
    Ox.parallelMap <f> Parallel map with asynchronous iterator
        (col, iterator[, that], callback) -> <u> undefined
        col <a|o|s> Collection
        iterator <f> Iterator function
            val <*> Value
            key <n|s> Key
            obj <a|o|s> The collection
            callback <f> Callback function
        that <o> The iterator's this binding
        callback <f> Callback function
            results <a|o|s> Results
        <script>
            var time = +new Date();
            Ox.parallelMap(
                Ox.range(10),
                function (value, index, array, callback) {
                    setTimeout(function() {
                        callback(+new Date() - time);
                    }, Ox.random(1000));
                },
                function(results) {
                    Ox.print(results);
                }
            );
        </script>
    @*/
    Ox.parallelMap = function() {
        asyncMap.apply(null, [Ox.parallelForEach].concat(Ox.toArray(arguments)));
    };

    /*@
    Ox.serialForEach <f> forEach with asynchronous iterator, run serially
        (col, iterator[, that], callback) -> <u> undefined
        col <a|o|s> Collection
        iterator <f> Iterator function
            val <*> Value
            key <n|s> Key
            obj <a|o> The collection
            callback <f> Callback function
        that <o> The iterator's this binding
        callback <f> Callback function
    @*/
    Ox.serialForEach = function(col, iterator, that, callback) {
        var i = 0, keys, n, type = Ox.typeOf(col);
        callback = Ox.last(arguments);
        col = type == 'array' || type == 'object' ? col : Ox.toArray(col);
        keys = type == 'object' ? Object.keys(col) : Ox.range(col.length);
        n = Ox.len(col);
        that = arguments.length == 4 ? that : null;
        iterate();
        function iterate() {
            keys[i] in col && iterator.call(that, col[keys[i]], keys[i], col, function() {
                ++i < n ? iterate() : callback();
            });
        }
    };

    /*@
    Ox.serialMap <f> Serial map with asynchronous iterator
        (col, iterator[, that], callback) -> <u> undefined
        col <a|o|s> Collection
        iterator <f> Iterator function
            val <*> Value
            key <n|s> Key
            obj <a|o> The collection
            callback <f> Callback function
        that <o> The iterator's this binding
        callback <f> Callback function
            err <o|u> Error object
            results <a|o> Results
        <script>
            var time = +new Date();
            Ox.serialMap(
                Ox.range(10),
                function (value, index, array, callback) {
                    setTimeout(function() {
                        callback(+new Date() - time);
                    }, Ox.random(1000));
                },
                function(results) {
                    Ox.print(results);
                }
            );
        </script>
    @*/
    Ox.serialMap = function(col, iterator, that, callback) {
        asyncMap.apply(null, [Ox.serialForEach].concat(Ox.toArray(arguments)));
    };

}());