Merge pull request #2219 from jackyluo-learning/protocol-amqp-shadow2

[ISSUE #2218]fixed bug in AmqpMessage
close #2218

Author: xwm1992

Diff for: eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/amqp/AmqpMessage.java

@@ -19,6 +19,7 @@
 
 package org.apache.eventmesh.common.protocol.amqp;
 
+import com.rabbitmq.client.impl.AMQContentHeader;
 import org.apache.eventmesh.common.protocol.ProtocolTransportObject;
 import org.apache.eventmesh.common.protocol.amqp.common.ProtocolKey;
 
@@ -35,7 +36,7 @@
  */
 @Data
 public class AmqpMessage implements ProtocolTransportObject {
-    private ContentHeader contentHeader;
+    private AMQContentHeader contentHeader;
 
     private byte[] contentBody;
 
@@ -45,7 +46,7 @@ public AmqpMessage() {
         this(null, null, null);
     }
 
-    public AmqpMessage(ContentHeader contentHeader, byte[] contentBody, Map<String, Object> extendInfo) {
+    public AmqpMessage(AMQContentHeader contentHeader, byte[] contentBody, Map<String, Object> extendInfo) {
         this.contentHeader = contentHeader;
         this.contentBody = contentBody;
         this.extendInfo = extendInfo;

Diff for: eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/amqp/common/ProtocolKey.java

@@ -19,8 +19,6 @@
 
 package org.apache.eventmesh.common.protocol.amqp.common;
 
-import java.util.Map;
-
 public class ProtocolKey {
     public static final String ROUTING_KEY = "routingKey";
 
@@ -32,7 +30,7 @@ public class ProtocolKey {
 
     public static final String QUEUE_NAME = "queueName";
 
-    public static final String BASIC_PROPERTIES = "basicProperties";
+    public static final String AMQ_CONTEND_HEADER = "contentHeader";
 
     public static class ContentHeaderProperties {
         public static final String CLASS_ID = "classId";

Diff for: eventmesh-protocol-plugin/eventmesh-protocol-amqp/src/main/java/org/apache/eventmesh/protocol/amqp/resolver/AmqpProtocolResolver.java

@@ -17,21 +17,24 @@
 
 package org.apache.eventmesh.protocol.amqp.resolver;
 
-import com.alibaba.fastjson.JSON;
-import com.rabbitmq.client.AMQP.BasicProperties;
-import io.cloudevents.CloudEvent;
-import io.cloudevents.core.v1.CloudEventBuilder;
-import org.apache.commons.collections4.MapUtils;
 import org.apache.eventmesh.common.protocol.amqp.AmqpMessage;
 import org.apache.eventmesh.common.protocol.amqp.common.ProtocolKey;
 import org.apache.eventmesh.protocol.amqp.AMQPProtocolConstant;
 
+import org.apache.commons.collections4.MapUtils;
+
 import java.net.URI;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
 import java.util.UUID;
 
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.v1.CloudEventBuilder;
+
+import com.alibaba.fastjson.JSON;
+import com.rabbitmq.client.impl.AMQContentHeader;
+
 /**
  * Resolve AmqpMessage or CloudEvent
  */
@@ -46,19 +49,19 @@ public static CloudEvent buildEvent(AmqpMessage amqpMessage) {
         String queueName = "";
         Map<String, Object> extendInfo = amqpMessage.getExtendInfo();
         if (MapUtils.isNotEmpty(extendInfo)) {
-            // routingKey, exchange去掉，路由信息在这之前完成
+            // removed routingKey, exchange, routing should be finished before
 //            routingKey = extendInfo.get(ProtocolKey.ROUTING_KEY).toString();
 //            exchange = extendInfo.get(ProtocolKey.EXCHANGE).toString();
             queueName = extendInfo.get(ProtocolKey.QUEUE_NAME).toString();
         }
-        BasicProperties amqBasicProperties = amqpMessage.getContentHeader();
+        AMQContentHeader amqContentHeader = amqpMessage.getContentHeader();
         byte[] contentBody = amqpMessage.getContentBody();
         cloudEventBuilder
                 .withId(id)
                 .withSource(source)
                 .withType(AMQPProtocolConstant.PROTOCOL_NAME)
                 .withSubject(queueName)
-                .withExtension(ProtocolKey.BASIC_PROPERTIES, JSON.toJSONString(amqBasicProperties))
+                .withExtension(ProtocolKey.AMQ_CONTEND_HEADER, JSON.toJSONString(amqContentHeader))
                 .withData(contentBody);
         return cloudEventBuilder.build();
     }
@@ -70,8 +73,8 @@ public static AmqpMessage buildAmqpMessage(CloudEvent cloudEvent) {
         extendInfo.put(ProtocolKey.ROUTING_KEY, routingKey);
         extendInfo.put(ProtocolKey.EXCHANGE, exchange);
         byte[] contentBody = Objects.requireNonNull(cloudEvent.getData()).toBytes();
-        BasicProperties amqBasicProperties = JSON.parseObject(Objects.requireNonNull(cloudEvent.getExtension(ProtocolKey.BASIC_PROPERTIES)).toString(), BasicProperties.class);
-        return new AmqpMessage(amqBasicProperties, contentBody, extendInfo);
+        AMQContentHeader amqContentHeader = JSON.parseObject(Objects.requireNonNull(cloudEvent.getExtension(ProtocolKey.AMQ_CONTEND_HEADER)).toString(), AMQContentHeader.class);
+        return new AmqpMessage(amqContentHeader, contentBody, extendInfo);
     }
 
 }