'use strict';

var kafkajs = require('kafkajs');

const DEFAULT_CLIENT = 'kafkajs';

class KafkaContext {
    #pause;
    #heartbeat;
    constructor({ pause, heartbeat }) {
        this.#pause = pause;
        this.#heartbeat = heartbeat;
    }
    pause() {
        return this.#pause();
    }
    heartbeat() {
        return this.#heartbeat();
    }
}
class KafkaBatchMessageContext extends KafkaContext {
    #batch;
    #isStale;
    #isRunning;
    #resolveOffset;
    #uncommittedOffsets;
    #commitOffsetsIfNecessary;
    constructor({ pause, batch, isStale, heartbeat, isRunning, resolveOffset, uncommittedOffsets, commitOffsetsIfNecessary, }) {
        super({
            pause,
            heartbeat,
        });
        this.#batch = batch;
        this.#isStale = isStale;
        this.#isRunning = isRunning;
        this.#resolveOffset = resolveOffset;
        this.#uncommittedOffsets = uncommittedOffsets;
        this.#commitOffsetsIfNecessary = commitOffsetsIfNecessary;
    }
    get batch() {
        return this.#batch;
    }
    resolveOffset(offset) {
        return this.#resolveOffset(offset);
    }
    commitOffsetsIfNecessary(offsets) {
        return this.#commitOffsetsIfNecessary(offsets);
    }
    uncommittedOffsets() {
        return this.#uncommittedOffsets();
    }
    get isRunning() {
        return this.#isRunning();
    }
    get isStale() {
        return this.#isStale();
    }
}
class KafkaMessageContext extends KafkaContext {
    #topic;
    #partition;
    #message;
    constructor({ topic, pause, message, heartbeat, partition, }) {
        super({
            pause,
            heartbeat,
        });
        this.#topic = topic;
        this.#message = message;
        this.#partition = partition;
    }
    get topic() {
        return this.#topic;
    }
    get message() {
        return this.#message;
    }
    get partition() {
        return this.#partition;
    }
}

function parseBrokers(brokers) {
    return brokers.map((broker) => {
        if (typeof broker === 'string') {
            return broker;
        }
        return [broker.domain, (broker.port || '9092').toString()].join(':');
    });
}
function parseMessage(message) {
    const { value } = message;
    let rawMessage = value?.toString() || '';
    try {
        rawMessage = JSON.parse(rawMessage);
    }
    catch {
        // do nothing
    }
    return rawMessage;
}

async function createKafka({ brokers, ...rest }) {
    let list;
    if (typeof brokers === 'function') {
        list = await brokers();
    }
    else {
        list = brokers;
    }
    const kafka = new kafkajs.Kafka({
        brokers: parseBrokers(list),
        ...rest,
        clientId: rest.clientId,
    });
    return kafka;
}
async function createKafkaClients(clientOptions) {
    if (!Array.isArray(clientOptions)) {
        clientOptions = [clientOptions];
    }
    const kafkas = await Promise.all(clientOptions.map(async (clientOption) => {
        const { clientId = DEFAULT_CLIENT } = clientOption;
        const kafka = await createKafka(clientOption);
        return {
            kafka,
            clientId,
        };
    }));
    const kafkaClients = kafkas.reduce((result, kafkaClient) => {
        const { clientId, kafka } = kafkaClient;
        result[clientId] = kafka;
        return result;
    }, {});
    const getClient = (clientId = DEFAULT_CLIENT) => {
        return kafkaClients[clientId];
    };
    const getAllClients = () => {
        return kafkaClients;
    };
    return {
        getClient,
        getAllClients,
        client: getClient(DEFAULT_CLIENT) || kafkas[0].kafka,
    };
}

Object.defineProperty(exports, 'Partitioners', {
    enumerable: true,
    get: function () { return kafkajs.Partitioners; }
});
exports.DEFAULT_CLIENT = DEFAULT_CLIENT;
exports.KafkaBatchMessageContext = KafkaBatchMessageContext;
exports.KafkaMessageContext = KafkaMessageContext;
exports.createKafkaClients = createKafkaClients;
exports.parseMessage = parseMessage;