"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
const constants_1 = require("../constants");
const server_1 = require("./server");
let mqttPackage = {};
class ServerMqtt extends server_1.Server {
    constructor(options) {
        super();
        this.options = options;
        this.url =
            this.getOptionsProp(options, 'url') || constants_1.MQTT_DEFAULT_URL;
        mqttPackage = this.loadPackage('mqtt', ServerMqtt.name);
    }
    async listen(callback) {
        this.mqttClient = this.createMqttClient();
        this.start(callback);
    }
    start(callback) {
        this.handleError(this.mqttClient);
        this.bindEvents(this.mqttClient);
        this.mqttClient.on(constants_1.CONNECT_EVENT, callback);
    }
    bindEvents(mqttClient) {
        mqttClient.on(constants_1.MESSAGE_EVENT, this.getMessageHandler(mqttClient).bind(this));
        const registeredPatterns = Object.keys(this.messageHandlers);
        registeredPatterns.forEach(pattern => mqttClient.subscribe(this.getAckQueueName(pattern)));
    }
    close() {
        this.mqttClient && this.mqttClient.end();
    }
    createMqttClient() {
        return mqttPackage.connect(this.url, this.options.options);
    }
    getMessageHandler(pub) {
        return async (channel, buffer) => await this.handleMessage(channel, buffer, pub);
    }
    async handleMessage(channel, buffer, pub) {
        const packet = this.deserialize(buffer.toString());
        const pattern = channel.replace(/_ack$/, '');
        const publish = this.getPublisher(pub, pattern, packet.id);
        const status = 'error';
        if (!this.messageHandlers[pattern]) {
            return publish({ id: packet.id, status, err: constants_1.NO_PATTERN_MESSAGE });
        }
        const handler = this.messageHandlers[pattern];
        const response$ = this.transformToObservable(await handler(packet.data));
        response$ && this.send(response$, publish);
    }
    getPublisher(client, pattern, id) {
        return response => client.publish(this.getResQueueName(pattern), JSON.stringify(Object.assign(response, { id })));
    }
    deserialize(content) {
        try {
            return JSON.parse(content);
        }
        catch (e) {
            return content;
        }
    }
    getAckQueueName(pattern) {
        return `${pattern}_ack`;
    }
    getResQueueName(pattern) {
        return `${pattern}_res`;
    }
    handleError(stream) {
        stream.on(constants_1.ERROR_EVENT, err => this.logger.error(err));
    }
}
exports.ServerMqtt = ServerMqtt;
