inceptum/dist/rabbitmq/RabbitmqConsumer.js

"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
const prom_client_1 = require("prom-client");
const NewrelicUtil_1 = require("../newrelic/NewrelicUtil");
const LogManager_1 = require("../log/LogManager");
const RabbitmqClient_1 = require("./RabbitmqClient");
const RabbitmqConsumerHandler_1 = require("./RabbitmqConsumerHandler");
const RabbitmqConsumerHandlerError_1 = require("./RabbitmqConsumerHandlerError");
const logger = LogManager_1.LogManager.getLogger(__filename);
const newrelic = NewrelicUtil_1.NewrelicUtil.getNewrelicIfAvailable();
class NewrelichandlerWrapper extends RabbitmqConsumerHandler_1.RabbitmqConsumerHandler {
    constructor(baseHandler) {
        super();
        this.baseHandler = baseHandler;
    }
    async handle(message) {
        await newrelic.startBackgroundTransaction(this.baseHandler.constructor.name, 'RabbitMQConsumer', async () => {
            const transaction = newrelic.getTransaction();
            try {
                await this.baseHandler.handle(message);
            }
            finally {
                transaction.end();
            }
        });
    }
}
const rabbitmqConsumeTime = new prom_client_1.Histogram({
    name: 'rabbitmq_consume_time',
    help: 'Time required to consumer a messages from RabbitMQ',
    labelNames: ['name'],
    buckets: [0.003, 0.01, 0.05, 0.1, 0.3, 0.5, 1, 2, 5]
});
const rabbitmqConsumeErrorCounter = new prom_client_1.Counter({
    name: 'rabbitmq_consume_error_counter',
    help: 'Number of errors encountered when consuming messages',
    labelNames: ['name'],
});
const rabbitmqConsumeDLQCounter = new prom_client_1.Counter({
    name: 'rabbitmq_consume_to_dlq_counter',
    help: 'Number of messages sent to the DLQ',
    labelNames: ['name'],
});
const rabbitmqConsumeDelayedCounter = new prom_client_1.Counter({
    name: 'rabbitmq_consume_delayed_counter',
    help: 'Number of messages sent to the delay queue',
    labelNames: ['name'],
});
class RabbitmqConsumer extends RabbitmqClient_1.RabbitmqClient {
    constructor(clientConfig, name, consumerConfig, handler) {
        super(clientConfig, name);
        this.messageHandler = newrelic ? new NewrelichandlerWrapper(handler) : handler;
        this.consumerConfig = Object.assign({}, consumerConfig);
        this.consumerConfig.options = this.consumerConfig.options || {};
        this.logger = logger;
        this.consumeDurationHistogram = rabbitmqConsumeTime.labels(name);
        this.consumeFailures = rabbitmqConsumeErrorCounter.labels(name);
        this.consumeFailuresDLQ = rabbitmqConsumeDLQCounter.labels(name);
        this.consumeFailuresDelayed = rabbitmqConsumeDelayedCounter.labels(name);
    }
    async init() {
        try {
            await super.init();
            await this.subscribe(this.consumerConfig.appQueueName, this.consumerConfig);
        }
        catch (e) {
            const c = Object.assign({}, this.consumerConfig);
            this.logger.error(e, `failed to subscribe with config - ${c.toString()}`);
            NewrelicUtil_1.NewrelicUtil.noticeError(e, { config: c });
            throw e;
        }
    }
    /**
     * Subscribe to a queue
     */
    async subscribe(queueName, consumerConfig) {
        if (consumerConfig.prefetch && consumerConfig.prefetch > 0) {
            this.channel.prefetch(consumerConfig.prefetch);
        }
        return this.channel.consume(queueName, (message) => {
            this.handleMessage(message);
        }, consumerConfig.options);
    }
    async handleMessage(message) {
        const timer = this.consumeDurationHistogram.startTimer();
        try {
            await this.messageHandler.handle(message);
            this.channel.ack(message);
        }
        catch (e) {
            this.logger.error(e, 'failed to handle message');
            this.consumeFailures.inc();
            NewrelicUtil_1.NewrelicUtil.noticeError(e, message);
            if (message.properties.headers.retriesCount === undefined) {
                message.properties.headers.retriesCount = 0;
            }
            const retriesCount = ++message.properties.headers.retriesCount;
            if (e instanceof RabbitmqConsumerHandlerError_1.RabbitmqConsumerHandlerUnrecoverableError || !this.allowRetry(retriesCount)) {
                // add to dlq
                this.consumeFailuresDLQ.inc();
                try {
                    this.sendMessageToDlq(message);
                    this.channel.ack(message);
                }
                catch (err) {
                    this.logger.error(err, 'failed to send message to dlq');
                    NewrelicUtil_1.NewrelicUtil.noticeError(err, message);
                    this.channel.nack(message);
                }
            }
            else {
                this.consumeFailuresDelayed.inc();
                try {
                    this.sendMessageToDelayedQueue(message, retriesCount, e);
                    this.channel.ack(message);
                }
                catch (error) {
                    // put message back to rabbitmq
                    this.logger.error(error, 'failed to send message to delayed queue');
                    NewrelicUtil_1.NewrelicUtil.noticeError(error, message);
                    this.channel.nack(message);
                }
            }
        }
        finally {
            timer();
        }
    }
    sendMessageToDlq(message) {
        this.channel.sendToQueue(this.consumerConfig.dlqName, message.content);
        this.logger.info(this.stringyifyMessageContent(message), 'sent message to dlq');
    }
    sendMessageToDelayedQueue(message, retriesCount, e) {
        const ct = this.stringyifyMessageContent(message);
        // depending on retries config, retry
        const ttl = this.getTtl(retriesCount);
        const options = {
            expiration: ttl,
            headers: message.properties.headers,
        };
        this.channel.sendToQueue(this.consumerConfig.delayQueueName, message.content, options);
        const data = {
            queueName: this.consumerConfig.delayQueueName,
            messageContent: ct,
            options,
        };
        this.logger.info(data, `sent message to delayed queue`);
    }
    stringyifyMessageContent(message) {
        return message.content.toString();
    }
    /**
     *
     * @param retriesCount number
     * @reutrn number in milliseconds
     */
    getTtl(retriesCount = 1) {
        if (this.allowRetry(retriesCount)) {
            return Math.pow(retriesCount, this.consumerConfig.retryDelayFactor)
                * this.consumerConfig.retryDelayInMinute * 60 * 1000;
        }
        return 0;
    }
    allowRetry(retriesCount) {
        return retriesCount && this.consumerConfig.maxRetries >= retriesCount;
    }
    async close() {
        await super.close();
    }
}
exports.RabbitmqConsumer = RabbitmqConsumer;
//# sourceMappingURL=RabbitmqConsumer.js.map