@rxdi/graphql-rabbitmq-subscriptions/dist/amqp-pubsub.js

"use strict";
var __awaiter = (this && this.__awaiter) || function (thisArg, _arguments, P, generator) {
    function adopt(value) { return value instanceof P ? value : new P(function (resolve) { resolve(value); }); }
    return new (P || (P = Promise))(function (resolve, reject) {
        function fulfilled(value) { try { step(generator.next(value)); } catch (e) { reject(e); } }
        function rejected(value) { try { step(generator["throw"](value)); } catch (e) { reject(e); } }
        function step(result) { result.done ? resolve(result.value) : adopt(result.value).then(fulfilled, rejected); }
        step((generator = generator.apply(thisArg, _arguments || [])).next());
    });
};
Object.defineProperty(exports, "__esModule", { value: true });
const pubsub_async_iterator_1 = require("./pubsub-async-iterator");
const rabbitmq_pub_sub_1 = require("rabbitmq-pub-sub");
const async_1 = require("async");
const child_logger_1 = require("./child-logger");
class AmqpPubSub {
    constructor(options = {}) {
        this.triggerTransform =
            options.triggerTransform || (trigger => trigger);
        const config = options.config || { host: '127.0.0.1', port: 5672 };
        const { logger } = options;
        this.logger = child_logger_1.createChildLogger(logger, 'AmqpPubSub');
        const factory = new rabbitmq_pub_sub_1.RabbitMqSingletonConnectionFactory(logger, config);
        this.consumer = new rabbitmq_pub_sub_1.RabbitMqSubscriber(logger, factory);
        this.producer = new rabbitmq_pub_sub_1.RabbitMqPublisher(logger, factory);
        this.subscriptionMap = {};
        this.subsRefsMap = {};
        this.currentSubscriptionId = 0;
    }
    publish(trigger, payload) {
        return __awaiter(this, void 0, void 0, function* () {
            this.logger.trace("publishing for queue '%s' (%j)", trigger, payload);
            this.producer.publish(trigger, payload);
        });
    }
    subscribe(trigger, onMessage, options) {
        const triggerName = this.triggerTransform(trigger, options);
        const id = this.currentSubscriptionId++;
        this.subscriptionMap[id] = [triggerName, onMessage];
        let refs = this.subsRefsMap[triggerName];
        if (refs && refs.length > 0) {
            const newRefs = [...refs, id];
            this.subsRefsMap[triggerName] = newRefs;
            this.logger.trace("subscriber exist, adding triggerName '%s' to saved list.", triggerName);
            return Promise.resolve(id);
        }
        else {
            return new Promise((resolve, reject) => {
                this.logger.trace("trying to subscribe to queue '%s'", triggerName);
                this.consumer
                    .subscribe(triggerName, msg => this.onMessage(triggerName, msg))
                    .then(disposer => {
                    this.subsRefsMap[triggerName] = [
                        ...(this.subsRefsMap[triggerName] || []),
                        id,
                    ];
                    this.unsubscribeChannel = disposer;
                    return resolve(id);
                })
                    .catch(err => {
                    this.logger.error(err, "failed to recieve message from queue '%s'", triggerName);
                    reject(id);
                });
            });
        }
    }
    unsubscribe(subId) {
        const [triggerName = null] = this.subscriptionMap[subId] || [];
        const refs = this.subsRefsMap[triggerName];
        if (!refs) {
            this.logger.error("There is no subscription of id '%s'", subId);
            throw new Error(`There is no subscription of id "{subId}"`);
        }
        let newRefs;
        if (refs.length === 1) {
            newRefs = [];
            this.unsubscribeChannel()
                .then(() => {
                this.logger.trace("cancelled channel from subscribing to queue '%s'", triggerName);
            })
                .catch(err => {
                this.logger.error(err, "channel cancellation failed from queue '%j'", triggerName);
            });
        }
        else {
            const index = refs.indexOf(subId);
            if (index !== -1) {
                newRefs = [...refs.slice(0, index), ...refs.slice(index + 1)];
            }
            this.logger.trace("removing triggerName from listening '%s' ", triggerName);
        }
        this.subsRefsMap[triggerName] = newRefs;
        delete this.subscriptionMap[subId];
        this.logger.trace("list of subscriptions still available '(%j)'", this.subscriptionMap);
    }
    asyncIterator(triggers) {
        return new pubsub_async_iterator_1.PubSubAsyncIterator(this, triggers);
    }
    onMessage(channel, message) {
        const subscribers = this.subsRefsMap[channel];
        // Don't work for nothing..
        if (!subscribers || !subscribers.length) {
            return;
        }
        this.logger.trace("sending message to subscriber callback function '(%j)'", message);
        async_1.each(subscribers, (subId, cb) => {
            // TODO Support pattern based subscriptions
            const [triggerName, listener] = this.subscriptionMap[subId];
            listener(message);
            cb();
        });
    }
}
exports.AmqpPubSub = AmqpPubSub;