update basicPublish method in AmqpChannel

Author: jackyluo-learning

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

@@ -24,6 +24,8 @@
 
 import java.util.Map;
 
+import com.rabbitmq.client.ContentHeader;
+
 import com.rabbitmq.client.AMQP.BasicProperties;
 
 import lombok.Data;
@@ -33,7 +35,7 @@
  */
 @Data
 public class AmqpMessage implements ProtocolTransportObject {
-    private BasicProperties contentHeader;
+    private ContentHeader contentHeader;
 
     private byte[] contentBody;
 
@@ -43,16 +45,12 @@ public AmqpMessage() {
         this(null, null, null);
     }
 
-    public AmqpMessage(BasicProperties contentHeader, byte[] contentBody, Map<String, Object> extendInfo) {
+    public AmqpMessage(ContentHeader contentHeader, byte[] contentBody, Map<String, Object> extendInfo) {
         this.contentHeader = contentHeader;
         this.contentBody = contentBody;
         this.extendInfo = extendInfo;
     }
 
-    public long getSize() {
-        return contentBody == null ? 0 : contentHeader == null ? contentBody.length : contentBody.length + contentHeader.getBodySize();
-    }
-
     public String getExchange() {
         return extendInfo.getOrDefault(ProtocolKey.EXCHANGE, "").toString();
     }

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshAmqpServer.java

@@ -17,13 +17,10 @@
 
 package org.apache.eventmesh.runtime.boot;
 
-import io.netty.bootstrap.ServerBootstrap;
-import io.netty.buffer.PooledByteBufAllocator;
-import io.netty.channel.*;
-import io.netty.channel.socket.SocketChannel;
-import io.netty.channel.socket.nio.NioServerSocketChannel;
 import org.apache.eventmesh.runtime.configuration.EventMeshAmqpConfiguration;
+import org.apache.eventmesh.runtime.core.protocol.amqp.exchange.RouteComponent;
 import org.apache.eventmesh.runtime.core.protocol.amqp.metadata.MetaStore;
+import org.apache.eventmesh.runtime.core.protocol.amqp.producer.AmqpProducerManager;
 import org.apache.eventmesh.runtime.core.protocol.amqp.remoting.codec.AmqpCodeDecoder;
 import org.apache.eventmesh.runtime.core.protocol.amqp.remoting.codec.AmqpCodeEncoder;
 import org.apache.eventmesh.runtime.core.protocol.amqp.service.ExchangeService;
@@ -32,6 +29,12 @@
 import org.apache.eventmesh.runtime.core.protocol.amqp.service.QueueServiceImpl;
 import org.apache.eventmesh.runtime.registry.Registry;
 
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.buffer.PooledByteBufAllocator;
+import io.netty.channel.*;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+
 public class EventMeshAmqpServer extends AbstractRemotingServer {
 
     private EventMeshServer eventMeshServer;
@@ -46,12 +49,18 @@ public class EventMeshAmqpServer extends AbstractRemotingServer {
 
     private MetaStore metaStore;
 
+    private RouteComponent routeComponent;
+
+    private AmqpProducerManager producerManager;
+
     public EventMeshAmqpServer(EventMeshServer eventMeshServer,
-                               EventMeshAmqpConfiguration eventMeshAmqpConfiguration, Registry registry) {
+                               EventMeshAmqpConfiguration eventMeshAmqpConfiguration, Registry registry, RouteComponent routeComponent, AmqpProducerManager producerManager) {
         super();
         this.eventMeshServer = eventMeshServer;
         this.eventMeshAmqpConfiguration = eventMeshAmqpConfiguration;
         this.registry = registry;
+        this.routeComponent = routeComponent;
+        this.producerManager = producerManager;
         this.metaStore=new MetaStore(this);
         this.exchangeService=new ExchangeServiceImpl(this,metaStore);
         this.queueService=new QueueServiceImpl(this,metaStore);
@@ -131,6 +140,18 @@ public QueueService getQueueService() {
         return queueService;
     }
 
+    public MetaStore getMetaStore() {
+        return metaStore;
+    }
+
+    public RouteComponent getRouteComponent() {
+        return routeComponent;
+    }
+
+    public AmqpProducerManager getProducerManager() {
+        return producerManager;
+    }
+
     /**
      * A channel initializer that initialize channels for amqp protocol.
      */