bugfix: Default exchange consume failed, when broker restart. (#597)

streamnative · Oct 17, 2022 · 74d7265 · 74d7265
1 parent cc78180
commit 74d7265
Show file tree

Hide file tree

Showing 3 changed files with 20 additions and 1 deletion.
diff --git a/amqp-impl/src/main/java/io/streamnative/pulsar/handlers/amqp/AbstractAmqpExchange.java b/amqp-impl/src/main/java/io/streamnative/pulsar/handlers/amqp/AbstractAmqpExchange.java
@@ -50,6 +50,17 @@ public CompletableFuture<Void> addQueue(AmqpQueue queue) {
         return CompletableFuture.completedFuture(null);
     }
 
+    @Override
+    public AmqpQueue getQueue(String queueName) {
+        AmqpQueue queue = null;
+        for (AmqpQueue q : queues) {
+            if (q.getName().equals(queueName)) {
+                queue = q;
+            }
+        }
+        return queue;
+    }
+
     @Override
     public void removeQueue(AmqpQueue queue) {
         queues.remove(queue);

diff --git a/amqp-impl/src/main/java/io/streamnative/pulsar/handlers/amqp/AmqpChannel.java b/amqp-impl/src/main/java/io/streamnative/pulsar/handlers/amqp/AmqpChannel.java
@@ -502,7 +502,8 @@ public void receiveBasicPublish(AMQShortString exchange, AMQShortString routingK
                                 return;
                             } else {
                                 // bind to default exchange.
-                                if (amqpQueue.getRouter(AbstractAmqpExchange.DEFAULT_EXCHANGE_DURABLE) == null) {
+                                if (amqpQueue.getRouter(AbstractAmqpExchange.DEFAULT_EXCHANGE_DURABLE) == null
+                                        || amqpExchange.getQueue(queueName) == null) {
                                     amqpQueue.bindExchange(amqpExchange,
                                             AbstractAmqpMessageRouter.generateRouter(AmqpExchange.Type.Direct),
                                             routingKey.toString(), null);

diff --git a/amqp-impl/src/main/java/io/streamnative/pulsar/handlers/amqp/AmqpExchange.java b/amqp-impl/src/main/java/io/streamnative/pulsar/handlers/amqp/AmqpExchange.java
@@ -138,6 +138,13 @@ public static Type value(String type) {
      */
     CompletableFuture<Void> addQueue(AmqpQueue queue);
 
+    /**
+     * Get a queue {@link AmqpQueue} with queue name.
+     * @param queueName AMQP queue name.
+     * @return AMQP queue.
+     */
+    AmqpQueue getQueue(String queueName);
+
     /**
      * Remove a queue {@link AmqpQueue} from the exchange.
      * @param queue AMQP queue.