//package com.yeejoin.equip.eqmx;
//
//import com.alibaba.fastjson.JSON;
//import com.yeejoin.equip.kafka.KafkaProducerService;
//import lombok.extern.slf4j.Slf4j;
//import net.sf.json.JSONObject;
//import org.eclipse.paho.client.mqttv3.MqttMessage;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.beans.factory.annotation.Value;
//import org.springframework.stereotype.Component;
//import org.typroject.tyboot.component.emq.EmqKeeper;
//import org.typroject.tyboot.component.emq.EmqxListener;
//import javax.annotation.PostConstruct;
//import java.util.concurrent.*;
//
///**
// * @author LiuLin
// * @date 2023/6/25
// * @apiNote Emq消息转发Kafka
// */
//@Slf4j
//@Component
//public class EmqMessageService extends EmqxListener {
//
//    @Autowired
//    protected EmqKeeper emqKeeper;
//
//    @Autowired
//    protected KafkaProducerService kafkaProducerService;
//
//    @Value("${emq.topic}")
//    private String emqTopic;
//
//    @Value("${kafka.topic}")
//    private String kafkaTopic;
//    ExecutorService service = Executors.newFixedThreadPool(10);
//    private static final BlockingQueue<JSONObject> blockingQueue = new LinkedBlockingQueue<>();
//
//    @PostConstruct
//    void init() throws Exception {
//        service.execute(new NumberThread());
//        emqKeeper.subscript(emqTopic, 0, this);
//       // emqKeeper.subscript(emqTopic, 0, this);
//    }
//
//    @Override
//    public void processMessage(String topic, MqttMessage message) throws Exception {
//        JSONObject result = JSONObject.fromObject(new String(message.getPayload()));
//        JSONObject messageResult = new JSONObject();
//        messageResult.put("result", result);
//        messageResult.put("topic", topic);
//        blockingQueue.add(messageResult);
//    }
//    class NumberThread implements Runnable {
//        @Override
//        public void run() {
//            while (true) {
//                try {
//                    JSONObject messageResult = blockingQueue.take();
//                    JSONObject result = messageResult.getJSONObject("result");
//                    if ((messageResult.getString("topic")).equals(emqTopic)) {
//                        String dataType = result.getString("dataType");
//                        String address = result.getString("address");
//                        String gatewayId = result.getString("gatewayId");
//                        String value = result.getString("value");
//                        String signalType = result.getString("signalType");
//                        log.info("===========接收MQTT订阅消息，address:{},gatewayId:{},dateType:{},value:{},signalType:{}", address,gatewayId,dataType,value,signalType);
//                        //log.info("===========接收IOT订阅消息，address:{},gatewayId:{},dateType:{},value:{},signalType:{}", address,gatewayId,dataType,value,signalType);
//                        kafkaProducerService.sendMessageAsync(kafkaTopic,JSON.toJSONString(result));
//                    }
//                } catch (Exception e) {
//                    Thread.currentThread().interrupt();
//                }
//            }
//        }
//    }
//}
