package com.base.orbit.mq.kafka.builder;

import com.base.orbit.mq.MqServer;
import com.base.orbit.mq.handler.MessageHandler;
import com.base.orbit.mq.kafka.KafkaMqServer;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Objects;
import java.util.Properties;

public class KafkaMqServerBuilder {
    private String topic;
    private String brokers;
    private int consumeSize = Integer.MAX_VALUE;
    private String group;
    private MessageHandler handler;
    private String autoOffsetResetConfig = "earliest";
    private Boolean enableAutoCommitConfig = true;
    private int autoCommitIntervalMsConfig = 1000;

    public KafkaMqServerBuilder() {
    }

    public KafkaMqServerBuilder withTopic(String topic) {
        this.topic = topic;
        return this;
    }

    public KafkaMqServerBuilder withBrokers(String brokers) {
        this.brokers = brokers;
        return this;
    }

    public KafkaMqServerBuilder withGroup(String group) {
        this.group = group;
        return this;
    }

    public KafkaMqServerBuilder withConsumeSize(int consumeSize) {
        this.consumeSize = consumeSize;
        return this;
    }

    public KafkaMqServerBuilder withHandler(MessageHandler handler) {
        this.handler = handler;
        return this;
    }

    public KafkaMqServerBuilder withAutoOffsetResetConfig(String autoOffsetResetConfig) {
        this.autoOffsetResetConfig = autoOffsetResetConfig;
        return this;
    }

    public KafkaMqServerBuilder withEnableAutoCommitConfig(Boolean enableAutoCommitConfig) {
        this.enableAutoCommitConfig = enableAutoCommitConfig;
        return this;
    }

    public KafkaMqServerBuilder withAutoCommitIntervalMsConfig(int autoCommitIntervalMsConfig) {
        this.autoCommitIntervalMsConfig = autoCommitIntervalMsConfig;
        return this;
    }

    public MqServer build() {
        Objects.requireNonNull(this.brokers, "kafka brokers cannot be null");
        Objects.requireNonNull(this.topic, "kafka topic cannot be null");
        Objects.requireNonNull(this.group, "kafka group cannot be null");
        Objects.requireNonNull(this.handler, "kafka data handler cannot be null");
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", this.brokers);
        properties.setProperty("group.id", this.group);
        properties.setProperty("max.poll.records", String.valueOf(this.consumeSize));
        properties.setProperty("key.deserializer", StringDeserializer.class.getName());
        properties.setProperty("value.deserializer", StringDeserializer.class.getName());
        properties.setProperty("enable.auto.commit", String.valueOf(this.enableAutoCommitConfig));
        properties.setProperty("auto.commit.interval.ms", String.valueOf(this.autoCommitIntervalMsConfig));
        properties.setProperty("auto.offset.reset", String.valueOf(this.autoOffsetResetConfig));
        KafkaConsumer<String, String> consumer = new KafkaConsumer(properties);
        return new KafkaMqServer(consumer, this.handler, this.topic);
    }
}
