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

import com.base.orbit.mq.MqClient;
import com.base.orbit.mq.kafka.KafkaMqClient;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.common.serialization.StringSerializer;

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

public class KafkaMqClientBuilder {
    private String brokers;
    private String acks = "all";

    public KafkaMqClientBuilder() {
    }

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

    public KafkaMqClientBuilder withAcks(String acks) {
        this.acks = acks;
        return this;
    }

    public MqClient build() {
        Objects.requireNonNull(this.brokers, "kafka broker server cannot be null");
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", this.brokers);
        properties.setProperty("acks", this.acks);
        properties.setProperty("key.serializer", StringSerializer.class.getName());
        properties.setProperty("value.serializer", StringSerializer.class.getName());
        KafkaProducer<String, String> producer = new KafkaProducer(properties);
        return new KafkaMqClient(producer);
    }
}
