提交部分代码

Author: MentosL

arex-instrumentation/mq/arex-integration-mqtt/pom.xml

@@ -14,9 +14,9 @@
 
     <dependencies>
         <dependency>
-            <groupId>org.springframework.integration</groupId>
-            <artifactId>spring-integration-mqtt</artifactId>
-            <version>5.5.10</version>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-messaging</artifactId>
+            <version>5.3.4</version>
             <scope>provided</scope>
         </dependency>
     </dependencies>

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

@@ -2,12 +2,32 @@
 
 import io.arex.agent.bootstrap.model.Mocker;
 import io.arex.inst.runtime.util.MockUtils;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
 
 /**
  * MQTTAdapterHelper
  */
 public class MQTTAdapterHelper {
 
+
+    public static <MC,T extends Message>  void onServiceEnter(MessageAdapter<MC,T> adapter, MessageChannel messageChannel,Message<?> message){
+        if (shouldSkip(messageChannel,message)){
+            return;
+        }
+
+
+    }
+
+
+
+    private static boolean shouldSkip(MessageChannel messageChannel,Message<?> mqttMessage){
+        if (messageChannel== null || mqttMessage == null) {
+            return true;
+        }
+        return false;
+    }
+
     public static Mocker createMocker(String operationName) {
         Mocker mocker = MockUtils.createMqttConsumer(operationName);
         mocker.getTargetRequest().setType(Byte.class.getName());

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

@@ -0,0 +1,14 @@
+package io.arex.inst.mqtt;
+
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+
+/**
+ * MessageAdapter
+ */
+public interface MessageAdapter<C,T extends Message> {
+
+    byte[] getMsg(C c, T msg);
+
+    MessageHeaders getHeader(C c, T msg);
+}

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

@@ -0,0 +1,36 @@
+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;
+
+/**
+ * GenericMessageAdapter
+ */
+public class GenericMessageImpl implements MessageAdapter<MessageChannel, GenericMessage> {
+
+    private static final GenericMessageImpl INSTANCE = new GenericMessageImpl();
+
+    public static GenericMessageImpl getInstance() {
+        return INSTANCE;
+    }
+
+    @Override
+    public byte[] getMsg(MessageChannel messageChannel, GenericMessage msg) {
+
+        Object payload = msg.getPayload();
+
+        if (payload instanceof  byte[]){
+            return ((byte[]) payload);
+        }
+        return payload.toString().getBytes(StandardCharsets.UTF_8);
+    }
+
+    @Override
+    public MessageHeaders getHeader(MessageChannel messageChannel, GenericMessage msg) {
+        return msg.getHeaders();
+    }
+}

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

@@ -0,0 +1,72 @@
+package io.arex.inst.mqtt.inst;
+
+import io.arex.agent.bootstrap.model.Mocker;
+import io.arex.agent.bootstrap.util.StringUtil;
+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.runtime.context.ContextManager;
+import io.arex.inst.runtime.listener.CaseEvent;
+import io.arex.inst.runtime.listener.CaseEventDispatcher;
+import io.arex.inst.runtime.util.MockUtils;
+import net.bytebuddy.asm.Advice;
+import net.bytebuddy.description.method.MethodDescription;
+import net.bytebuddy.description.type.TypeDescription;
+import net.bytebuddy.matcher.ElementMatcher;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+
+import java.util.Base64;
+import java.util.Collections;
+import java.util.List;
+
+import static io.arex.inst.extension.matcher.SafeExtendsClassMatcher.extendsClass;
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.takesArgument;
+
+/**
+ * EclipseInstrumentationV3
+ */
+public class MQGenericInstrumentationV3 extends TypeInstrumentation {
+    @Override
+    protected ElementMatcher<TypeDescription> typeMatcher() {
+        return extendsClass(named("org.springframework.messaging.core.AbstractMessageSendingTemplate"), false)
+                .and(named("org.springframework.messaging.core.GenericMessagingTemplate"));
+    }
+
+    @Override
+    public List<MethodInstrumentation> methodAdvices() {
+        ElementMatcher<MethodDescription> matcher = named("doSend")
+                .and(takesArgument(0, named("org.springframework.messaging.MessageChannel")))
+                .and(takesArgument(1, named("org.springframework.messaging.Message")));
+        return Collections.singletonList(new MethodInstrumentation(matcher, ArrivedAdvice.class.getName()));
+    }
+
+    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);
+
+
+           if (shouldSkip(messageChannel,message)){
+               return;
+           }
+            CaseEventDispatcher.onEvent(CaseEvent.ofEnterEvent());
+//            if (ContextManager.needRecordOrReplay()){
+//                Mocker mocker = MQTTAdapterHelper.createMocker(topic);
+//                mocker.getTargetRequest().setBody(Base64.getEncoder().encodeToString(mqttMessage.getPayload()));
+//                if (ContextManager.needReplay()) {
+//                    MockUtils.replayMocker(mocker);
+//                } else if (ContextManager.needRecord()) {
+//                    MockUtils.recordMocker(mocker);
+//                }
+//            }
+        }
+
+    }
+
+}

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

@@ -19,6 +19,6 @@ public MQTTAdapterModuleInstrumentation() {
 
     @Override
     public List<TypeInstrumentation> instrumentationTypes() {
-        return Arrays.asList(new EclipseInstrumentationV3());
+        return Arrays.asList(new MQGenericInstrumentationV3());
     }
 }

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

@@ -0,0 +1,26 @@
+package io.arex.inst.mqtt.inst;
+
+
+import net.bytebuddy.description.type.TypeDescription;
+import org.junit.jupiter.api.Test;
+import org.springframework.messaging.core.AbstractMessageSendingTemplate;
+import org.springframework.messaging.core.GenericMessagingTemplate;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class EclipseInstrumentationV3Test {
+    MQGenericInstrumentationV3 inst = new MQGenericInstrumentationV3();
+
+    @Test
+    void typeMatcher() {
+        assertFalse(inst.typeMatcher().matches(TypeDescription.ForLoadedType.of(AbstractMessageSendingTemplate.class)));
+        assertTrue(inst.typeMatcher().matches(TypeDescription.ForLoadedType.of(GenericMessagingTemplate.class)));
+    }
+
+    @Test
+    void methodAdvices() {
+        assertEquals(1, inst.methodAdvices().size());
+    }
+}