<commit_before>/*
 * moleculer
 * Copyright (c) 2018 MoleculerJS (https://github.com/moleculerjs/moleculer)
 * MIT Licensed
 */

"use strict";

const url = require("url");
const Promise		= require("bluebird");
const Transporter 	= require("./base");
const { isPromise }	= require("../utils");

const {
	PACKET_REQUEST,
	PACKET_RESPONSE,
	PACKET_UNKNOW,
	PACKET_EVENT,
	PACKET_DISCOVER,
	PACKET_INFO,
	PACKET_DISCONNECT,
	PACKET_HEARTBEAT,
	PACKET_PING,
	PACKET_PONG,
} = require("../packets");

/**
 * Transporter for AMQP
 *
 * More info: https://www.amqp.org/
 *
 * For test:
 *
 * 	 docker run -d -p 5672:5672 -p 15672:15672 --name rabbit rabbitmq:3-management
 *
 * @class AmqpTransporter
 * @extends {Transporter}
 */
class AmqpTransporter extends Transporter {

	/**
	 * Creates an instance of AmqpTransporter.
	 *
	 * @param {any} opts
	 *
	 * @memberof AmqpTransporter
	 */
	constructor(opts) {
		if (typeof opts === "string")
			opts = { url: opts };
		else if (opts == null)
			opts = {};

		// Number of requests a broker will handle concurrently
		if (typeof opts.prefetch !== "number")
			opts.prefetch = 1;

		// Number of milliseconds before an event expires
		if (typeof opts.eventTimeToLive !== "number")
			opts.eventTimeToLive = null;

		if (typeof opts.heartbeatTimeToLive !== "number")
			opts.heartbeatTimeToLive = null;

		if (typeof opts.queueOptions !== "object")
			opts.queueOptions = {};

		if (typeof opts.exchangeOptions !== "object")
			opts.exchangeOptions = {};

		if (typeof opts.messageOptions !== "object")
			opts.messageOptions = {};

		if (typeof opts.consumeOptions !== "object")
			opts.consumeOptions = {};

		opts.autoDeleteQueues =
			opts.autoDeleteQueues === true ? 2*60*1000 :
				typeof opts.autoDeleteQueues === "number" ? opts.autoDeleteQueues :
					opts.autoDeleteQueues === false ? -1 :
						-1; // Eventually we could change default

		// Support for multiple URLs (clusters)
		opts.url = Array.isArray(opts.url)
			? opts.url
			: !opts.url
				? ['']
				: opts.url.split(';').filter(s => !!s)

		super(opts);

		this.hasBuiltInBalancer = true;
		this.connection = null;
		this.channel = null;
		this.bindings = [];

		this.channelDisconnecting = false;
		this.connectionDisconnecting = false;
	}

