//
// Source code recreated from a .class file by IntelliJ IDEA
// (powered by FernFlower decompiler)
//

package com.base.orbit.mq.kafka;

import com.base.orbit.mq.MqServer;
import com.base.orbit.mq.handler.MessageHandler;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.*;

public class KafkaMqServer implements MqServer {
    private static final Logger logger = LoggerFactory.getLogger(KafkaMqServer.class);
    private KafkaConsumer<String, String> consumer;
    private MessageHandler handler;
    private String topic;

    public KafkaMqServer(KafkaConsumer<String, String> consumer, MessageHandler handler, String topic) {
        this.consumer = consumer;
        this.handler = handler;
        this.topic = topic;
    }

    public void start() {
        this.subscribe();
        ThreadFactory threadFactory = (r) -> new Thread(r, "kafka-consumer-thread");
        ExecutorService executorService = new ThreadPoolExecutor(1, 1, Long.MAX_VALUE, TimeUnit.SECONDS, new LinkedBlockingQueue(), threadFactory);
        executorService.submit(this::startProcess);
    }

    private void startProcess() {
        while(!Thread.currentThread().isInterrupted()) {
            try {
                //  poll(time)，如果拉到数据则立即返回，拉不到数据，time为最长的等待时间
                ConsumerRecords<String, String> records = this.consumer.poll(Long.MAX_VALUE);
                List<String> list = new ArrayList<>();
                records.forEach((record) -> list.add(record.value()));
                this.handler.process(list);
            } catch (Exception var3) {
                logger.error("kafka consume fail ", var3);
            }
        }

    }

    private void subscribe() {
        Objects.requireNonNull(this.topic, "subscribe topic cannot be null");
        this.consumer.subscribe(Arrays.asList(this.topic.split(",")));
    }

    public KafkaConsumer<String, String> getConsumer() {
        return this.consumer;
    }

    public String getTopic() {
        return this.topic;
    }
}
