<commit_before>var
  stringify = require('json-stable-stringify'),
// Main library for manipulate amqp protocol like RabbitMQ
  amqp = require('amqplib'),
// Promise created by amqplib for handle amqp connection
  pConnection,
  mqEchangeName;

module.exports = broker =  {

  kuzzle: null,

  /**
   * Initialize the connection with the amqp broker
   * @param kuzzle
   * @returns {boolean}
   */
  init: function (kuzzle) {
    if (pConnection) {
      return false;
    }

    pConnection = amqp.connect('amqp://' + kuzzle.config.broker.host);
    mqEchangeName = 'amq.topic';

    broker.kuzzle = kuzzle;
  },

  /**
   * Allow to add an object in a specific room.
   * @param room
   * @param data object that must be insert in queue
   */
  add: function (room, data) {
    if (!room) {
      return false;
    }

    pConnection.then(function (conn) {
      return conn.createChannel()
        .then(function (channel) {
          channel.assertQueue(room, {durable: true});
          channel.sendToQueue(room, new Buffer(stringify(data)), {deliveryMode: true});
        })
        .catch(function (error) {
          broker.kuzzle.log.error(error);
        });
    });
  },

  /**
   * Allow to add an object to a specific exchange/routing_key
   * @param routingKey
   * @param data object that must be insert into routing key
   */
  addExchange: function (routingKey, data) {
    pConnection.then(function (conn) {
      return conn.createChannel().then(function (channel) {
        return channel.assertExchange(mqEchangeName, 'topic', {durable: true})
          .then(function () {
            return channel.publish(mqEchangeName,routingKey,new Buffer(stringify(data)), {deliveryMode: true});
          })
          .catch(function (error) {
            broker.kuzzle.log.error(error);
          });
      });
    });
  },

  /**
   * Allow to add an object to a specific exchange/routing_key
   * @param room
   * @param data object that must be insert into routing key
   */
  replyTo: function (room, data) {
    pConnection.then(function (conn) {
      return conn.createChannel().then(function (channel) {
        return channel.sendToQueue(room,new Buffer(stringify(data)), {deliveryMode: true});
      });
    });
  },
  /**
   * Listen a specific room and execute a callback for each messages
   *
   * @param {String} room
   * @param {Function} onListenCB called each times a message is received
   * @param {Boolean} destroy if true, the channel is closed after message reading
   * @returns promise
   */
  listen: function (room, onListenCB, destroy) {
    pConnection.then(function (conn) {
      return conn.createChannel().then(function (channel) {
        return channel.assertQueue(room, {durable: true})
          .then(function () {
            channel.prefetch(1);
          })
          .then(function () {
            channel.consume(room, function doWork (msg) {
              channel.ack(msg);
              onListenCB(JSON.parse(msg.content.toString()));

              if(destroy) {
                channel.close();
              }
            });
          })
          .catch(function (error) {
            broker.kuzzle.log.error(error);
          });
      });
    });
  },

  /**
   * Listen an exchange to a specific routing key and execute a callback for each message
   *
   * @param routingKey
   * @param onListenCB called each times a message is received
   * @returns promise
   */
  listenExchange: function (routingKey, onListenCB) {
    pConnection.then(function (conn) {
      return conn.createChannel().then(function (channel) {
        return channel.assertExchange(mqEchangeName, 'topic', {durable: true})
          .then(function () {
            return channel.assertQueue('', {exclusive: true});
          })
          .then(function (qok) {
            var queue = qok.queue;
            channel.bindQueue(queue, mqEchangeName, routingKey);
            channel.consume(queue, function doWork (msg) {
              onListenCB(msg);
            });
          })
          .catch(function (error) {
            broker.kuzzle.log.error(error);
          });
      });
    });
  },

  close: function () {
    pConnection.then(function (conn) {
      process.once('SIGINT', function () {
        conn.close();
      });
    });
  }

};<commit_msg>AMQP Broker management - Added a panic function in case RabbitMQ sends us an error event: this will restart the connection instead of letting Kuzzle unresponsive - Instead of creating 1 channel per data sent, the broker now uses only one all-around channel. This unique channel is only used for 1-time write operations: listeners still use their own channels.<commit_after>var
  stringify = require('json-stable-stringify'),
  amqp = require('amqplib'),
  // AMQP management variables
  amqpConnection,
  amqpAdminChannel,
  amqpDomain = require('domain').create(),
  // Default AMQP exchange name. Constant.
  AMQ_EXCHANGE = 'amq.topic';

