Rate limited map-stream

limited-map-stream.js

var es = require("event-stream");

// Wraps a map-stream to only process *limit* chunks at a time
module.exports = function(mapStream, limit) {
        var waiting = 0, ended = false, buffer = [];

        var stream = es.through(write, function() {
                ended = true;
                maybeEnd();
        });

        function write(data) {
                if (waiting >= limit) {
                        buffer.push(data);
                }
                else {
                        waiting += 1;
                        mapStream.write(data);
                }
        }

        function unbuffer() {
                if (buffer.length && waiting < limit) {
                        write(buffer.shift());
                }
        }

        function maybeEnd() {
                if (ended && waiting === 0) {
                        stream.emit("end");
                }
        }

        mapStream.on("data", function(data) {
                stream.emit("data", data);
                waiting -= 1;
                unbuffer();
                maybeEnd();
        });

        return stream;
};