新增mqtt组件

Author: MentosL

arex-instrumentation/mq/arex-integration-mqtt/src/main/java/io/arex/inst/mqtt/MessageQueueExtractor.java

@@ -51,6 +51,8 @@ private void executePostProcess() {
         if (ContextManager.needReplay()) {
             adapter.addHeader(messageChannel,message, ArexConstants.REPLAY_ID,ContextManager.currentContext().getReplayId());
         }
+        // Think about other ways to replace the head
+        adapter.resetMsg(message);
     }
 
     private void doExecute() {

arex-instrumentation/mq/arex-integration-mqtt/src/main/java/io/arex/inst/mqtt/adapter/MessageAdapter.java

@@ -23,4 +23,6 @@ public interface MessageAdapter<MC,Msg> {
 
     void addHeader(MC mc,Msg msg,String key,String value);
 
+    Msg resetMsg(Msg msg);
+
 }

arex-instrumentation/mq/arex-integration-mqtt/src/main/java/io/arex/inst/mqtt/adapter/MessageAdapterImpl.java

@@ -2,11 +2,13 @@
 
 import io.arex.inst.mqtt.warp.GenericMessageWarp;
 import io.arex.inst.mqtt.warp.MessageHeaderWarp;
+import io.arex.inst.runtime.util.LogUtil;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.support.GenericMessage;
 
+import java.lang.reflect.Field;
 import java.nio.charset.StandardCharsets;
 
 /**
@@ -34,7 +36,10 @@ public MessageChannel warpMC(Object messageChannel) {
         if (messageChannel == null){
             return null;
         }
-        return (MessageChannel) messageChannel;
+        if (messageChannel instanceof MessageChannel){
+            return (MessageChannel) messageChannel;
+        }
+        return null;
     }
 
     @Override
@@ -57,6 +62,10 @@ public Message warpMessage(Object message) {
 
     @Override
     public MessageHeaders getHeader(MessageChannel messageChannel, Message msg) {
+        if (msg instanceof  GenericMessageWarp){
+            GenericMessageWarp messageTemp = (GenericMessageWarp) msg;
+            return messageTemp.getMessageHeaderWarp();
+        }
         return msg.getHeaders();
     }
 
@@ -74,6 +83,12 @@ public boolean markProcessed(Message message, String flagKey) {
 
     @Override
     public String getHeader(MessageChannel messageChannel, Message message, String key) {
+        if (message instanceof GenericMessageWarp) {
+            GenericMessageWarp genericMessageWarp = (GenericMessageWarp) message;
+            Object object = genericMessageWarp.get(key);
+            return object != null ? object.toString() : null;
+        }
+
         if(message instanceof  GenericMessage){
             Object obj = message.getHeaders().get(key);
             return obj != null ? obj.toString() : null;
@@ -96,4 +111,23 @@ public void addHeader(MessageChannel messageChannel, Message message, String key
             genericMessageWarp.put(key,value);
         }
     }
+
+    @Override
+    public Message resetMsg(Message message) {
+        if (message instanceof GenericMessageWarp){
+            try {
+                GenericMessageWarp messageWarp = (GenericMessageWarp) message;
+                Field headers = message.getClass().getSuperclass().getDeclaredField("headers");
+                headers.setAccessible(true);
+                headers.set(message, messageWarp.getMessageHeaderWarp());
+            } catch (NoSuchFieldException e) {
+                LogUtil.warn("MessageAdapterImpl.resetMsg - NoSuchFieldException", e);
+            } catch (IllegalAccessException e) {
+                LogUtil.warn("MessageAdapterImpl.resetMsg - IllegalAccessException", e);
+            }
+        }
+        return message;
+    }
+
+
 }

arex-instrumentation/mq/arex-integration-mqtt/src/main/java/io/arex/inst/mqtt/warp/GenericMessageWarp.java

@@ -38,4 +38,15 @@ public void put(String key, String value) {
             this.messageHeaderWarp.put(key, value);
         }
     }
+
+    public Object get(String key){
+        if (this.messageHeaderWarp != null) {
+            return this.messageHeaderWarp.get(key);
+        }
+        return getHeaders().get(key);
+    }
+
+    public MessageHeaderWarp getMessageHeaderWarp() {
+        return messageHeaderWarp;
+    }
 }

arex-instrumentation/mq/arex-integration-mqtt/src/main/java/io/arex/inst/mqtt/warp/MessageHeaderWarp.java

@@ -14,6 +14,8 @@ public class MessageHeaderWarp extends MessageHeaders {
 
     public MessageHeaderWarp(MessageHeaders messageHeaders) {
         super(messageHeaders);
+        this.put(ID,messageHeaders.get(ID));
+        this.put(TIMESTAMP,messageHeaders.get(TIMESTAMP));
     }
 
     public MessageHeaderWarp(Map<String, Object> headers) {
@@ -25,6 +27,9 @@ public MessageHeaderWarp(Map<String, Object> headers, UUID id, Long timestamp) {
     }
 
     public Object put(String key, Object value){
+        if (value == null){
+            return null;
+        }
         super.getRawHeaders().put(key,value);
         return value;
     }