pipeline処理

schedpipeline.js

var SchedPipeline = exports.SchedPipeline = function(){
    this.q = [];
    this.runflag = false;
}
SchedPipeline.prototype = {
    run : function(){
        this.runflag = true;
        var self = this;
        process.nextTick(function T(){
            if(self.runflag){
                var f = self.q.shift();
                if(f){
                    f();
                }
                process.nextTick(T);
            }
        });
    },
    stop : function(){
        this.runflag = false;
    },
    add : function(funcs){
        var i = 0;
        var len = funcs.length;
        var q = this.q;
        var err = null;
        return function f(cb, arg){
            if(i >= len){
                if(cb){
                    cb(err, arg);
                }
            }else{
                q.push(function(){
                    if(i < len){
                        try{
                            funcs[i++](function(argp){
                                f(cb, argp);
                            }, arg);
                        }catch(e){
                            cb(e, arg);
                        }
                    }
                });
            }
        };
    },
}
if(!module.parent){
    var p = new SchedPipeline();
    p.run();
    var f1 = p.add([
        function(n,arg){
            console.log(":1");
            n(arg);
        },
        function(n,arg){
            console.log(":2");
            n(arg);
        },
        function(n,arg){
            console.log(":3");
throw new Error("aaaa");
            n(arg);
        }
    ]);
    f1(function(err,arg){
        console.log("end");
        console.log(err);
        console.log(arg);
        p.stop();
    },"aaaa");
}

test.js

var sched = require("./schedpipeline");
var p = new sched.SchedPipeline();
p.run();

var rand = function(){
    return (Math.random() * 1000)|0;
}

var r=0;
[
function(){
    var f = p.add([
        function(n,arg){
            setTimeout(function(){
                n(arg+1);
            }, rand());
        },
        function(n,arg){
            setTimeout(function(){
                n(arg+1);
            }, rand());
        },
        function(n,arg){
            setTimeout(function(){
                n(arg+1);
            }, rand());
        },
        function(n,arg){
            setTimeout(function(){
                n(arg+1);
            }, rand());
        },
        function(n,arg){
            setTimeout(function(){
                n(arg+1);
            }, rand());
        },
        function(n,arg){
            setTimeout(function(){
                n(arg+1);
            }, rand());
        },
        function(n,arg){
            setTimeout(function(){
                n(arg+1);
            }, rand());
        },
        function(n,arg){
            n(arg);
        }
    ]);
    r++;
    f(function(err,arg){
console.log(arg);
        r--;
        if(r===0){
console.log("aa");
            p.stop();
        }
    },0);
}
].forEach(function(f){
    for(var i=0;i<50000;++i){f();}
});