module.exports = broker =  {
  kuzzle: null,

  /**
   * Initialize the connection with the amqp broker
   * @param kuzzle
   */
  init: function (kuzzle) {
    if (amqpConnection) {
      return false;
    }

    amqpDomain.on('error', this.onErrorRestart);

    amqpConnection = amqp.connect('amqp://' + kuzzle.config.broker.host);

    amqpConnection
      .then(function (connection) {
        amqpDomain.add(connection);
        amqpAdminChannel = connection.createChannel();
      });

    broker.kuzzle = kuzzle;
  },

  /**
   * Sends data to a room
   * @param room
   * @param data object that must be insert in queue
   */
  add: function (room, data) {
    if (!room) {
      return false;
    }

    amqpAdminChannel
      .then(function (channel) {
        channel.assertQueue(room, {durable: true})
          .then(function () {
            channel.sendToQueue(room, new Buffer(stringify(data)), {deliveryMode: true});
          });
      })
      .catch(function (error) {
        broker.kuzzle.log.error(new Error(error));
      });
  },

  /**
   * Allow to add an object to a specific exchange/routing_key
   * @param routingKey
   * @param data object that must be insert into routing key
   */
  addExchange: function (routingKey, data) {
    amqpAdminChannel
      .then(function (channel) {
        return channel.assertExchange(AMQ_EXCHANGE, 'topic', {durable: true})
          .then(function () {
            return channel.publish(AMQ_EXCHANGE, routingKey, new Buffer(stringify(data)), {deliveryMode: true});
          });
      })
      .catch(function (error) {
        broker.kuzzle.log.error(new Error(error));
      });
  },

  /**
   * Sends a reply to a room
   * @param room
   * @param data object that must be insert into routing key
   */
  replyTo: function (room, data) {
    amqpAdminChannel
      .then(function(channel) {
        channel.sendToQueue(room, new Buffer(stringify(data)), {deliveryMode: true});
      });
  },

  /**
   * Listen a specific room and execute a callback for each messages
   *
   * @param {String} room
   * @param {Function} onListenCB called each times a message is received
   * @param {Boolean} destroy if true, the channel is closed after a message has been read
   * @returns promise
   */
  listen: function (room, onListenCB, destroy) {
    amqpConnection.then(function (conn) {
      return conn.createChannel().then(function (channel) {
        return channel.assertQueue(room, {durable: true})
          .then(function () {
            channel.prefetch(1);
          })
          .then(function () {
            channel.consume(room, function doWork (msg) {
              channel.ack(msg);
              onListenCB(JSON.parse(msg.content.toString()));

              if(destroy) {
                channel.close();
              }
            });
          })
          .catch(function (error) {
            broker.kuzzle.log.error(new Error(error));
          });
      });
    });
  },

  /**
   * Listen an exchange to a specific routing key and execute a callback for each message
   *
   * @param routingKey
   * @param onListenCB called each times a message is received
   * @returns promise
   */
  listenExchange: function (routingKey, onListenCB) {
    amqpConnection.then(function (conn) {
      return conn.createChannel().then(function (channel) {
        return channel.assertExchange(AMQ_EXCHANGE, 'topic', {durable: true})
          .then(function () {
            return channel.assertQueue('', {exclusive: true});
          })
          .then(function (qok) {
            var queue = qok.queue;
            channel.bindQueue(queue, AMQ_EXCHANGE, routingKey);
            channel.consume(queue, function doWork (msg) {
              onListenCB(msg);
            });
          })
          .catch(function (error) {
            broker.kuzzle.log.error(new Error(error));
          });
      });
    });
  },

  close: function () {
    amqpConnection.then(function (conn) {
      process.once('SIGINT', function () {
        conn.close();
      });
    });
  },

  /**
   * Broker Error: now the AMQP connection is either closed or in a very bad shape.
   * Two options:
   *    - Restart the connection, hoping we didn't lose too many data
   *    - Panic
   *
   * Here is the implementation for the 1st option.
   */
  onErrorRestart: function () {
    broker.kuzzle.log.error('Catched an error from the AMQP broker service. Restarting connection...');

    // Trying to close the connection cleanly.
    if (amqpConnection) {
      amqpDomain.remove(amqpConnection);
      amqpConnection.close()
        .catch(function (error) {
          broker.kuzzle.log.error(new Error(error));
        });

      amqpConnection = null;
    }

    this.init(this.kuzzle);
  }
};
{'fraction_non_alphanumeric': 0.1209843492586491, 'fraction_numerical': 0.000514827018121911, 'mean_word_length': 2.328649760109664, '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': '6502068', 'n_tokens_mistral': 2839, 'n_tokens_neox': 2765, 'n_words': 740}