package com.yeejoin.amos.boot.module.jxiop.biz.emqx;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.toolkit.ObjectUtils;
import com.yeejoin.amos.boot.module.jxiop.api.entity.PersonBasic;
import com.yeejoin.amos.boot.module.jxiop.api.mapper.PersonBasicMapper;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Autowired;
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.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

@Component
@Slf4j
public class PersonYardMessage extends EmqxListener {

    @Autowired
    protected EmqKeeper emqKeeper;

    @Autowired
    private PersonBasicMapper personBasicMapper;


    // 江西电建接收红黄绿码主题
    private static final String RYFM_WARNING_CODING = "ryfm/warning/coding";

    private static final BlockingQueue<JSONObject> blockingQueue = new LinkedBlockingQueue<JSONObject>();

    @PostConstruct
    void init() throws Exception {
        new Thread(taskRunnable).start();
        emqKeeper.subscript(RYFM_WARNING_CODING, 2, this);
    }

    @Override
    public void processMessage(String topic, MqttMessage message) throws Exception {
        log.info("人员赋码消息{}", new String(message.getPayload()));
        JSONObject ja = JSON.parseObject(new String(message.getPayload()));
        blockingQueue.add(ja);
    }

    Runnable taskRunnable = new Runnable() {
        @Override
        public void run() {
            boolean isRun = true;
            int k = 0;
            while (isRun) {
                k++;
                isRun = k < Integer.MAX_VALUE;
                try {
                    JSONObject analysisResult = blockingQueue.take();
//                    RequestContext.setAppKey(amosAuth.getAppKey());
//                    RequestContext.setProduct(amosAuth.getProduct());
//                    RequestContext.setToken(amosAuth.getToken());
                    jxIopUpdate(analysisResult);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    };

    public void jxIopUpdate(JSONObject analysisResult) {
        log.info("人员赋码数据：{}", analysisResult);
        String warningObjectCode = String.valueOf(analysisResult.get("warningObjectCode"));
        String codeColor = String.valueOf(analysisResult.get("codeColor"));
        PersonBasic personBasic = personBasicMapper.selectById(warningObjectCode);
        if (ObjectUtils.isEmpty(personBasic)) {
            try {
                emqKeeper.getMqttClient().publish("ryfm/warning/coding", analysisResult.toJSONString().getBytes(), 2, Boolean.FALSE);
            } catch (MqttException e) {
                e.printStackTrace();
            }
        } else {
            personBasic.setQrcodeColor(codeColor);
            personBasicMapper.updateById(personBasic);
        }
    }
}
