package org.well.well.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.kafka.listener.MessageListener;
import org.well.well.base.AbstractResponse;
import org.well.well.util.ResponseResolver;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class KafkaConsumer implements MessageListener<Integer, String> {
    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaConsumer.class);

    private ClassPathXmlApplicationContext ac;
    private static ExecutorService threadPool = Executors.newCachedThreadPool();

    public void intContext() {
        ClassPathXmlApplicationContext cac = new ClassPathXmlApplicationContext(
                new String[]{"classpath:wellSensor/*.xml"});
        cac.start();
        ac = cac;
    }

    @Override
    public void onMessage(ConsumerRecord<Integer, String> record) {

        try {
            String value = record.value();

            LOGGER.info("######收到kafka消息：" + value);
            AbstractResponse resp = ResponseResolver.makeResponse(value);
            if (resp == null) return;
            if (ac == null) intContext();
            if (resp.getAc() == null) {
                resp.setAc(ac);
            }
            resp.setContent(value);
            threadPool.execute(resp);
//        resp.process(value);//对应的消息处理

            Thread.sleep(100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }
}
