增加mqtt代码

Author: MentosL

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

@@ -2,10 +2,13 @@
 
 import io.arex.agent.bootstrap.model.Mocker;
 import io.arex.agent.bootstrap.util.StringUtil;
+import io.arex.inst.mqtt.adapter.MessageAdapter;
 import io.arex.inst.runtime.config.Config;
 import io.arex.inst.runtime.listener.CaseEvent;
 import io.arex.inst.runtime.listener.CaseEventDispatcher;
+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;
 
 /**
@@ -29,6 +32,23 @@ public static <MC, Msg> void onServiceEnter(MessageAdapter<MC, Msg> adapter, Obj
             return;
         }
         CaseEventDispatcher.onEvent(CaseEvent.ofEnterEvent());
+        if (shouldSkip(adapter, mc, msg)) {
+            return ;
+        }
+        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)));
+    }
+
+
+    public static <MC, Msg> void onServiceExit(MessageAdapter<MC, Msg> adapter, Object messageChannel, Object message){
+        Msg msg = adapter.warpMessage(message);
+        MC mc = adapter.warpMC(messageChannel);
+        if (msg == null || mc == null) {
+            return;
+        }
+        adapter.removeHeader(mc,msg,PROCESSED_FLAG);
+        new MessageQueueExtractor<>( mc, msg,adapter).execute();
     }
 
 
@@ -50,19 +70,14 @@ private static<MC, Msg>  boolean shouldSkip(MessageAdapter<MC, Msg> adapter,MC m
         if (Boolean.parseBoolean(adapter.getHeader(mc, msg, ArexConstants.REPLAY_WARM_UP))) {
             return true;
         }
-
+        String topic = adapter.getHeader(mc, msg, ArexConstants.REPLAY_WARM_UP);
+        if (StringUtil.isEmpty(topic)) {
+            return false;
+        }
+        if (IgnoreUtils.ignoreOperation(topic)) {
+            return true;
+        }
+        return Config.get().invalidRecord(topic);
     }
 
-
-
-
-
-
-
-
-    public static Mocker createMocker(String operationName) {
-        Mocker mocker = MockUtils.createMqttConsumer(operationName);
-        mocker.getTargetRequest().setType(Byte.class.getName());
-        return mocker;
-    }
 }

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

@@ -51,8 +51,6 @@ 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/MessageQueueExtractor.java

@@ -1,11 +1,12 @@
 package io.arex.inst.mqtt.adapter;
 
-import io.arex.inst.mqtt.MessageAdapter;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.support.GenericMessage;
 
 import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.UUID;
 
 /**
  * GenericMessageAdapter
@@ -20,9 +21,7 @@ public static GenericMessageImpl getInstance() {
 
     @Override
     public byte[] getMsg(MessageChannel messageChannel, GenericMessage msg) {
-
         Object payload = msg.getPayload();
-
         if (payload instanceof  byte[]){
             return ((byte[]) payload);
         }
@@ -67,4 +66,14 @@ public String getHeader(MessageChannel messageChannel, GenericMessage genericMes
         Object obj = genericMessage.getHeaders().get(key);
         return obj != null ? obj.toString() : null;
     }
+
+    @Override
+    public void removeHeader(MessageChannel messageChannel, GenericMessage genericMessage, String key) {
+        genericMessage.getHeaders().remove(key);
+    }
+
+    @Override
+    public void addHeader(MessageChannel messageChannel, GenericMessage genericMessage, String key, String value) {
+        genericMessage.getHeaders().put(key,value);
+    }
 }

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

@@ -23,6 +23,4 @@ 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/MessageAdapter.java

@@ -1,10 +1,9 @@
 package io.arex.inst.mqtt.inst;
 
-import io.arex.agent.bootstrap.internal.Pair;
 import io.arex.inst.extension.MethodInstrumentation;
 import io.arex.inst.extension.TypeInstrumentation;
 import io.arex.inst.mqtt.MQTTAdapterHelper;
-import io.arex.inst.mqtt.adapter.MessageAdapterImpl;
+import io.arex.inst.mqtt.adapter.GenericMessageImpl;
 import net.bytebuddy.asm.Advice;
 import net.bytebuddy.description.method.MethodDescription;
 import net.bytebuddy.description.type.TypeDescription;
@@ -41,24 +40,14 @@ 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,
-                                   @Advice.Local("channelMsgPair") Pair<MessageChannel, Message> pair) {
-            Pair<MessageChannel, Message> messageChannelMessagePair =
-                    MQTTAdapterHelper.onServiceEnter(MessageAdapterImpl.getInstance(), messageChannel, message);
-            if (messageChannelMessagePair == null){
-                return;
-            }
-            pair = messageChannelMessagePair;
+                                   @Advice.Argument(value = 1, readOnly = false) Message<?> message) {
+            MQTTAdapterHelper.onServiceEnter(GenericMessageImpl.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,
-                                  @Advice.Local("channelMsgPair") Pair<MessageChannel, Message> pair) {
-            if (pair != null){
-                message = pair.getSecond();
-            }
-            MQTTAdapterHelper.onServiceExit(MessageAdapterImpl.getInstance(), messageChannel, message);
+                                  @Advice.Argument(value = 1, readOnly = false) Message<?> message) {
+            MQTTAdapterHelper.onServiceExit(GenericMessageImpl.getInstance(), messageChannel, message);
         }
 
     }