package com.hzjt.config;

import com.alibaba.fastjson.JSONObject;
import com.hzjt.domain.MqttResult;
import com.hzjt.service.TraffalarmrecordService;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import java.util.concurrent.ScheduledExecutorService;

@Configuration
public class PushCallback implements MqttCallback,IMQTTPublisher {
    private static final Logger log = LoggerFactory.getLogger(PushCallback.class);
    @Autowired
    private MQTTSubsribe mqttSubsribe;

    @Autowired
    private TraffalarmrecordService traffalarmrecordService;



    //    定时任务——定时缓存查询的数据
    private ScheduledExecutorService scheduled;

    @Override
    public void connectionLost(Throwable throwable) {
        // 连接丢失后，一般在这里面进行重连
        log.info("连接断开，可以做重连");
        this.mqttSubsribe.startReconnect();
    }


    @Override

    public void messageArrived(String topic, MqttMessage message){
        log.info("topic result"+topic+new String(message.getPayload()));
        try {
            MqttResult result = JSONObject.parseObject(message.getPayload(), MqttResult.class);
            //根据mqtt返回的处理结果进行处理,更新表中数据
            if(result.getCompanyId().startsWith("66211") && null!=result.getStatus())
            {
                traffalarmrecordService.updateState(result);
            }
        } catch (Exception ex){
            log.info("解析错误:{}",ex);
        }
        // subscribe后得到的消息会执行到这里面
//        String[] topics = topic.split("/");
//        this.scheduled = Executors.newSingleThreadScheduledExecutor();
//        //周期定时方法，可以在里面进行定时数据存储操作，我测试时是先将数据存储到了redis中，可做实时数据来用
//        this.scheduled.scheduleAtFixedRate(new Runnable() {
//            @Override
//            public void run() {
//                String key = topics[0] + topics[1];
//                String value = new String(message.getPayload());
//                log.info("topic result");
////                if (redisTemplate.hasKey(key)){
////                    redisTemplate.delete(key);
////                }
////                redisTemplate.opsForValue().set(key,value);
////                System.out.println("redis缓存数据"+value);
////                //下面是我对数据的一些处理，仅供参考
////                JSONObject jsonObject = JSONObject.parseObject(value);
////                Map map = jsonObject;
////                List<Map<String,Object>> list = (List<Map<String, Object>>) ((Map) map.get("lines")).get("data");
////                log.error("-----------------------------"+list);
//            }
//            //此处 120为每120秒执行一次
//        },0,120, TimeUnit.SECONDS);
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        log.info("deliveryComplete---------" + token.isComplete());
    }

    @Override
    public void publishMessage(String topic, String message) {

    }

    @Override
    public void disconnect() {
        log.info("disconnect---------" );
    }

}