提交

Author: MentosL

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

@@ -1,6 +1,6 @@
 package io.arex.inst.mqtt;
 
-import io.arex.agent.bootstrap.model.Mocker;
+import io.arex.agent.bootstrap.internal.Pair;
 import io.arex.agent.bootstrap.util.StringUtil;
 import io.arex.inst.mqtt.adapter.MessageAdapter;
 import io.arex.inst.runtime.config.Config;
@@ -9,7 +9,6 @@
 import io.arex.inst.runtime.listener.EventSource;
 import io.arex.inst.runtime.model.ArexConstants;
 import io.arex.inst.runtime.util.IgnoreUtils;
-import io.arex.inst.runtime.util.MockUtils;
 
 /**
  * MQTTAdapterHelper
@@ -19,25 +18,26 @@ public class MQTTAdapterHelper {
     public static final String PROCESSED_FLAG = "arex-processed-flag";
 
 
-    public static <MC, Msg> void onServiceEnter(MessageAdapter<MC, Msg> adapter, Object messageChannel, Object message) {
+    public static <MC, Msg> Pair<MC, Msg> onServiceEnter(MessageAdapter<MC, Msg> adapter, Object messageChannel, Object message) {
         Msg msg = adapter.warpMessage(message);
         if (msg == null) {
-            return;
+            return null;
         }
         if (adapter.markProcessed(msg, PROCESSED_FLAG)) {
-            return;
+            return null;
         }
         MC mc = adapter.warpMC(messageChannel);
         if (mc == null){
-            return;
+            return null;
         }
         CaseEventDispatcher.onEvent(CaseEvent.ofEnterEvent());
         if (shouldSkip(adapter, mc, msg)) {
-            return ;
+            return null;
         }
         String caseId = adapter.getHeader(mc, msg, ArexConstants.RECORD_ID);
         String excludeMockTemplate = adapter.getHeader(mc, msg, ArexConstants.HEADER_EXCLUDE_MOCK);
         CaseEventDispatcher.onEvent(CaseEvent.ofCreateEvent(EventSource.of(caseId, excludeMockTemplate)));
+        return Pair.of(mc,msg);
     }
 
 

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

@@ -2,17 +2,15 @@
 
 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;
 
 /**
- * MessageImpl
+ * MessageAdapterImpl
  */
 public class MessageAdapterImpl implements MessageAdapter<MessageChannel, Message> {
 
@@ -36,10 +34,7 @@ public MessageChannel warpMC(Object messageChannel) {
         if (messageChannel == null){
             return null;
         }
-        if (messageChannel instanceof MessageChannel){
-            return (MessageChannel) messageChannel;
-        }
-        return null;
+        return (MessageChannel) messageChannel;
     }
 
     @Override
@@ -62,10 +57,6 @@ 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();
     }
 
@@ -83,12 +74,6 @@ 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;
@@ -111,23 +96,4 @@ 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/adapter/MessageAdapterImpl.java

@@ -3,7 +3,7 @@
 import io.arex.inst.extension.MethodInstrumentation;
 import io.arex.inst.extension.TypeInstrumentation;
 import io.arex.inst.mqtt.MQTTAdapterHelper;
-import io.arex.inst.mqtt.adapter.GenericMessageImpl;
+import io.arex.inst.mqtt.adapter.MessageAdapterImpl;
 import net.bytebuddy.asm.Advice;
 import net.bytebuddy.description.method.MethodDescription;
 import net.bytebuddy.description.type.TypeDescription;
@@ -41,13 +41,13 @@ public static class ArrivedAdvice {
         @Advice.OnMethodEnter(suppress = Throwable.class)
         public static void onEnter(@Advice.Argument(value = 0, readOnly = false) MessageChannel messageChannel,
                                    @Advice.Argument(value = 1, readOnly = false) Message<?> message) {
-            MQTTAdapterHelper.onServiceEnter(GenericMessageImpl.getInstance(),messageChannel,message);
+            MQTTAdapterHelper.onServiceEnter(MessageAdapterImpl.getInstance(), messageChannel, message);
         }
 
         @Advice.OnMethodExit(suppress = Throwable.class)
         public static void onExit(@Advice.Argument(value = 0, readOnly = false) MessageChannel messageChannel,
                                   @Advice.Argument(value = 1, readOnly = false) Message<?> message) {
-            MQTTAdapterHelper.onServiceExit(GenericMessageImpl.getInstance(), messageChannel, message);
+            MQTTAdapterHelper.onServiceExit(MessageAdapterImpl.getInstance(), messageChannel, message);
         }
 
     }

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

@@ -12,18 +12,23 @@
 public class GenericMessageWarp extends GenericMessage {
 
     private MessageHeaderWarp messageHeaderWarp;
+    private Object payload;
 
     public GenericMessageWarp(Object payload) {
         super(payload);
+        this.payload = payload;
+        this.messageHeaderWarp = null;
     }
 
     public GenericMessageWarp(Object payload, Map headers) {
         super(payload, headers);
+        this.payload = payload;
         this.messageHeaderWarp = new MessageHeaderWarp(headers);
     }
 
     public GenericMessageWarp(Object payload, MessageHeaders headers) {
         super(payload, headers);
+        this.payload = payload;
         this.messageHeaderWarp = new MessageHeaderWarp(headers);
     }
 
@@ -38,15 +43,4 @@ 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/GenericMessageWarp.java

@@ -18,7 +18,7 @@ public class MessageHeaderWarp extends MessageHeaders {
     private final Long timestamp;
 
     public MessageHeaderWarp(MessageHeaders messageHeaders) {
-        super(null);
+        super(messageHeaders);
         this.headers = new HashMap<>();
         for (Map.Entry<String, Object> entry : messageHeaders.entrySet()) {
             this.headers.put(entry.getKey(), entry.getValue());