package xyz.zhangyi.ddd.core.gateway.south.adapter;

import com.alibaba.fastjson.JSON;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import xyz.zhangyi.ddd.core.gateway.south.port.Destination;
import xyz.zhangyi.ddd.core.gateway.south.port.EventPublisher;
import xyz.zhangyi.ddd.core.event.Event;

import java.util.Properties;

public class EventKafkaPublisher<T> implements EventPublisher {
    private Destination destination;
    private Producer<String, String> producer;

    public EventKafkaPublisher(Destination destination) {
        this.destination = destination;
    }

    @Override
    public void publish(Event event) {
        producer = createProducer();
        ProducerRecord<String, String> producerRecord = new ProducerRecord<>(destination.topic(), JSON.toJSONString(event));
        producer.send(producerRecord);
    }

    protected Producer<String, String> createProducer() {
        Properties props = buildProperties(destination);
        return new KafkaProducer<>(props);
    }

    private Properties buildProperties(Destination destination) {
        Properties props = new Properties();
        props.put("bootstrap.servers", destination.server());
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        return props;
    }
}