Compatible with 3.x and 4.x RabbitMQ Client (#389)

CHANGES.md

@@ -11,6 +11,7 @@ Release Notes.
 * Support collecting dubbo thread pool metrics
 * Bump up byte-buddy to 1.12.19
 * Upgrade agent test tools
+* [Breaking Change] Compatible with 3.x and 4.x RabbitMQ Client, rename `rabbitmq-5.x-plugin` to `rabbitmq-plugin`
 
 #### Documentation
 

apm-sniffer/apm-sdk-plugin/pom.xml

@@ -70,7 +70,7 @@
         <module>elasticsearch-6.x-plugin</module>
         <module>elasticsearch-7.x-plugin</module>
         <module>undertow-plugins</module>
-        <module>rabbitmq-5.x-plugin</module>
+        <module>rabbitmq-plugin</module>
         <module>dubbo-conflict-patch</module>
         <module>canal-1.x-plugin</module>
         <module>dubbo-2.7.x-plugin</module>

apm-sniffer/apm-sdk-plugin/rabbitmq-plugin/pom.xml

@@ -25,8 +25,8 @@
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
-    <artifactId>apm-rabbitmq-5.x-plugin</artifactId>
-    <name>rabbitmq-5.x-plugin</name>
+    <artifactId>apm-rabbitmq-plugin</artifactId>
+    <name>rabbitmq-plugin</name>
     <packaging>jar</packaging>
 
     <properties>

apm-sniffer/apm-sdk-plugin/rabbitmq-plugin/src/main/java/org/apache/skywalking/apm/plugin/rabbitmq/define/ChannelNInstrumentation.java

@@ -45,7 +45,7 @@ public class ChannelNInstrumentation extends ClassInstanceMethodsEnhancePluginDe
             new ConstructorInterceptPoint() {
                 @Override
                 public ElementMatcher<MethodDescription> getConstructorMatcher() {
-                    return takesArgumentWithType(3, "com.rabbitmq.client.MetricsCollector");
+                    return takesArgumentWithType(0, "com.rabbitmq.client.impl.AMQConnection");
                 }
 
                 @Override

apm-sniffer/apm-sdk-plugin/rabbitmq-plugin/src/main/resources/skywalking-plugin.def

@@ -14,4 +14,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-rabbitmq-5.x=org.apache.skywalking.apm.plugin.rabbitmq.define.ChannelNInstrumentation
+rabbitmq=org.apache.skywalking.apm.plugin.rabbitmq.define.ChannelNInstrumentation

docs/en/setup/service-agent/java-agent/Plugin-list.md

@@ -75,7 +75,7 @@
 - pulsar-2.2-2.7
 - quasar
 - quartz-scheduler-2.x
-- rabbitmq-5.x
+- rabbitmq
 - redisson-3.x
 - resteasy-server-3.x
 - resteasy-server-4.x

docs/en/setup/service-agent/java-agent/Supported-list.md

@@ -69,7 +69,7 @@ metrics based on the tracing data.
   * [Kafka](http://kafka.apache.org) 0.11.0.0 -> 2.8.0
   * [Spring-Kafka](https://github.com/spring-projects/spring-kafka) Spring Kafka Consumer 1.3.x -> 2.3.x (2.0.x and 2.1.x not tested and not recommended by [the official document](https://spring.io/projects/spring-kafka))
   * [ActiveMQ](https://github.com/apache/activemq) 5.10.0 -> 5.15.4
-  * [RabbitMQ](https://www.rabbitmq.com/) 5.x
+  * [RabbitMQ](https://www.rabbitmq.com/) 3.x-> 5.x
   * [Pulsar](http://pulsar.apache.org) 2.2.x -> 2.9.x
   * [NATS](https://github.com/nats-io/nats.java) 2.14.x -> 2.15.x
   * Aliyun ONS 1.x (Optional¹)

test/plugin/scenarios/rabbitmq-scenario/src/main/java/org/apache/skywalking/apm/testcase/rabbitmq/controller/CaseController.java

@@ -22,7 +22,10 @@ import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
-import com.rabbitmq.client.DeliverCallback;
+import com.rabbitmq.client.DefaultConsumer;
+import com.rabbitmq.client.Envelope;
+
+import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -78,13 +81,16 @@ public class CaseController {
             LOGGER.info("Message has been published-------------->" + MESSAGE);
 
             final CountDownLatch waitForConsume = new CountDownLatch(1);
-            DeliverCallback deliverCallback = (consumerTag, delivery) -> {
-                String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
-                LOGGER.info("Message received-------------->" + message);
-                waitForConsume.countDown();
+            DefaultConsumer callback = new DefaultConsumer(channel) {
+                @Override
+                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
+                    String message = new String(body, "UTF-8");
+                    LOGGER.info("Message received-------------->" + message);
+                    waitForConsume.countDown();
+                }
             };
-            channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> {
-            });
+
+            channel.basicConsume(QUEUE_NAME, true, callback);
             waitForConsume.await(5000L, TimeUnit.MILLISECONDS);
             LOGGER.info("Message Consumed-------------->");
 

test/plugin/scenarios/rabbitmq-scenario/support-version.list

@@ -14,9 +14,9 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-5.5.1
-5.4.3
-5.3.0
-5.2.0
-5.1.2
+5.16.0
 5.0.0
+4.12.0
+4.0.0
+3.6.6
+3.0.0