node.js http.agent の agent.maxSockets の上限を避ける その２

gistfile1.js

var events,http, url, path;

events = require('events');
http   = require('http');
url    = require('url');
path   = require('path');

var agents = {};
agent.onSocketsLengthChange;
if (http.getAgent) {
        agents.onSocketsLengthChange = function (agent) {
            var helper;
            while (agent.sockets.length < agent.maxSockets && agent.waitRequests.length > 0) {
                helper = agent.waitRequests.shift();
                helper();
            }

            agent.sockets.forEach(function (socket, i) {
                    if (! socket.hasOnCloseListener) {
                        socket.hasOnCloseListener = 1;
                        socket.once('close', function (had_error) {
                                if (had_error) {
                                    console.log('sockets[' + i + '/' + agent.sockets.length + '] is closing by error');
                                }
                                agents.onSocketsLengthChange(agent);
                        });
                        console.log('set agent.sockets[' + i + '] "hasOnCloseListener"');
                    }
            });
        };
}

function httpRequest (/* getURL, method, options, onResponse */) {
        var that, agentid, helper,
        args, getURL, uri, uriPath, method, options, headers, body, requestOptions;

        that       = this;
        args       = Array.prototype.slice.apply(arguments);
        getURL     = args.shift();
        onResponse = args.pop();
        method     = args.shift() || 'GET';
        method     = method.toUpperCase();
        options    = args.shift() || {};
        if (options.body) {
            body = options.body || null;
            delete options.body;
        }
        headers   = options;

        uri     = url.parse(getURL());
        uriPath = (! uri.pathname)                   ? '/'
                : (uri.pathname.slice(0, 1) === '/') ? uri.pathname
                :                                      '/' + uri.pathname;
        if (uri.search) uriPath += uri.search;

        if (method === 'POST') {
            headers['content-type'] = headers['content-type'] || 'application/x-www-form-urlencoded';
            headers['content-length'] = (body) ? Buffer.byteLength(body) : 0;
        }

        if (! headers.referer && this.referer) headers.referer = this.referer;
        if (! headers.cookie  && this.cookie)  headers.cookie  = this.cookie;

        requestOptions = {
            method  : method,
            host    : uri.host,
            port    : (uri.port) ? uri.port : (uri.protocol === 'https') ? '443' : '80',
            path    : uriPath,
            headers : headers
        };

        helper = function () {
                var req;
                req  = http.request(requestOptions);
                req.on('error', function (error) {
                        console.log(error);
                        process.exit(1);
                });
                req.on('response', function (response) {
                        var statusCode, responseHeaders, _onResponse;
                        statusCode      = response.statusCode;
                        responseHeaders = response.headers;
                        if (responseHeaders['set-cookie']) that.cookie = responseHeaders['set-cookie'][0];
                        that.referer = uri.href;
                
                        if (that.redirect && statusCode >= 300 && statusCode < 400 && responseHeaders.location) {
                            _onResponse = onResponse;
                            httpRequest(
                                function () { return responseHeaders.location; }, 'GET',
                                { referer : that.referer, cookie : that.cookie },
                                function (response, request) {
                                        _onResponse(response, request);
                                }
                            )
                            return;
                        }
                
                        onResponse(response, [ uri.href, headers, body ]);
                });
                if (body) req.write(body);
                req.end();
        };

        if (agents.onSocketsLengthChange) {
                agentid = [ requestOptions.host, requestOptions.port ].join(':');
                if (! agents[agentid]) {
                    agents[agentid] = http.getAgent(requestOptions.host, requestOptions.port);
                    agents[agentid].waitRequests = [];
                    agents[agentid].maxSockets   = 2;
                }
                agents[agentid].waitRequests.push(helper);
                agents.onSocketsLengthChange(agents[agentid]);
                return;
        }

        helper();
}

exports.httpRequest = httpRequest;

ダウンローダーのサンプル

var httpRequest, path, fs;

httpRequest = require('./httpclient').httpRequest;
path        = require('path');
fs          = require('fs');

var argv = process.argv;

if (! (argv.length > 2)) {
    console.log('arguments (url) is not found');
    process.exit(1);
}

argv.shift();
argv.shift();

argv.forEach(dl);

function dl (src) {
    var client = {};
    client.httpRequest = httpRequest;
    client.httpRequest(function () { return src; }, function (response, request) {
            var size, target, writeStream;
            size        = 0;
            target      = path.basename(request[0]);
            writeStream = fs.createWriteStream(target);

            console.log('target: "' + target + '".');
            response.pipe(writeStream);

            writeStream.on('error', function (e) {
                console.log(e);
            });
            writeStream.on('pipe', function (src) {
                console.log(src);
            });
            response.on('data', function (chunk) {
                size += chunk.length;
                console.log(target + ': ' + size);
            });
            response.on('end', function () {
                console.log(target + ': end');
            });
    });
}

node 0.5.3 以降には、HTTP クライアントリクエストのソケットを プーリングするために新しい HTTP Agent の実装が存在します。

以前は、エージェントの一つのインスタンスが一つのホスト + ポートのプールを 助けていましたが、現在の実装では任意の数のホストに対するソケットを 保持できるようになりました。

現在の HTTP Agent では、クライアントリクエストはデフォルトで Connection:keep-alive を使うようにもなりました。 ソケットを待ってペンディングになっている HTTP リクエストがなければ、 ソケットはクローズされます。 これは、node のプールは高負荷時に keep-alive のメリットを持ちながら、 keep-alive を使用する HTTP クライアントを開発者が手動でクローズする 必要がないことを意味します。

ということなので、v5.3+ だとこんな処理の必要がない

追記: v5.3+ 以上でもエラーが起きないように対応した 2012.04.03

勉強になりました。助かりました、ありがとうございます。

var httpRequest = function(/* getURL, method, options, onResponse */)
{
//..
}

のほうが良いように思います。WebStormでは、コンスタントなバリューをエクスポートしようとしているというエラーがでました。

kenokabe

ご指摘ありがとうございます。
WebStrom は使ったことなかったので見直したいと思います。

どういたしまして、
あと、どうしてもわからないので、教えていただきたいのですが、
Cannot read property 'onSocketsLengthChange' of undefined
という例外がでまして、
それは多分agentがundefinedなのだと思いますが、
特に、ライブラリの
var agents;
if (http.getAgent) {
agents = {};
agents.onSocketsLengthChange = function (agent) {
......
の部分は、
続くダウンローダーサンプルのように、
httpRequest = require('./httpclient').httpRequest;
としても、けしてリーチできないブロックにあるのではないでしょうか？

node のバージョンはなんでしょうか？
v0.4.12 では agents.onSocketsLengthChange は正常に呼び出されてるんで、なんだろう...

追記:
バグでした。直しました。

v0.6.17(現時点で最新のバージョン)です。
IRCでこの辺よくわからず質問したところ、このバージョンでhttp周りのメモリリークなど多くの改善がされているようです。
修正どうもありがとうございました。

こちらこそ勉強になりました

8行目の

var agents;

の所で、 agents が定義されていないのが問題だったので

var agents = {};
agents.onSocketsLengthChange;

とすることでバグを回避しました。

実際の所、古いバージョンを使わない限り agent 周りは弄る必要がないので、不要な部分は削除していいと思います