inceptum/dist/rabbitmq/RabbitmqClient.js

"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
const amqplib_1 = require("amqplib");
const ReadyGate_1 = require("../util/ReadyGate");
const PromiseUtil_1 = require("../util/PromiseUtil");
const RabbitmqConfig_1 = require("./RabbitmqConfig");
class RabbitmqClient {
    constructor(clientConfig, name) {
        /**
         * Whether the connection should be closed. This is not a statud indicator that show whether the connection is closed.
         * It is a flag that indicates whether the client has been purposely closed by code, as opposed to being closed because of an error.
         */
        this.closed = false;
        this.reconnecting = false;
        this.readyGate = new ReadyGate_1.ReadyGate();
        this.connectFunction = amqplib_1.connect;
        this.clientConfig = Object.assign({ protocol: 'amqp', maxConnectionAttempts: RabbitmqConfig_1.DEFAULT_MAX_CONNECTION_ATTEMPTS }, clientConfig);
        this.name = name;
        this.readyGate.channelNotReady();
    }
    async init() {
        await this.connect();
    }
    /**
     * Connect to RabbitMQ broker
     */
    async connect() {
        await this.createConnection();
        await this.createChannel();
        this.reconnecting = false;
        this.readyGate.channelReady();
    }
    async createConnection() {
        const newConnection = await this.connectFunction(this.clientConfig);
        newConnection.on('close', (err) => { this.handleErrorOrClose('Connection Closed', err); });
        newConnection.on('error', (err) => { this.handleErrorOrClose('Connection Error', err); });
        this.connection = newConnection;
        this.logger.info('Connection established');
    }
    async createChannel() {
        const newChannel = await this.connection.createChannel();
        newChannel.on('close', (err) => { this.handleErrorOrClose('Channel closed unexpectedly', err); });
        newChannel.on('error', (err) => { this.handleErrorOrClose('Channel error', err); });
        this.channel = newChannel;
        this.logger.info('Channel opened');
    }
    handleErrorOrClose(cause, err) {
        if (err) {
            this.logger.error(err, this.debugMsg(`${cause}.`));
        }
        else {
            this.logger.error(this.debugMsg(`${cause}.`));
        }
        this.closeAllAndScheduleReconnection();
    }
    async closeAllAndScheduleReconnection() {
        if (!this.reconnecting) {
            this.readyGate.channelNotReady();
            this.reconnecting = true;
            if (this.channel) {
                try {
                    await this.closeChannel();
                }
                catch (e) {
                    // Do nothing... we tried to play nice
                }
            }
            if (this.connection) {
                try {
                    await this.closeConnection();
                }
                catch (e) {
                    // Do nothing... we tried to play nice
                    // this.logger.error(e);
                }
            }
            await this.attemptReconnection();
        }
        else {
            this.logger.warn(this.debugMsg('already reconnecting'));
        }
    }
    backoffWait(tryNum) {
        // 1 second, 5 seconds, 25 seconds, 30 seconds, 30 seconds, ....
        const waitBase = Math.min(Math.pow(5, Math.max(0, tryNum - 1)), 30) * 1000;
        const waitMillis = waitBase + (Math.round(Math.random() * 800));
        this.logger.warn(this.debugMsg(`Waiting for attempt #${tryNum} - ${waitMillis} ms`));
        return PromiseUtil_1.PromiseUtil.sleepPromise(waitMillis, null);
    }
    async attemptReconnection() {
        this.logger.warn(this.debugMsg(`reconnecting... max attempts ${this.clientConfig.maxConnectionAttempts}`));
        let attempts = 0;
        while (attempts < this.clientConfig.maxConnectionAttempts) {
            attempts++;
            await this.backoffWait(attempts);
            try {
                this.logger.warn(this.debugMsg(`initialising attempt #${attempts}`));
                await this.init();
                this.logger.warn(this.debugMsg(`attempt #${attempts} is successful.`));
                return;
            }
            catch (e) {
                this.logger.warn(e, this.debugMsg(`Failed reconnection attempt #${attempts}`));
            }
        }
        this.logger.error(this.debugMsg(`Couldn't reconnect after ${this.clientConfig.maxConnectionAttempts} attempts`));
        this.reconnecting = false;
        // tslint:disable-next-line
        if (this.clientConfig.exitOnIrrecoverableReconnect !== false) {
            this.logger.error(this.debugMsg('Cowardly refusing to continue. Calling shutdown function'));
            this.shutdownFunction();
        }
    }
    async close() {
        this.closed = true;
        await this.closeChannel();
        await this.closeConnection();
    }
    async closeChannel() {
        if (this.channel) {
            this.channel.removeAllListeners();
            await this.channel.close();
            this.channel = undefined;
        }
    }
    async closeConnection() {
        if (this.connection) {
            this.connection.removeAllListeners();
            await this.connection.close();
            this.connection = undefined;
        }
    }
    debugMsg(str) {
        return `${this.name}: ${str}`;
    }
}
exports.RabbitmqClient = RabbitmqClient;
//# sourceMappingURL=RabbitmqClient.js.map