// Path: jbase-spring/src/main/java/com/jayqqaa12/jbase/spring/mqtt/protocol/MqttReq.java
// @Data
// public class MqttReq<T> {
// 
//     /**
//      * 请求序列 客户端传递用来标示 返回的消息是哪次请求
//      */
//     private String reqNo;
// 
//     /**
//      * 时间戳
//      */
//     private Long timestamp;
//     
//     /**
//      * 当前接口的版本号
//      */
//     private String  version;
// 
//     /**
//      * 平台 device android ios
//      */
//     private String  platform;
// 
//     /**
//      * 设备唯一标示
//      */
//     private String uniqueCode;
// 
// 
//     private T data;
// 
//     
// 
// }


import com.alibaba.fastjson.JSON;
import com.jayqqaa12.jbase.spring.mqtt.protocol.MqttReq;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.MessageHandler;
import org.springframework.stereotype.Component;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.nio.charset.Charset;
import java.util.List;
package com.jayqqaa12.jbase.spring.mqtt.handler;



@Slf4j
@Component
public class MqttReceiver {

    @Autowired(required = false)
    List<MqttHandler> handlerList;


    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler handler() {
        return (message) -> {
            String topic = (String) message.getHeaders().get(MqttHeaders.TOPIC);

            byte[] payload = (byte[]) message.getPayload();

            log.info("receiver byte len {}", payload.length);

            try {
                if (handlerList == null) return;

                for (MqttHandler mqttHandler : handlerList) {
                    TopicMapping topicMapping = mqttHandler.getClass().getAnnotation(TopicMapping.class);
                    if (topicMapping == null) throw new RuntimeException("MqttHandler must use @TopicMapping ");

                    Type type = mqttHandler.getClass().getGenericInterfaces()[0];

                    ParameterizedType parameterizedType = (ParameterizedType) type;
                    Type[] types = parameterizedType.getActualTypeArguments();

                    Type rawType = types[0];
                    if (rawType instanceof ParameterizedType) {
                        rawType = ((ParameterizedType) rawType).getRawType();
                    }
                    Class clazz = (Class) rawType;

                    // 这里要判断一下 topic
                    if (isContainsTopic(topic, topicMapping.value())) {
                        Object data = payload;
                        if (clazz.isAssignableFrom(String.class))
                            data = new String(payload, Charset.defaultCharset());
// Path: jbase-spring/src/main/java/com/jayqqaa12/jbase/spring/mqtt/protocol/MqttReq.java
// @Data
// public class MqttReq<T> {
// 
//     /**
//      * 请求序列 客户端传递用来标示 返回的消息是哪次请求
//      */
//     private String reqNo;
// 
//     /**
//      * 时间戳
//      */
//     private Long timestamp;
//     
//     /**
//      * 当前接口的版本号
//      */
//     private String  version;
// 
//     /**
//      * 平台 device android ios
//      */
//     private String  platform;
// 
//     /**
//      * 设备唯一标示
//      */
//     private String uniqueCode;
// 
// 
//     private T data;
// 
//     
// 
// }


// Path: jbase-spring/src/main/java/com/jayqqaa12/jbase/spring/mqtt/handler/MqttReceiver.java
import com.alibaba.fastjson.JSON;
import com.jayqqaa12.jbase.spring.mqtt.protocol.MqttReq;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.MessageHandler;
import org.springframework.stereotype.Component;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.nio.charset.Charset;
import java.util.List;

package com.jayqqaa12.jbase.spring.mqtt.handler;



@Slf4j
@Component
public class MqttReceiver {

    @Autowired(required = false)
    List<MqttHandler> handlerList;


    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler handler() {
        return (message) -> {
            String topic = (String) message.getHeaders().get(MqttHeaders.TOPIC);

            byte[] payload = (byte[]) message.getPayload();

            log.info("receiver byte len {}", payload.length);

            try {
                if (handlerList == null) return;

                for (MqttHandler mqttHandler : handlerList) {
                    TopicMapping topicMapping = mqttHandler.getClass().getAnnotation(TopicMapping.class);
                    if (topicMapping == null) throw new RuntimeException("MqttHandler must use @TopicMapping ");

                    Type type = mqttHandler.getClass().getGenericInterfaces()[0];

                    ParameterizedType parameterizedType = (ParameterizedType) type;
                    Type[] types = parameterizedType.getActualTypeArguments();

                    Type rawType = types[0];
                    if (rawType instanceof ParameterizedType) {
                        rawType = ((ParameterizedType) rawType).getRawType();
                    }
                    Class clazz = (Class) rawType;

                    // 这里要判断一下 topic
                    if (isContainsTopic(topic, topicMapping.value())) {
                        Object data = payload;
                        if (clazz.isAssignableFrom(String.class))
                            data = new String(payload, Charset.defaultCharset());
                        else if (clazz.isAssignableFrom(MqttReq.class)) {