Merge pull request #2517 from jackyluo-learning/protocol-amqp-shadow1

[ISSUE #2516] AmqpConsumer bugfix

Author: qqeasonchen

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

@@ -19,15 +19,12 @@
 
 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;
 
 import java.util.Map;
 
-import com.rabbitmq.client.ContentHeader;
-
-import com.rabbitmq.client.AMQP.BasicProperties;
+import com.rabbitmq.client.impl.AMQContentHeader;
 
 import lombok.Data;
 

Diff for: eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/amqp/consumer/AmqpConsumerImpl.java

@@ -8,14 +8,14 @@
 import org.apache.eventmesh.protocol.api.exception.ProtocolHandleException;
 import org.apache.eventmesh.runtime.core.plugin.MQConsumerWrapper;
 import org.apache.eventmesh.runtime.core.protocol.amqp.processor.AmqpChannel;
+import org.apache.eventmesh.runtime.core.protocol.amqp.remoting.protocol.ErrorCodes;
 
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
 
-import org.apache.eventmesh.runtime.core.protocol.amqp.remoting.protocol.ErrorCodes;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -46,7 +46,7 @@ public class AmqpConsumerImpl implements AmqpConsumer {
     /**
      * amqpChannel that current consumer used
      */
-    private AmqpChannel amqpChannel;
+    private AmqpChannel channel;
 
     /**
      * a map that store all un ack message which has been pushed to client
@@ -65,18 +65,20 @@ public void pushMessage(PushMessageContext pushMessageContext) {
             // TODO: 2022/10/20 exception handle
             throw new RuntimeException(e);
         }
-        long deliveryTag = this.amqpChannel.getNextDeliveryTag();
+        long deliveryTag = this.channel.getNextDeliveryTag();
         if (!autoAck) {
             addUnAckMsg(deliveryTag, pushMessageContext);
         }
 
         try {
-            amqpChannel.getConnection().getAmqpOutputConverter().writeDeliver(amqpMessage, this.amqpChannel.getChannelId(),
+            channel.getConnection().getAmqpOutputConverter().writeDeliver(amqpMessage, channel.getChannelId(),
                     false, deliveryTag, consumerTag);
         } catch (IOException e) {
             logger.error("sendMessages IOException", e);
-            amqpChannel.closeChannel(ErrorCodes.INTERNAL_ERROR, "system error");
+            channel.closeChannel(ErrorCodes.INTERNAL_ERROR, "system error");
         }
+
+
     }
 
     @Override
@@ -126,10 +128,10 @@ public void notifyConsumer() {
     }
 
     /**
-     * 添加unAck msg
+     * add unacked msg
      */
     private void addUnAckMsg(Long deliveryTag, PushMessageContext pushMessageContext) {
-        this.amqpChannel.getUnackMessageMap().put(deliveryTag, pushMessageContext);
+        this.channel.getUnacknowledgedMessageMap().add(deliveryTag, pushMessageContext.getMessageId(), this, 1);
         this.unAckMap.put(pushMessageContext.getMessageId(), pushMessageContext);
     }
 }

Diff for: eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/amqp/consumer/QueueConsumerMapping.java

@@ -18,32 +18,29 @@ public class QueueConsumerMapping {
 
     public void registerConsumer(String virtualHost, String queue, AmqpConsumer consumer) throws AmqpNotFoundException {
 
-        ConcurrentHashMap<String, Set<AmqpConsumer>> infoMap = queueConsumerMapping.computeIfAbsent(virtualHost, m -> new ConcurrentHashMap<>());
-        if (infoMap == null) {
-            logger.error("virtualHost not found {}", virtualHost);
-            throw new AmqpNotFoundException("vhost not found");
-        }
+        ConcurrentHashMap<String, Set<AmqpConsumer>> infoMap = getInfoMap(virtualHost);
         Set<AmqpConsumer> consumers = infoMap.computeIfAbsent(queue, m -> new HashSet<>());
         consumers.add(consumer);
     }
 
     public void removeConsumer(String virtualHost, String queue, AmqpConsumer consumer) throws AmqpNotFoundException {
 
-        ConcurrentHashMap<String, Set<AmqpConsumer>> infoMap = queueConsumerMapping.computeIfAbsent(virtualHost, m -> new ConcurrentHashMap<>());
-        if (infoMap == null) {
-            logger.error("virtualHost not found {}", virtualHost);
-            throw new AmqpNotFoundException("vhost not found");
-        }
+        ConcurrentHashMap<String, Set<AmqpConsumer>> infoMap = getInfoMap(virtualHost);
         Set<AmqpConsumer> consumers = infoMap.computeIfAbsent(queue, m -> new HashSet<>());
         consumers.remove(consumer);
     }
 
-    public Set<AmqpConsumer> getConsumers(String virtualHost, String queue) {
+    private ConcurrentHashMap<String, Set<AmqpConsumer>> getInfoMap(String virtualHost) throws AmqpNotFoundException {
         ConcurrentHashMap<String, Set<AmqpConsumer>> infoMap = queueConsumerMapping.computeIfAbsent(virtualHost, m -> new ConcurrentHashMap<>());
-        if (infoMap == null) {
-            return null;
+        if (infoMap.isEmpty()) {
+            logger.error("virtualHost not found {}", virtualHost);
+            throw new AmqpNotFoundException("vhost not found");
         }
+        return infoMap;
+    }
 
+    public Set<AmqpConsumer> getConsumers(String virtualHost, String queue) throws AmqpNotFoundException {
+        ConcurrentHashMap<String, Set<AmqpConsumer>> infoMap = getInfoMap(virtualHost);
         return infoMap.get(queue);
 
     }

Diff for: eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/amqp/downstreamstrategy/RandomDispatchStrategy.java

@@ -2,15 +2,12 @@
 
 import org.apache.eventmesh.runtime.core.protocol.amqp.consumer.AmqpConsumer;
 import org.apache.eventmesh.runtime.core.protocol.amqp.consumer.QueueConsumerMapping;
-
-import org.apache.commons.collections4.MapUtils;
-import org.apache.commons.lang3.StringUtils;
+import org.apache.eventmesh.runtime.core.protocol.amqp.exception.AmqpNotFoundException;
 
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
 
 /**
  * randomly select channel
@@ -19,9 +16,11 @@ public class RandomDispatchStrategy implements DownstreamDispatchStrategy {
     @Override
     public AmqpConsumer select(String queue, QueueConsumerMapping queueConsumerMapping) {
 
-        Set<AmqpConsumer> consumers = queueConsumerMapping.getConsumers(null, queue);
-        if (consumers == null || consumers.size() <= 0) {
-            return null;
+        Set<AmqpConsumer> consumers = null;
+        try {
+            consumers = queueConsumerMapping.getConsumers(null, queue);
+        } catch (AmqpNotFoundException e) {
+            throw new RuntimeException(e);
         }
 
         List<AmqpConsumer> amqpConsumerList = new ArrayList<>(consumers);