Implement : Delay-Queue, Message-TTL, Queue-TTL, Dead-Letter-Queue, Multi-Tenant in AoP(Implementation based on PR#748)

amqp-client-auth/pom.xml

@@ -20,7 +20,7 @@
     <parent>
         <artifactId>pulsar-protocol-handler-amqp-parent</artifactId>
         <groupId>io.streamnative.pulsar.handlers</groupId>
-        <version>3.0.0.1-SNAPSHOT</version>
+        <version>2.11.0-SNAPSHOT</version>
     </parent>
 
     <modelVersion>4.0.0</modelVersion>

amqp-impl/pom.xml

@@ -22,7 +22,7 @@
   <parent>
     <groupId>io.streamnative.pulsar.handlers</groupId>
     <artifactId>pulsar-protocol-handler-amqp-parent</artifactId>
-    <version>3.0.0.1-SNAPSHOT</version>
+    <version>2.11.0-SNAPSHOT</version>
   </parent>
 
   <artifactId>pulsar-protocol-handler-amqp</artifactId>
@@ -31,12 +31,29 @@
 
   <!-- include the dependencies -->
   <dependencies>
+
     <dependency>
-      <groupId>com.rabbitmq</groupId>
-      <artifactId>amqp-client</artifactId>
-      <version>${rabbitmq.version}</version>
-      <scope>test</scope>
+      <groupId>org.apache.qpid</groupId>
+      <artifactId>qpid-broker-core</artifactId>
+      <version>${qpid-protocol-plugin.version}</version>
+<!--      <exclusions>-->
+<!--        <exclusion>-->
+<!--          <groupId>com.fasterxml.jackson.core</groupId>-->
+<!--          <artifactId>jackson-core</artifactId>-->
+<!--        </exclusion>-->
+<!--        <exclusion>-->
+<!--          <groupId>com.fasterxml.jackson.core</groupId>-->
+<!--          <artifactId>jackson-databind</artifactId>-->
+<!--        </exclusion>-->
+<!--      </exclusions>-->
     </dependency>
+
+    <dependency>
+      <groupId>org.apache.qpid</groupId>
+      <artifactId>qpid-broker-plugins-amqp-0-8-protocol</artifactId>
+      <version>${qpid-protocol-plugin.version}</version>
+    </dependency>
+
   </dependencies>
 
   <build>

amqp-impl/src/main/java/io/streamnative/pulsar/handlers/amqp/AbstractAmqpExchange.java

@@ -18,12 +18,15 @@
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
+import lombok.Getter;
 
 /**
  * Base class of AMQP exchange.
  */
 public abstract class AbstractAmqpExchange implements AmqpExchange {
 
+    @Getter
+    protected final Map<String, String> properties;
     protected final String exchangeName;
     protected final AmqpExchange.Type exchangeType;
     protected Set<AmqpQueue> queues;
@@ -37,7 +40,8 @@ public abstract class AbstractAmqpExchange implements AmqpExchange {
 
     protected AbstractAmqpExchange(String exchangeName, AmqpExchange.Type exchangeType,
                                    Set<AmqpQueue> queues, boolean durable, boolean autoDelete, boolean internal,
-                                   Map<String, Object> arguments) {
+                                   Map<String, Object> arguments, Map<String, String> properties) {
+        this.properties = properties;
         this.exchangeName = exchangeName;
         this.exchangeType = exchangeType;
         this.queues = queues;

amqp-impl/src/main/java/io/streamnative/pulsar/handlers/amqp/AbstractAmqpQueue.java

@@ -14,10 +14,12 @@
 package io.streamnative.pulsar.handlers.amqp;
 
 import java.util.Collection;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
+import lombok.Getter;
 
 /**
  * Base class for AMQP queue.
@@ -31,6 +33,9 @@ public abstract class AbstractAmqpQueue implements AmqpQueue {
     protected boolean exclusive;
     protected boolean autoDelete;
     protected final Map<String, AmqpMessageRouter> routers = new ConcurrentHashMap<>();
+    protected final Map<String, Object> arguments = new HashMap<>();
+    @Getter
+    protected Map<String, String> properties;
 
     protected AbstractAmqpQueue(String queueName, boolean durable, long connectionId) {
         this.queueName = queueName;
@@ -42,12 +47,13 @@ protected AbstractAmqpQueue(String queueName, boolean durable, long connectionId
 
     protected AbstractAmqpQueue(String queueName,
                                 boolean durable, long connectionId,
-                                boolean exclusive, boolean autoDelete) {
+                                boolean exclusive, boolean autoDelete, Map<String, String> properties) {
         this.queueName = queueName;
         this.durable = durable;
         this.connectionId = connectionId;
         this.exclusive = exclusive;
         this.autoDelete = autoDelete;
+        this.properties = properties;
     }
 
     @Override
@@ -60,6 +66,20 @@ public boolean getDurable() {
         return durable;
     }
 
+    @Override
+    public boolean getExclusive() {
+        return exclusive;
+    }
+    @Override
+    public boolean getAutoDelete() {
+        return autoDelete;
+    }
+
+    @Override
+    public Map<String, Object> getArguments() {
+        return arguments;
+    }
+
     @Override
     public AmqpMessageRouter getRouter(String exchangeName) {
         return routers.get(exchangeName);
@@ -133,4 +153,8 @@ public boolean isAutoDelete() {
         return autoDelete;
     }
 
+    @Override
+    public void close() {
+
+    }
 }

amqp-impl/src/main/java/io/streamnative/pulsar/handlers/amqp/AmqpBrokerService.java

@@ -15,7 +15,9 @@
 package io.streamnative.pulsar.handlers.amqp;
 
 import io.netty.util.concurrent.DefaultThreadFactory;
-import java.util.concurrent.Executor;
+import io.streamnative.pulsar.handlers.amqp.admin.AmqpAdmin;
+import io.streamnative.pulsar.handlers.amqp.admin.prometheus.PrometheusAdmin;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import lombok.Getter;
 import org.apache.pulsar.broker.PulsarService;
@@ -39,19 +41,26 @@ public class AmqpBrokerService {
     private ConnectionContainer connectionContainer;
     @Getter
     private PulsarService pulsarService;
+    @Getter
+    private AmqpAdmin amqpAdmin;
+    @Getter
+    PrometheusAdmin prometheusAdmin;
 
     public AmqpBrokerService(PulsarService pulsarService, AmqpServiceConfiguration config) {
+        String clusterName = pulsarService.getBrokerService().getPulsar().getConfiguration().getClusterName();
+        this.amqpAdmin = new AmqpAdmin(config.getAdvertisedAddress(), config.getAmqpAdminPort());
+        this.prometheusAdmin = new PrometheusAdmin(config.getAmqpPrometheusUrl(), clusterName);
         this.pulsarService = pulsarService;
         this.amqpTopicManager = new AmqpTopicManager(pulsarService);
-        this.exchangeContainer = new ExchangeContainer(amqpTopicManager, pulsarService, initRouteExecutor(config),
-                config.getAmqpExchangeRouteQueueSize());
-        this.queueContainer = new QueueContainer(amqpTopicManager, pulsarService, exchangeContainer);
+        this.exchangeContainer = new ExchangeContainer(amqpTopicManager, pulsarService,
+                initRouteExecutor(config), config, amqpAdmin);
+        this.queueContainer = new QueueContainer(amqpTopicManager, pulsarService, exchangeContainer, config);
         this.exchangeService = new ExchangeServiceImpl(exchangeContainer);
-        this.queueService = new QueueServiceImpl(exchangeContainer, queueContainer);
+        this.queueService = new QueueServiceImpl(exchangeContainer, queueContainer, amqpTopicManager);
         this.connectionContainer = new ConnectionContainer(pulsarService, exchangeContainer, queueContainer);
     }
 
-    private Executor initRouteExecutor(AmqpServiceConfiguration config) {
+    private ExecutorService initRouteExecutor(AmqpServiceConfiguration config) {
         return Executors.newFixedThreadPool(config.getAmqpExchangeRouteExecutorThreads(),
                 new DefaultThreadFactory("exchange-route"));
     }

amqp-impl/src/main/java/io/streamnative/pulsar/handlers/amqp/AmqpChannel.java

@@ -83,56 +83,56 @@
 @Log4j2
 public class AmqpChannel implements ServerChannelMethodProcessor {
 
-    private final int channelId;
-    private final AmqpConnection connection;
+    protected final int channelId;
+    protected final AmqpConnection connection;
     private final AtomicBoolean blocking = new AtomicBoolean(false);
-    private final AtomicBoolean closing = new AtomicBoolean(false);
+    protected final AtomicBoolean closing = new AtomicBoolean(false);
     private final java.util.Queue<AsyncCommand> unfinishedCommandsQueue = new ConcurrentLinkedQueue<>();
-    private long confirmedMessageCounter;
+    protected long confirmedMessageCounter;
     private volatile ServerTransaction transaction;
-    private boolean confirmOnPublish;
+    protected boolean confirmOnPublish;
     /** A channel has a default queue (the last declared) that is used when no queue name is explicitly set. */
-    private volatile AmqpQueue defaultQueue;
+    protected volatile AmqpQueue defaultQueue;
 
-    private final UnacknowledgedMessageMap unacknowledgedMessageMap;
+    protected final UnacknowledgedMessageMap unacknowledgedMessageMap;
 
     /** Maps from consumer tag to consumers instance. */
-    private final Map<String, Consumer> tag2ConsumersMap = new ConcurrentHashMap<>();
+    protected final Map<String, Consumer> tag2ConsumersMap = new ConcurrentHashMap<>();
 
-    private final Map<String, AmqpConsumer> fetchConsumerMap = new ConcurrentHashMap<>();
+    protected final Map<String, AmqpConsumer> fetchConsumerMap = new ConcurrentHashMap<>();
 
     /**
      * The current message - which may be partial in the sense that not all frames have been received yet - which has
      * been received by this channel. As the frames are received the message gets updated and once all frames have been
      * received the message can then be routed.
      */
-    private IncomingMessage currentMessage;
+    protected IncomingMessage currentMessage;
 
-    private final String defaultSubscription = "defaultSubscription";
+    protected final String defaultSubscription = "defaultSubscription";
     public static final AMQShortString EMPTY_STRING = AMQShortString.createAMQShortString((String) null);
     /**
      * ConsumerTag prefix, the tag is unique per subscription to a queue.
      * The server returns this in response to a basic.consume request.
      */
-    private static final String CONSUMER_TAG_PREFIX = "aop.ctag-";
+    protected static final String CONSUMER_TAG_PREFIX = "aop.ctag-";
 
     /**
      * The consumer ID.
      */
-    private static final AtomicLong CONSUMER_ID = new AtomicLong(0);
+    protected static final AtomicLong CONSUMER_ID = new AtomicLong(0);
 
     /**
      * The delivery tag is unique per channel. This is pre-incremented before putting into the deliver frame so that
      * value of this represents the <b>last</b> tag sent out.
      */
-    private volatile long deliveryTag = 0;
-    private final AmqpFlowCreditManager creditManager;
-    private final AtomicBoolean blockedOnCredit = new AtomicBoolean(false);
+    protected volatile long deliveryTag = 0;
+    protected final AmqpFlowCreditManager creditManager;
+    protected final AtomicBoolean blockedOnCredit = new AtomicBoolean(false);
     public static final int DEFAULT_CONSUMER_PERMIT = 1000;
-    private ExchangeService exchangeService;
-    private QueueService queueService;
-    private ExchangeContainer exchangeContainer;
-    private QueueContainer queueContainer;
+    protected ExchangeService exchangeService;
+    protected QueueService queueService;
+    protected ExchangeContainer exchangeContainer;
+    protected QueueContainer queueContainer;
 
     public AmqpChannel(int channelId, AmqpConnection connection, AmqpBrokerService amqpBrokerService) {
         this.channelId = channelId;
@@ -253,7 +253,8 @@ public void receiveQueueDeclare(AMQShortString queue, boolean passive, boolean d
                     channelId, queue, passive, durable, exclusive, autoDelete, nowait, arguments);
         }
         queueService.queueDeclare(connection.getNamespaceName(), queue.toString(), passive, durable, exclusive,
-                autoDelete, nowait, arguments, connection.getConnectionId()).thenAccept(amqpQueue -> {
+                autoDelete, nowait, FieldTable.convertToMap(arguments), connection.getConnectionId())
+                .thenAccept(amqpQueue -> {
             setDefaultQueue(amqpQueue);
             MethodRegistry methodRegistry = connection.getMethodRegistry();
             QueueDeclareOkBody responseBody = methodRegistry.createQueueDeclareOkBody(
@@ -401,7 +402,7 @@ public void receiveBasicConsume(AMQShortString queue, AMQShortString consumerTag
         });
     }
 
-    private String getConsumerTag(AMQShortString consumerTag) {
+    protected String getConsumerTag(AMQShortString consumerTag) {
         if (consumerTag == null) {
             return CONSUMER_TAG_PREFIX + connection.remoteAddress + "-" + UUID.randomUUID();
         } else {
@@ -411,7 +412,7 @@ private String getConsumerTag(AMQShortString consumerTag) {
     }
 
     private synchronized void subscribe(String consumerTag, String queueName, Topic topic,
-                           boolean ack, boolean exclusive, boolean nowait){
+                           boolean ack, boolean exclusive, boolean nowait) {
 
         CompletableFuture<Void> future = new CompletableFuture<>();
         future.whenComplete((ignored, e) -> {
@@ -685,7 +686,7 @@ private void publishContentHeader(ContentHeaderBody contentHeaderBody) {
         deliverCurrentMessageIfComplete();
     }
 
-    private void deliverCurrentMessageIfComplete() {
+    protected void deliverCurrentMessageIfComplete() {
         if (currentMessage.allContentReceived()) {
             MessagePublishInfo info = currentMessage.getMessagePublishInfo();
             String routingKey = AMQShortString.toString(info.getRoutingKey());
@@ -750,6 +751,8 @@ public void messageNAck(long deliveryTag, boolean multiple, boolean requeue) {
         if (!ackedMessages.isEmpty()) {
             if (requeue) {
                 requeue(ackedMessages);
+            } else {
+                discardMessage(ackedMessages);
             }
         } else {
             closeChannel(ErrorCodes.IN_USE, "deliveryTag not found");
@@ -759,6 +762,17 @@ public void messageNAck(long deliveryTag, boolean multiple, boolean requeue) {
         }
     }
 
+    private void discardMessage(Collection<UnacknowledgedMessageMap.MessageConsumerAssociation> messages) {
+        Map<UnacknowledgedMessageMap.MessageProcessor, List<PositionImpl>> positionMap = new HashMap<>();
+        messages.forEach(association -> {
+            UnacknowledgedMessageMap.MessageProcessor consumer = association.getConsumer();
+            List<PositionImpl> positions = positionMap.computeIfAbsent(consumer,
+                    list -> new ArrayList<>());
+            positions.add((PositionImpl) association.getPosition());
+        });
+        positionMap.forEach(UnacknowledgedMessageMap.MessageProcessor::discardMessage);
+    }
+
     @Override
     public void receiveBasicRecover(boolean requeue, boolean sync) {
         Collection<UnacknowledgedMessageMap.MessageConsumerAssociation> ackedMessages =
@@ -777,15 +791,15 @@ public void receiveBasicRecover(boolean requeue, boolean sync) {
     }
 
     private void requeue(Collection<UnacknowledgedMessageMap.MessageConsumerAssociation> messages) {
-        Map<AmqpConsumer, List<PositionImpl>> positionMap = new HashMap<>();
+        Map<UnacknowledgedMessageMap.MessageProcessor, List<PositionImpl>> positionMap = new HashMap<>();
         messages.stream().forEach(association -> {
-            AmqpConsumer consumer = association.getConsumer();
+            UnacknowledgedMessageMap.MessageProcessor consumer = association.getConsumer();
             List<PositionImpl> positions = positionMap.computeIfAbsent(consumer,
                 list -> new ArrayList<>());
             positions.add((PositionImpl) association.getPosition());
         });
         positionMap.entrySet().stream().forEach(entry -> {
-            entry.getKey().redeliverAmqpMessages(entry.getValue());
+            entry.getKey().requeue(entry.getValue());
         });
     }
 
@@ -799,10 +813,10 @@ public void receiveBasicAck(long deliveryTag, boolean multiple) {
 
     private void messageAck(long deliveryTag, boolean multiple) {
         Collection<UnacknowledgedMessageMap.MessageConsumerAssociation> ackedMessages =
-            unacknowledgedMessageMap.acknowledge(deliveryTag, multiple);
+                unacknowledgedMessageMap.acknowledge(deliveryTag, multiple);
         if (!ackedMessages.isEmpty()) {
             ackedMessages.stream().forEach(entry -> {
-                entry.getConsumer().messagesAck(entry.getPosition());
+                entry.getConsumer().messageAck(entry.getPosition());
             });
         } else {
             closeChannel(ErrorCodes.IN_USE, "deliveryTag not found");
@@ -1012,7 +1026,7 @@ public AmqpFlowCreditManager getCreditManager() {
         return creditManager;
     }
 
-    private void handleAoPException(Throwable t) {
+    protected void handleAoPException(Throwable t) {
         Throwable cause = FutureUtil.unwrapCompletionException(t);
         if (!(cause instanceof AoPException)) {
             connection.sendConnectionClose(INTERNAL_ERROR, t.getMessage(), channelId);