<commit_before>var amqp = require('amqp');
var uuid = require('node-uuid').v4;
var os = require('os');
var queueNo = 0;

function rpc(opt) {

    if (!opt) opt = {};
    this.opt = opt;
    this.__conn = opt.connection ? opt.connection : null;
    this.__url = opt.url ? opt.url : 'amqp://guest:guest@localhost:5672';
    this.__exchange = opt.exchangeInstance ? opt.exchangeInstance : null;
    this.__exchange_name = opt.exchange ? opt.exchange : 'rpc_exchange';
    this.__exchange_options = opt.exchange_options ? opt.exchange_options : {exclusive: false, autoDelete: true};
    this.__impl_options = opt.ipml_options || {defaultExchangeName: this.__exchange_name};
    this.__conn_options = opt.conn_options || {};

    this.__results_queue = null;
    this.__results_queue_name = null;
    this.__results_cb = {};
    this.__make_results_cb = [];

    this.__cmds = {};

    this.__connCbs = [];
    this.__exchangeCbs = [];
}

/**
 * generate unique name for new queue
 * @returns {string}
 */

rpc.prototype.generateQueueName = function (type) {
    return uuid() + ':' + os.hostname() + ':pid' + process.pid + ':' + type;
}


rpc.prototype._connect = function (cb) {

    if (!cb) {
        cb = function () {
        };
    }

    if (this.__conn) {

        if (this.__connCbs.length > 0) {

            this.__connCbs.push(cb);

            return true;
        }

        return cb(this.__conn);
    } else {
        this.__conn = this.opt.connection ? this.opt.connection : null;
    }

    var $this = this;

    this.__connCbs.push(cb);
    var options = this.__conn_options;
    if (!options.url && !options.host) options.url = this.__url;
    this.__conn = amqp.createConnection(
        options,
        this.__impl_options
    );

    this.__conn.on('error', function (err) {
        throw err;
    });

    this.__conn.on('ready', function () {
        var cbs = $this.__connCbs;
        $this.__connCbs = [];

        for (var i = 0; i < cbs.length; i++) {
            cbs[i]($this.__conn);
        }
    });
}
/**
 * disconnect from MQ broker
 */

rpc.prototype.disconnect = function () {
    if (!this.__conn) return;
    this.__conn.end();
    this.__conn = null;
}

rpc.prototype._makeExchange = function (cb) {

    if (!cb) {
        cb = function () {
        };
    }

    if (this.__exchange) {

        if (this.__exchangeCbs.length > 0) {

            this.__exchangeCbs.push(cb);

            return true;
        }

        return cb(this.__exchange);
    }
<commit_msg>bugfix discinnection problem<commit_after>var amqp = require('amqp');
var uuid = require('node-uuid').v4;
var os = require('os');
var queueNo = 0;

var instanceId = 0;

function rpc(opt) {
    this.instanceId = instanceId++;

    if (!opt) opt = {};
    this.opt = opt;
    this.__conn = opt.connection ? opt.connection : null;
    this.__url = opt.url ? opt.url : 'amqp://guest:guest@localhost:5672';
    this.__exchange = opt.exchangeInstance ? opt.exchangeInstance : null;
    this.__exchange_name = opt.exchange ? opt.exchange : 'rpc_exchange';
    this.__exchange_options = opt.exchange_options ? opt.exchange_options : {exclusive: false, autoDelete: true};
    this.__impl_options = opt.ipml_options || {defaultExchangeName: this.__exchange_name};
    this.__conn_options = opt.conn_options || {};

    this.__results_queue = null;
    this.__results_queue_name = null;
    this.__results_cb = {};
    this.__make_results_cb = [];

    this.__cmds = {};

    this.__connCbs = [];
    this.__exchangeCbs = [];
}

/**
 * generate unique name for new queue
 * @returns {string}
 */

rpc.prototype.generateQueueName = function (type) {
    return uuid() + ':' + os.hostname() + ':pid' + process.pid + ':' + type;
}


rpc.prototype._connect = function (cb) {

    if (!cb) {
        cb = function () {
        };
    }

    if (this.__conn) {

        if (this.__connCbs.length > 0) {

            this.__connCbs.push(cb);

            return true;
        }

        return cb(this.__conn);
    } else {
        this.__conn = this.opt.connection ? this.opt.connection : null;
    }

    var $this = this;

    this.__connCbs.push(cb);
    var options = this.__conn_options;
    if (!options.url && !options.host) options.url = this.__url;
    this.__conn = amqp.createConnection(
        options,
        this.__impl_options
    );

    this.__conn.on('error', function (err) {
        throw err;
    });

    this.__conn.on('ready', function () {
        var cbs = $this.__connCbs;
        $this.__connCbs = [];

        for (var i = 0; i < cbs.length; i++) {
            cbs[i]($this.__conn);
        }
    });
}
/**
 * disconnect from MQ broker
 */

rpc.prototype.disconnect = function () {
    if (!this.__conn) return;
    this.__conn.removeAllListeners('error');
    var self = this;
    this.__conn.on('error', function () {
        console.log('--> error disconnecting amqp-rpc #' + self.instanceId);
    });
    this.__conn.disconnect();
    this.__conn = null;
};

rpc.prototype._makeExchange = function (cb) {

    if (!cb) {
        cb = function () {
        };
    }

    if (this.__exchange) {

        if (this.__exchangeCbs.length > 0) {

            this.__exchangeCbs.push(cb);

            return true;
        }

        return cb(this.__exchange);
    }

{'fraction_non_alphanumeric': 0.1506719666855953, 'fraction_numerical': 0.0035964414158621994, 'mean_word_length': 2.3656050955414014, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 8, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3965249', 'n_tokens_mistral': 1767, 'n_tokens_neox': 1675, 'n_words': 338}