Merge branch '5.x.x-stable'

Author: michaelklishin

LICENSE-MPL-RabbitMQ

@@ -364,4 +364,4 @@ file, then You may include the notice in a location (such as a LICENSE
 file in a relevant directory) where a recipient would be likely to look
 for such a notice.
 
-Copyright (c) 2007-2020 VMware, Inc. or its affiliates.
+Copyright (c) 2007-2021 VMware, Inc. or its affiliates.

RUNNING_TESTS.md

@@ -107,4 +107,4 @@ Launch the tests:
 ```
 
 Note the `rabbitmqctl.bin` system property uses the syntax
-`DOCKER:{containerId}`.
+`DOCKER:{containerId}`.

src/main/java/com/rabbitmq/client/ConnectionFactory.java

@@ -19,8 +19,11 @@
 import com.rabbitmq.client.impl.nio.NioParams;
 import com.rabbitmq.client.impl.nio.SocketChannelFrameHandlerFactory;
 import com.rabbitmq.client.impl.recovery.AutorecoveringConnection;
+import com.rabbitmq.client.impl.recovery.RecoveredQueueNameSupplier;
 import com.rabbitmq.client.impl.recovery.RetryHandler;
 import com.rabbitmq.client.impl.recovery.TopologyRecoveryFilter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.Map.Entry;
 import java.util.function.BiConsumer;
@@ -49,6 +52,7 @@
  */
 public class ConnectionFactory implements Cloneable {
 
+    private static final Logger LOGGER = LoggerFactory.getLogger(ConnectionFactory.class);
     private static final int MAX_UNSIGNED_SHORT = 65535;
 
     /** Default user name */
@@ -186,6 +190,7 @@ public class ConnectionFactory implements Cloneable {
      * @since 5.4.0
      */
     private RetryHandler topologyRecoveryRetryHandler;
+    private RecoveredQueueNameSupplier recoveredQueueNameSupplier;
 
     /**
      * Traffic listener notified of inbound and outbound {@link Command}s.
@@ -1261,6 +1266,7 @@ public ConnectionParams params(ExecutorService consumerWorkServiceExecutor) {
         result.setTopologyRecoveryFilter(topologyRecoveryFilter);
         result.setConnectionRecoveryTriggeringCondition(connectionRecoveryTriggeringCondition);
         result.setTopologyRecoveryRetryHandler(topologyRecoveryRetryHandler);
+        result.setRecoveredQueueNameSupplier(recoveredQueueNameSupplier);
         result.setTrafficListener(trafficListener);
         result.setCredentialsRefreshService(credentialsRefreshService);
         return result;
@@ -1639,6 +1645,15 @@ public void setTopologyRecoveryRetryHandler(RetryHandler topologyRecoveryRetryHa
         this.topologyRecoveryRetryHandler = topologyRecoveryRetryHandler;
     }
 
+    /**
+     * Set the recovered queue name supplier. Default is use the same queue name when recovering queues.
+     * 
+     * @param recoveredQueueNameSupplier queue name supplier
+     */
+    public void setRecoveredQueueNameSupplier(RecoveredQueueNameSupplier recoveredQueueNameSupplier) {
+        this.recoveredQueueNameSupplier = recoveredQueueNameSupplier;
+    }
+
     /**
      * Traffic listener notified of inbound and outbound {@link Command}s.
      * <p>

src/main/java/com/rabbitmq/client/MetricsCollector.java

@@ -36,13 +36,21 @@ public interface MetricsCollector {
 
     void basicPublish(Channel channel, long deliveryTag);
 
-    void basicPublishFailure(Channel channel, Throwable cause);
+    default void basicPublishFailure(Channel channel, Throwable cause) {
 
-    void basicPublishAck(Channel channel, long deliveryTag, boolean multiple);
+    }
 
-    void basicPublishNack(Channel channel, long deliveryTag, boolean multiple);
+    default void basicPublishAck(Channel channel, long deliveryTag, boolean multiple) {
 
-    void basicPublishUnrouted(Channel channel);
+    }
+
+    default void basicPublishNack(Channel channel, long deliveryTag, boolean multiple) {
+
+    }
+
+    default void basicPublishUnrouted(Channel channel) {
+
+    }
 
     void consumedMessage(Channel channel, long deliveryTag, boolean autoAck);
 

src/main/java/com/rabbitmq/client/impl/AMQConnection.java

@@ -397,35 +397,35 @@ public void start()
         }
 
         try {
-            int channelMax =
+            int negotiatedChannelMax =
                 negotiateChannelMax(this.requestedChannelMax,
                                     connTune.getChannelMax());
 
-            if (!checkUnsignedShort(channelMax)) {
-                throw new IllegalArgumentException("Negotiated channel max must be between 0 and " + MAX_UNSIGNED_SHORT + ": " + channelMax);
+            if (!checkUnsignedShort(negotiatedChannelMax)) {
+                throw new IllegalArgumentException("Negotiated channel max must be between 0 and " + MAX_UNSIGNED_SHORT + ": " + negotiatedChannelMax);
             }
 
-            _channelManager = instantiateChannelManager(channelMax, threadFactory);
+            _channelManager = instantiateChannelManager(negotiatedChannelMax, threadFactory);
 
             int frameMax =
                 negotiatedMaxValue(this.requestedFrameMax,
                                    connTune.getFrameMax());
             this._frameMax = frameMax;
 
-            int heartbeat =
+            int negotiatedHeartbeat =
                 negotiatedMaxValue(this.requestedHeartbeat,
                                    connTune.getHeartbeat());
 
-            if (!checkUnsignedShort(heartbeat)) {
-                throw new IllegalArgumentException("Negotiated heartbeat must be between 0 and " + MAX_UNSIGNED_SHORT + ": " + heartbeat);
+            if (!checkUnsignedShort(negotiatedHeartbeat)) {
+                throw new IllegalArgumentException("Negotiated heartbeat must be between 0 and " + MAX_UNSIGNED_SHORT + ": " + negotiatedHeartbeat);
             }
 
-            setHeartbeat(heartbeat);
+            setHeartbeat(negotiatedHeartbeat);
 
             _channel0.transmit(new AMQP.Connection.TuneOk.Builder()
-                                .channelMax(channelMax)
+                                .channelMax(negotiatedChannelMax)
                                 .frameMax(frameMax)
-                                .heartbeat(heartbeat)
+                                .heartbeat(negotiatedHeartbeat)
                               .build());
             _channel0.exnWrappingRpc(new AMQP.Connection.Open.Builder()
                                       .virtualHost(_virtualHost)

src/main/java/com/rabbitmq/client/impl/AbstractMetricsCollector.java

@@ -151,9 +151,9 @@ public void basicPublishNack(Channel channel, long deliveryTag, boolean multiple
     @Override
     public void basicPublishUnrouted(Channel channel) {
         try {
-            markPublishedMessageNotRouted();
+            markPublishedMessageUnrouted();
         } catch(Exception e) {
-            LOGGER.info("Error while computing metrics in markPublishedMessageNotRouted: " + e.getMessage());
+            LOGGER.info("Error while computing metrics in markPublishedMessageUnrouted: " + e.getMessage());
         }
     }
 
@@ -420,5 +420,5 @@ private ChannelState(Channel channel) {
     /**
      * Marks the event of a published message not being routed.
      */
-    protected abstract void markPublishedMessageNotRouted();
+    protected abstract void markPublishedMessageUnrouted();
 }

src/main/java/com/rabbitmq/client/impl/ConnectionParams.java

@@ -21,6 +21,7 @@
 import com.rabbitmq.client.SaslConfig;
 import com.rabbitmq.client.ShutdownSignalException;
 import com.rabbitmq.client.TrafficListener;
+import com.rabbitmq.client.impl.recovery.RecoveredQueueNameSupplier;
 import com.rabbitmq.client.impl.recovery.RetryHandler;
 import com.rabbitmq.client.impl.recovery.TopologyRecoveryFilter;
 
@@ -54,7 +55,7 @@ public class ConnectionParams {
     private TopologyRecoveryFilter topologyRecoveryFilter;
     private Predicate<ShutdownSignalException> connectionRecoveryTriggeringCondition;
     private RetryHandler topologyRecoveryRetryHandler;
-
+    private RecoveredQueueNameSupplier recoveredQueueNameSupplier;
     private ExceptionHandler exceptionHandler;
     private ThreadFactory threadFactory;
 
@@ -272,6 +273,14 @@ public RetryHandler getTopologyRecoveryRetryHandler() {
         return topologyRecoveryRetryHandler;
     }
 
+    public void setRecoveredQueueNameSupplier(RecoveredQueueNameSupplier recoveredQueueNameSupplier) {
+        this.recoveredQueueNameSupplier = recoveredQueueNameSupplier;
+    }
+
+    public RecoveredQueueNameSupplier getRecoveredQueueNameSupplier() {
+        return recoveredQueueNameSupplier;
+    }
+
     public void setTrafficListener(TrafficListener trafficListener) {
         this.trafficListener = trafficListener;
     }

src/main/java/com/rabbitmq/client/impl/MicrometerMetricsCollector.java

@@ -148,7 +148,7 @@ protected void markMessagePublishNotAcknowledged() {
     }
 
     @Override
-    protected void markPublishedMessageNotRouted() {
+    protected void markPublishedMessageUnrouted() {
         unroutedPublishedMessages.increment();
     }
 

src/main/java/com/rabbitmq/client/impl/OAuth2ClientCredentialsGrantCredentialsProvider.java

@@ -595,5 +595,4 @@ private SSLSocketFactory sslSocketFactory() {
         }
 
     }
-
 }

src/main/java/com/rabbitmq/client/impl/StandardMetricsCollector.java

@@ -125,7 +125,7 @@ protected void markMessagePublishNotAcknowledged() {
     }
 
     @Override
-    protected void markPublishedMessageNotRouted() {
+    protected void markPublishedMessageUnrouted() {
         publishUnroutedMessages.mark();
     }
 

src/main/java/com/rabbitmq/client/impl/recovery/AutorecoveringChannel.java

@@ -355,7 +355,8 @@ public void queueDeclareNoWait(String queue,
             durable(durable).
             exclusive(exclusive).
             autoDelete(autoDelete).
-            arguments(arguments);
+            arguments(arguments).
+            recoveredQueueNameSupplier(connection.getRecoveredQueueNameSupplier());
         delegate.queueDeclareNoWait(queue, durable, exclusive, autoDelete, arguments);
         recordQueue(queue, meta);
 
@@ -848,7 +849,8 @@ private void recordQueue(AMQP.Queue.DeclareOk ok, String queue, boolean durable,
                 durable(durable).
                 exclusive(exclusive).
                 autoDelete(autoDelete).
-                arguments(arguments);
+                arguments(arguments).
+                recoveredQueueNameSupplier(connection.getRecoveredQueueNameSupplier());
         if (queue.equals(RecordedQueue.EMPTY_STRING)) {
             q.serverNamed(true);
         }