<commit_before>/*
 * Copyright (c) 2012, Joyent, Inc. All rights reserved.
 *
 * A brief overview of this source file: what is its purpose.
 */


var util = require('util');
var amqp = require('amqp');
var assert = require('assert');
var common = require('./common');
var EventEmitter = require('events').EventEmitter;



/*
 * Heartbeater constructor. options is an object with the following properties:
 *  - host: AMQP host
 *  - queue: AMQP queue. Defaults to 'heartbeat.vmapi'
 *  - log: bunyan logger instance
 */
function Heartbeater(options) {
    if (typeof (options) !== 'object')
        throw new TypeError('amqp options (Object) required');
    if (typeof (options.host) !== 'string')
        throw new TypeError('amqp host (String) required');

    this.host = options.host;
    this.queue = options.queue || 'heartbeat.vmapi';
    this.log = options.log;
    // Ths array helps us find which vms are no longer on a server and
    // call ufds.updateVm
    this.lastSeenUuids = [];

    EventEmitter.call(this);

    this.connection = amqp.createConnection({ host: this.host });
    this.reconnectTimeout = options.reconnect * 1000;

    this.connection.on('error', this.onError.bind(this));
    this.connection.on('ready', this.onReady.bind(this));
}

util.inherits(Heartbeater, EventEmitter);



/*
 * Reconnects to the AMQP host
 */
Heartbeater.prototype.reconnect = function () {
    this.connection.reconnect();
};



/*
 * On error gets called when a AMQP connection error is produced
 */
Heartbeater.prototype.onError = function (err) {
    var self = this;

    this.log.error('AMQP Connection Error ' + err.code +
                   ', re-trying in 5 seconds...');

    setTimeout(function () {
        self.reconnect();
    }, this.reconnectTimeout);
};



/*
 * On ready gets called when the connection to the AMQP was successfully
 * established. From here we can start using AMQP
 */
Heartbeater.prototype.onReady = function () {
    var self = this;
    var connection = self.connection;

    self.log.debug('Connected to AMQP');
    var queue = connection.queue(self.queue);

    queue.on('open', function () {
        self.log.debug('Binded queue to exchange');
        queue.bind('heartbeat.*');

        queue.subscribeJSON(function (message, headers, deliveryInfo) {
            assert(message);
            assert(deliveryInfo.routingKey);

            var serverUuid = deliveryInfo.routingKey.split('.')[1];
            var missing = self.getMissingUuids(message.zoneStatus);

            self.emit('heartbeat', serverUuid, message.zoneStatus, missing);
        });
    });
};



/*
 * Compares the new set of UUIDs with the previously cached. If there are
 * missing UUIDs we can assume the vm was destroyed and mark as destroyed
 * on UFDS.
 */
Heartbeater.prototype.getMissingUuids = function (hbs) {
    var uuids = [];
    var missing = [];

    hbs.forEach(function (hb) {
        var uuid = hb[1];
        if (uuid != 'global')
            uuids.push(uuid);
    });

    this.lastSeenUuids.forEach(function (uuid) {
        if (uuids.indexOf(uuid) == -1)
            missing.push(uuid);
    });

    this.lastSeenUuids = uuids;

    return missing;
};

<commit_msg>ZAPI-96: Heartbeater can't handle more than one server<commit_after>/*
 * Copyright (c) 2012, Joyent, Inc. All rights reserved.
 *
 * A brief overview of this source file: what is its purpose.
 */


var util = require('util');
var amqp = require('amqp');
var assert = require('assert');
var common = require('./common');
var EventEmitter = require('events').EventEmitter;



/*
 * Heartbeater constructor. options is an object with the following properties:
 *  - host: AMQP host
 *  - queue: AMQP queue. Defaults to 'heartbeat.vmapi'
 *  - log: bunyan logger instance
 */
function Heartbeater(options) {
    if (typeof (options) !== 'object')
        throw new TypeError('amqp options (Object) required');
    if (typeof (options.host) !== 'string')
        throw new TypeError('amqp host (String) required');

    this.host = options.host;
    this.queue = options.queue || 'heartbeat.vmapi';
    this.log = options.log;
    // Ths helps us find which vms are no longer on a server and
    // call ufds.updateVm
    this.lastSeenUuids = {};

    EventEmitter.call(this);

    this.connection = amqp.createConnection({ host: this.host });
    this.reconnectTimeout = options.reconnect * 1000;

    this.connection.on('error', this.onError.bind(this));
    this.connection.on('ready', this.onReady.bind(this));
}

util.inherits(Heartbeater, EventEmitter);



/*
 * Reconnects to the AMQP host
 */
Heartbeater.prototype.reconnect = function () {
    this.connection.reconnect();
};



/*
 * On error gets called when a AMQP connection error is produced
 */
Heartbeater.prototype.onError = function (err) {
    var self = this;

    this.log.error('AMQP Connection Error ' + err.code +
                   ', re-trying in 5 seconds...');

    setTimeout(function () {
        self.reconnect();
    }, this.reconnectTimeout);
};



/*
 * On ready gets called when the connection to the AMQP was successfully
 * established. From here we can start using AMQP
 */
Heartbeater.prototype.onReady = function () {
    var self = this;
    var connection = self.connection;

    self.log.debug('Connected to AMQP');
    var queue = connection.queue(self.queue);

    queue.on('open', function () {
        self.log.debug('Binded queue to exchange');
        queue.bind('heartbeat.*');

        queue.subscribeJSON(function (message, headers, deliveryInfo) {
            assert(message);
            assert(deliveryInfo.routingKey);

            var serverUuid = deliveryInfo.routingKey.split('.')[1];

            if (!self.lastSeenUuids[serverUuid])
                self.lastSeenUuids[serverUuid] = [];

            var missing = self.getMissingUuids(serverUuid, message.zoneStatus);
            self.emit('heartbeat', serverUuid, message.zoneStatus, missing);
        });
    });
};



/*
 * Compares the new set of UUIDs with the previously cached. If there are
 * missing UUIDs we can assume the vm was destroyed and mark as destroyed
 * on UFDS.
 */
Heartbeater.prototype.getMissingUuids = function (server, hbs) {
    var uuids = [];
    var missing = [];

    hbs.forEach(function (hb) {
        var uuid = hb[1];
        if (uuid != 'global')
            uuids.push(uuid);
    });

    this.lastSeenUuids[server].forEach(function (uuid) {
        if (uuids.indexOf(uuid) == -1)
            missing.push(uuid);
    });

    this.lastSeenUuids[server] = uuids;

    return missing;
};


{'fraction_non_alphanumeric': 0.13114754098360656, 'fraction_numerical': 0.003946569520340012, 'mean_word_length': 2.8713278495887193, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '2084582', 'n_tokens_mistral': 2053, 'n_tokens_neox': 1959, 'n_words': 541}