package com.mvc.iot.onenet;

import com.alibaba.fastjson.JSONObject;
import com.mvc.iot.IotStoreBlockingQueue;
import com.mvc.iot.onenet.auth.AESBase64Utils;
import com.mvc.iot.onenet.auth.IoTConsumer;
import com.mvc.iot.onenet.auth.IoTMessage;
import com.mvc.iot.onenet.config.OneNetConfig;
import com.mvc.iot.onenet.model.OneNetRootModel;
import io.netty.util.internal.StringUtil;
import org.apache.pulsar.client.api.MessageId;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

@Component
public class IoTPulsarConsume implements Runnable {
    private static final Logger logger = LoggerFactory.getLogger(IoTPulsarConsume.class);

    @Resource
    private OneNetConfig oneNetConfig;

    @Override
    public void run() {
        if (StringUtil.isNullOrEmpty(oneNetConfig.getIotAccessId())) {
            logger.error("iotAccessId is null,please input iotAccessId");
            return;
        }
        if (StringUtil.isNullOrEmpty(oneNetConfig.getIotSecretKey())) {
            logger.error("iotSecretKey is null,please input iotSecretKey");
            return;
        }
        if (StringUtil.isNullOrEmpty(oneNetConfig.getIotSubscriptionName())) {
            logger.error("iotSubscriptionName is null,please input iotSubscriptionName");
            return;
        }
        try {
            //TODO 建议收到消息后将消息转到中间件后立即ACK。避免消息量过大导致消息过期
            IoTConsumer iotConsumer = IoTConsumer.IOTConsumerBuilder.anIOTConsumer().brokerServerUrl(OneNetConfig.brokerSSLServerUrl)
                    .iotAccessId(oneNetConfig.getIotAccessId())
                    .iotSecretKey(oneNetConfig.getIotSecretKey())
                    .subscriptionName(oneNetConfig.getIotSubscriptionName())
                    .iotMessageListener(message -> {
                        //MessageId msgId = message.getMessageId();
                        //long publishTime = message.getPublishTime();
                        String payload = new String(message.getData());
                        IoTMessage iotMessage = JSONObject.parseObject(payload, IoTMessage.class);
                        String originalMsg = AESBase64Utils.decrypt(iotMessage.getData(), oneNetConfig.getIotSecretKey().substring(8, 24));

                        /*logger.info("IOT consume message======>>>>>>> messageId={}, publishTime={},  payload={}",
                                msgId, publishTime, payload);*/
                        logger.info("[OneNet]originalMsg:{}", originalMsg);
                        OneNetRootModel rootModel = JSONObject.parseObject(originalMsg, OneNetRootModel.class);
                        if (rootModel == null || rootModel.getSubData() == null) {
                            return;
                        }
                        IotStoreBlockingQueue.oneDateCacheQueue.put(rootModel);

                    }).build();
            iotConsumer.run();
        } catch (Exception e) {
            logger.error("IoTConsumer:{}", e);
        }
    }
}