	/**
	 * Connect to a AMQP server
	 *
	 * @memberof AmqpTransporter
	 */
	connect(errorCallback) {
		return new Promise((_resolve, _reject) => {
			let _isResolved = false;
			const resolve = () => {
				_isResolved = true;
				_resolve();
			};
			const reject = (err) => {
				_reject(err);
				if (_isResolved) errorCallback(err);
			};

			let amqp;
			try {
				amqp = require("amqplib");
			} catch(err) {
				/* istanbul ignore next */
				this.broker.fatal("The 'amqplib' package is missing. Please install it with 'npm install amqplib --save' command.", err, true);
			}

			// Pick url
			this.connectAttempt = (this.connectAttempt||0)+1;
			const urlIndex = (this.connectAttempt-1) % this.opts.url.length;
			const uri = this.opts.url[urlIndex];
			const urlParsed = url.parse(uri);

			amqp.connect(uri, {
				...(this.opts.socketOptions || {}),
				servername: urlParsed.hostname
			})
				.then(connection => {
					this.connection = connection;
					this.logger.info("AMQP is connected.");

					/* istanbul ignore next*/
					connection
						.on("error", (err) => {
							// No need to reject here since close event will be fired after
							// if not connected at all connection promise will be rejected
							// this.connected = false;
							// reject(err);
							this.logger.error("AMQP connection error.", err);
						})
						.on("close", (err) => {
							this.connected = false;
							reject(err);
							if (!this.connectionDisconnecting)
								this.logger.error("AMQP connection is closed.");
							else
								this.logger.info("AMQP connection is closed gracefully.");
						})
						.on("blocked", (reason) => {
							this.logger.warn("AMQP connection is blocked.", reason);
						})
						.on("unblocked", () => {
							this.logger.info("AMQP connection is unblocked.");
						});

					connection
						.createChannel()
						.then((channel) => {
							this.channel = channel;
							this.logger.info("AMQP channel is created.");

							channel.prefetch(this.opts.prefetch);

							/* istanbul ignore next*/
							channel
								.on("close", () => {<commit_msg>Fix AMQP connection recovery: don't report connection error on graceful disconnect<commit_after>/*
 * moleculer
 * Copyright (c) 2018 MoleculerJS (https://github.com/moleculerjs/moleculer)
 * MIT Licensed
 */

"use strict";

const url = require("url");
const Promise		= require("bluebird");
const Transporter 	= require("./base");
const { isPromise }	= require("../utils");

const {
	PACKET_REQUEST,
	PACKET_RESPONSE,
	PACKET_UNKNOW,
	PACKET_EVENT,
	PACKET_DISCOVER,
	PACKET_INFO,
	PACKET_DISCONNECT,
	PACKET_HEARTBEAT,
	PACKET_PING,
	PACKET_PONG,
} = require("../packets");

/**
 * Transporter for AMQP
 *
 * More info: https://www.amqp.org/
 *
 * For test:
 *
 * 	 docker run -d -p 5672:5672 -p 15672:15672 --name rabbit rabbitmq:3-management
 *
 * @class AmqpTransporter
 * @extends {Transporter}
 */
class AmqpTransporter extends Transporter {

	/**
	 * Creates an instance of AmqpTransporter.
	 *
	 * @param {any} opts
	 *
	 * @memberof AmqpTransporter
	 */
	constructor(opts) {
		if (typeof opts === "string")
			opts = { url: opts };
		else if (opts == null)
			opts = {};

		// Number of requests a broker will handle concurrently
		if (typeof opts.prefetch !== "number")
			opts.prefetch = 1;

		// Number of milliseconds before an event expires
		if (typeof opts.eventTimeToLive !== "number")
			opts.eventTimeToLive = null;

		if (typeof opts.heartbeatTimeToLive !== "number")
			opts.heartbeatTimeToLive = null;

		if (typeof opts.queueOptions !== "object")
			opts.queueOptions = {};

		if (typeof opts.exchangeOptions !== "object")
			opts.exchangeOptions = {};

		if (typeof opts.messageOptions !== "object")
			opts.messageOptions = {};

		if (typeof opts.consumeOptions !== "object")
			opts.consumeOptions = {};

		opts.autoDeleteQueues =
			opts.autoDeleteQueues === true ? 2*60*1000 :
				typeof opts.autoDeleteQueues === "number" ? opts.autoDeleteQueues :
					opts.autoDeleteQueues === false ? -1 :
						-1; // Eventually we could change default

		// Support for multiple URLs (clusters)
		opts.url = Array.isArray(opts.url)
			? opts.url
			: !opts.url
				? ['']
				: opts.url.split(';').filter(s => !!s)

		super(opts);

		this.hasBuiltInBalancer = true;
		this.connection = null;
		this.channel = null;
		this.bindings = [];

		this.channelDisconnecting = false;
		this.connectionDisconnecting = false;
	}

	/**
	 * Connect to a AMQP server
	 *
	 * @memberof AmqpTransporter
	 */
	connect(errorCallback) {
		return new Promise((_resolve, _reject) => {
			let _isResolved = false;
			const resolve = () => {
				_isResolved = true;
				_resolve();
			};
			const reject = (err) => {
				_reject(err);
				if (_isResolved) errorCallback(err);
			};

			let amqp;
			try {
				amqp = require("amqplib");
			} catch(err) {
				/* istanbul ignore next */
				this.broker.fatal("The 'amqplib' package is missing. Please install it with 'npm install amqplib --save' command.", err, true);
			}

			// Pick url
			this.connectAttempt = (this.connectAttempt||0)+1;
			const urlIndex = (this.connectAttempt-1) % this.opts.url.length;
			const uri = this.opts.url[urlIndex];
			const urlParsed = url.parse(uri);

			amqp.connect(uri, {
				...(this.opts.socketOptions || {}),
				servername: urlParsed.hostname
			})
				.then(connection => {
					this.connection = connection;
					this.logger.info("AMQP is connected.");

					/* istanbul ignore next*/
					connection
						.on("error", (err) => {
							// No need to reject here since close event will be fired after
							// if not connected at all connection promise will be rejected
							// this.connected = false;
							// reject(err);
							this.logger.error("AMQP connection error.", err);
						})
						.on("close", (err) => {
							this.connected = false;
							if (!this.connectionDisconnecting) {
								this.logger.error("AMQP connection is closed.");
								reject(err);
							} else {
								this.logger.info("AMQP connection is closed gracefully.");
							}
						})
						.on("blocked", (reason) => {
							this.logger.warn("AMQP connection is blocked.", reason);
						})
						.on("unblocked", () => {
							this.logger.info("AMQP connection is unblocked.");
						});

					connection
						.createChannel()
						.then((channel) => {
							this.channel = channel;
							this.logger.info("AMQP channel is created.");

							channel.prefetch(this.opts.prefetch);

							/* istanbul ignore next*/
							channel
								.on("close", () => {
{'fraction_non_alphanumeric': 0.14595392123482012, 'fraction_numerical': 0.008171603677221655, 'mean_word_length': 2.866608161474331, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 25, 'https://': 4, 'lorem ipsum': 0, 'www.': 2, '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': '3482090', 'n_tokens_mistral': 3581, 'n_tokens_neox': 2795, 'n_words': 674}