connect(errorCallback) {
		return new this.broker.Promise((_resolve, _reject) => {
			let _isResolved = false;
			const resolve = () => {
				_isResolved = true;
				_resolve();
			};
			const reject = err => {
				_reject(err);

				// Returning callback to avoid race condition in the tests
				if (_isResolved) return 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,
				Object.assign({}, 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.logger.error("AMQP connection error.", err);

							this.broker.broadcastLocal("$transporter.error", {
								error: err,
								module: "transporter",
								type: C.FAILED_CONNECTION_ERROR
							});
						})
						.on("close", err => {
							this.connected = false;
							if (!this.connectionDisconnecting) {
								this.logger.error("AMQP connection is closed.");
								return 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", () => {
									this.channel = null;
									// No need to reject here since close event on connection will handle
									if (!this.channelDisconnecting)
										this.logger.warn("AMQP channel is closed.");
									else this.logger.info("AMQP channel is closed gracefully.");
								})
								.on("error", err => {
									// No need to reject here since close event will be fired after
									this.logger.error("AMQP channel error.", err);

									this.broker.broadcastLocal("$transporter.error", {
										error: err,
										module: "transporter",
										type: C.FAILED_CHANNEL_ERROR
									});
								})
								.on("drain", () => {
									this.logger.info("AMQP channel is drained.");
								})
								.on("return", msg => {
									this.logger.warn("AMQP channel returned a message.", msg);
								});

							this.connectionCount += 1;
							const isReconnect = this.connectionCount > 1;

							// HACK: We have to do this because subscriptions aren't persistent,
							// and only balanced subscriptions happen automatically on reconnects
							const p = isReconnect
								? this.transit.makeSubscriptions()
								: this.broker.Promise.resolve();
							return p.then(() => this.onConnected(isReconnect));
						})
						.then(resolve)
						.catch(err => {
							/* istanbul ignore next*/
							this.logger.error("AMQP failed to create channel.");
							this.connected = false;
							return reject(err);
						});
					return null;
				})
				.catch(err => {
					/* istanbul ignore next*/
					this.logger.warn("AMQP failed to connect!");
					this.connected = false;
					return reject(err);
				});
		});
	}
Connect to a AMQP server

@memberof AmqpTransporter