inceptum/dist/rabbitmq/RabbitmqProducer.js

"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
const prom_client_1 = require("prom-client");
const LogManager_1 = require("../log/LogManager");
const RabbitmqConfig_1 = require("./RabbitmqConfig");
const RabbitmqClient_1 = require("./RabbitmqClient");
const logger = LogManager_1.LogManager.getLogger(__filename);
const rabbitmqPublishTime = new prom_client_1.Histogram({
    name: 'rabbitmq_publish_time',
    help: 'Time required to publish messages to RabbitMQ',
    labelNames: ['name'],
    buckets: [0.001, 0.003, 0.005, 0.01, 0.05, 0.1, 0.3]
});
const rabbitmqPublishErrorCounter = new prom_client_1.Counter({
    name: 'rabbitmq_publish_error_counter',
    help: 'Number of errors encountered when publishing messages',
    labelNames: ['name'],
});
class RabbitmqProducer extends RabbitmqClient_1.RabbitmqClient {
    constructor(clientConfig, name, producerConfig) {
        super(clientConfig, name);
        this.producerConfig = Object.assign({}, producerConfig);
        this.producerConfig.backPressureStrategy = producerConfig.backPressureStrategy || RabbitmqConfig_1.RabbitmqBackPressureStrategy.ERROR;
        this.logger = logger;
        this.publishDurationHistogram = rabbitmqPublishTime.labels(name);
        this.publishFailures = rabbitmqPublishErrorCounter.labels(name);
    }
    /**
     * publish msg with routing key
     * @param msg
     * @param routingKey
     */
    async publish(msg, routingKey, optionsPublish = {}) {
        if (this.closed) {
            throw new Error('Connection to RabbitMQ is closed, can\'t publish message');
        }
        if (!this.readyGate.isChannelReady()) {
            await this.readyGate.awaitChannelReady();
        }
        optionsPublish.headers = {
            retriesCount: 0,
        };
        const timer = this.publishDurationHistogram.startTimer();
        try {
            while (!this.channel.publish(this.producerConfig.exchangeName, routingKey, new Buffer(msg), optionsPublish)) {
                if (this.producerConfig.backPressureStrategy === RabbitmqConfig_1.RabbitmqBackPressureStrategy.ERROR) {
                    throw new Error('Failed to publish message');
                }
                if (optionsPublish.headers.retriesCount >= 3) {
                    throw new Error('Failed to publish message after 3 attempts');
                }
                this.publishFailures.inc();
                optionsPublish.headers.retriesCount++;
                this.logger.error(`publish failed. ====> ${msg}`);
                await new Promise((resolve, reject) => {
                    this.channel.once('drain', function () {
                        this.logger.info(`drain event received. ====> ${msg}`);
                        resolve();
                    });
                });
            }
        }
        finally {
            timer();
        }
        return true;
    }
    init() {
        return super.init();
    }
    close() {
        return super.close();
    }
}
exports.RabbitmqProducer = RabbitmqProducer;
//# sourceMappingURL=RabbitmqProducer.js.map