提交初次mqtt代码

增加mqtt相关参数

Author: MentosL

arex-agent-bootstrap/src/main/java/io/arex/agent/bootstrap/model/MockCategoryType.java

@@ -19,6 +19,8 @@ public class MockCategoryType implements Serializable {
     public static final MockCategoryType DUBBO_CONSUMER = createDependency("DubboConsumer");
     public static final MockCategoryType DUBBO_PROVIDER = createEntryPoint("DubboProvider");
     public static final MockCategoryType DUBBO_STREAM_PROVIDER = createDependency("DubboStreamProvider");
+    public static final MockCategoryType MQTT_MESSAGE_CONSUMER = createEntryPoint("MqttMessageConsumer");
+
 
     private String name;
     private boolean entryPoint;

arex-agent/pom.xml

@@ -147,6 +147,11 @@
             <artifactId>arex-jcasbin</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>arex-integration-mqtt</artifactId>
+            <version>${project.version}</version>
+        </dependency>
         <!--Agent instrumentation end-->
 
         <dependency>

arex-instrumentation-api/src/main/java/io/arex/inst/runtime/util/MockUtils.java

@@ -68,6 +68,11 @@ public static ArexMocker createDubboStreamProvider(String operationName) {
         return create(MockCategoryType.DUBBO_STREAM_PROVIDER, operationName);
     }
 
+    public static ArexMocker createMqttConsumer(String operationName){
+        return create(MockCategoryType.MQTT_MESSAGE_CONSUMER,operationName);
+    }
+
+
     public static ArexMocker create(MockCategoryType categoryType, String operationName) {
         ArexMocker mocker = new ArexMocker();
         long createTime = System.currentTimeMillis();

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

@@ -0,0 +1,23 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>arex-instrumentation-parent</artifactId>
+        <groupId>io.arex</groupId>
+        <version>0.2.0</version>
+        <relativePath>../../pom.xml</relativePath>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>arex-integration-mqtt</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-mqtt</artifactId>
+            <version>5.5.10</version>
+            <scope>provided</scope>
+        </dependency>
+    </dependencies>
+</project>

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

@@ -0,0 +1,16 @@
+package io.arex.inst.mqtt;
+
+import io.arex.agent.bootstrap.model.Mocker;
+import io.arex.inst.runtime.util.MockUtils;
+
+/**
+ * MQTTAdapterHelper
+ */
+public class MQTTAdapterHelper {
+
+    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/inst/EclipseInstrumentationV3.java

@@ -0,0 +1,63 @@
+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.runtime.context.ContextManager;
+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.eclipse.paho.client.mqttv3.MqttMessage;
+
+import java.util.Base64;
+import java.util.Collections;
+import java.util.List;
+
+import static net.bytebuddy.matcher.ElementMatchers.hasSuperType;
+import static net.bytebuddy.matcher.ElementMatchers.isInterface;
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.not;
+import static net.bytebuddy.matcher.ElementMatchers.takesArgument;
+
+/**
+ * EclipseInstrumentationV3
+ */
+public class EclipseInstrumentationV3 extends TypeInstrumentation {
+    @Override
+    protected ElementMatcher<TypeDescription> typeMatcher() {
+        return not(isInterface()).and(hasSuperType(named("org.eclipse.paho.client.mqttv3.MqttCallback")));
+    }
+
+    @Override
+    public List<MethodInstrumentation> methodAdvices() {
+        ElementMatcher<MethodDescription> matcher = named("messageArrived")
+                .and(takesArgument(0, named("java.lang.String")))
+                .and(takesArgument(1, named("org.eclipse.paho.client.mqttv3.MqttMessage")));
+        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) String topic,
+                                   @Advice.Argument(value = 1, readOnly = false) MqttMessage mqttMessage) {
+            if (StringUtil.isEmpty(topic) || mqttMessage == null) {
+                return;
+            }
+            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/MQTTAdapterModuleInstrumentation.java

@@ -0,0 +1,24 @@
+package io.arex.inst.mqtt.inst;
+
+import com.google.auto.service.AutoService;
+import io.arex.inst.extension.ModuleInstrumentation;
+import io.arex.inst.extension.TypeInstrumentation;
+
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * MQTTAdapterModuleInstrumentation
+ */
+@AutoService(ModuleInstrumentation.class)
+public class MQTTAdapterModuleInstrumentation extends ModuleInstrumentation {
+
+    public MQTTAdapterModuleInstrumentation() {
+        super("mqtt-adapter");
+    }
+
+    @Override
+    public List<TypeInstrumentation> instrumentationTypes() {
+        return Arrays.asList(new EclipseInstrumentationV3());
+    }
+}

arex-instrumentation/pom.xml

@@ -43,6 +43,7 @@
         <module>authentication/arex-shiro</module>
         <module>authentication/arex-jcasbin</module>
         <module>foundation/arex-serializer</module>
+        <module>mq/arex-integration-mqtt</module>
     </modules>
     <dependencies>
         <dependency>