Merge branch '4.x.x-stable' into 5.1.x-stable

Author: acogoluegnes

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

@@ -209,6 +209,11 @@ public void publish(AMQP.BasicProperties props, byte[] message)
     }
 
     public Response doCall(AMQP.BasicProperties props, byte[] message)
+        throws IOException, TimeoutException {
+        return doCall(props, message, _timeout);
+    }
+
+    public Response doCall(AMQP.BasicProperties props, byte[] message, int timeout)
         throws IOException, ShutdownSignalException, TimeoutException {
         checkConsumer();
         BlockingCell<Object> k = new BlockingCell<Object>();
@@ -220,7 +225,7 @@ public Response doCall(AMQP.BasicProperties props, byte[] message)
             _continuationMap.put(replyId, k);
         }
         publish(props, message);
-        Object reply = k.uninterruptibleGet(_timeout);
+        Object reply = k.uninterruptibleGet(timeout);
         if (reply instanceof ShutdownSignalException) {
             ShutdownSignalException sig = (ShutdownSignalException) reply;
             ShutdownSignalException wrapper =
@@ -238,7 +243,13 @@ public Response doCall(AMQP.BasicProperties props, byte[] message)
     public byte[] primitiveCall(AMQP.BasicProperties props, byte[] message)
         throws IOException, ShutdownSignalException, TimeoutException
     {
-        return doCall(props, message).getBody();
+        return primitiveCall(props, message, _timeout);
+    }
+
+    public byte[] primitiveCall(AMQP.BasicProperties props, byte[] message, int timeout)
+        throws IOException, ShutdownSignalException, TimeoutException
+    {
+        return doCall(props, message, timeout).getBody();
     }
 
     /**
@@ -266,7 +277,23 @@ public byte[] primitiveCall(byte[] message)
      * @throws TimeoutException if a response is not received within the configured timeout
      */
     public Response responseCall(byte[] message) throws IOException, ShutdownSignalException, TimeoutException {
-        return doCall(null, message);
+        return responseCall(message, _timeout);
+    }
+
+    /**
+     * Perform a simple byte-array-based RPC roundtrip
+     *
+     * Useful if you need to get at more than just the body of the message
+     *
+     * @param message the byte array request message to send
+     * @param timeout milliseconds before timing out on wait for response
+     * @return The response object is an envelope that contains all of the data provided to the `handleDelivery` consumer
+     * @throws ShutdownSignalException if the connection dies during our wait
+     * @throws IOException if an error is encountered
+     * @throws TimeoutException if a response is not received within the configured timeout
+     */
+    public Response responseCall(byte[] message, int timeout) throws IOException, ShutdownSignalException, TimeoutException {
+        return doCall(null, message, timeout);
     }
 
     /**

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

@@ -768,8 +768,8 @@ public void automaticallyRecover(AutorecoveringConnection connection, Connection
         final RecoveryAwareChannelN newChannel = (RecoveryAwareChannelN) connDelegate.createChannel(this.getChannelNumber());
         if (newChannel == null)
             throw new IOException("Failed to create new channel for channel number=" + this.getChannelNumber() + " during recovery");
+        newChannel.inheritOffsetFrom(defunctChannel);
         this.delegate = newChannel;
-        this.delegate.inheritOffsetFrom(defunctChannel);
 
         this.notifyRecoveryListenersStarted();
         this.recoverShutdownListeners();

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

@@ -779,6 +779,28 @@ void deleteRecordedQueue(String queue) {
             this.maybeDeleteRecordedAutoDeleteExchange(b.getSource());
         }
     }
+    
+    /**
+     * Exclude the queue from the list of queues to recover after connection failure.
+     * Intended to be used in usecases where you want to remove the queue from this connection's recovery list but don't want to delete the queue from the server.
+     * 
+     * @param queue queue name to exclude from recorded recovery queues
+     * @param ifUnused if true, the RecordedQueue will only be excluded if no local consumers are using it.
+     */
+    public void excludeQueueFromRecovery(final String queue, final boolean ifUnused) {
+        if (ifUnused) {
+            // Note: This is basically the same as maybeDeleteRecordedAutoDeleteQueue except it works for non auto-delete queues as well.
+            synchronized (this.consumers) {
+                synchronized (this.recordedQueues) {
+                    if (!hasMoreConsumersOnQueue(this.consumers.values(), queue)) {
+                        deleteRecordedQueue(queue);
+                    }
+                }
+            }
+        } else {
+            deleteRecordedQueue(queue);
+        }
+    }
 
     void recordExchange(String exchange, RecordedExchange x) {
         this.recordedExchanges.put(exchange, x);
@@ -807,7 +829,7 @@ void maybeDeleteRecordedAutoDeleteQueue(String queue) {
                     RecordedQueue q = this.recordedQueues.get(queue);
                     // last consumer on this connection is gone, remove recorded queue
                     // if it is auto-deleted. See bug 26364.
-                    if((q != null) && q.isAutoDelete()) {
+                    if(q != null && q.isAutoDelete()) {
                         deleteRecordedQueue(queue);
                     }
                 }
@@ -822,7 +844,7 @@ void maybeDeleteRecordedAutoDeleteExchange(String exchange) {
                     RecordedExchange x = this.recordedExchanges.get(exchange);
                     // last binding where this exchange is the source is gone, remove recorded exchange
                     // if it is auto-deleted. See bug 26364.
-                    if((x != null) && x.isAutoDelete()) {
+                    if(x != null && x.isAutoDelete()) {
                         deleteRecordedExchange(exchange);
                     }
                 }

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

@@ -34,8 +34,8 @@
  * @since 3.3.0
  */
 public class RecoveryAwareChannelN extends ChannelN {
-    private long maxSeenDeliveryTag = 0;
-    private long activeDeliveryTagOffset = 0;
+    private volatile long maxSeenDeliveryTag = 0;
+    private volatile long activeDeliveryTagOffset = 0;
 
     /**
      * Construct a new channel on the given connection with the given
@@ -83,21 +83,19 @@ private AMQImpl.Basic.Deliver offsetDeliveryTag(AMQImpl.Basic.Deliver method) {
 
     @Override
     public void basicAck(long deliveryTag, boolean multiple) throws IOException {
-        // FIXME no check if deliveryTag = 0 (ack all)
         long realTag = deliveryTag - activeDeliveryTagOffset;
-        // 0 tag means ack all
-        if (realTag >= 0) {
+        // 0 tag means ack all when multiple is set
+        if (realTag > 0 || (multiple && realTag == 0)) {
             transmit(new Basic.Ack(realTag, multiple));
             metricsCollector.basicAck(this, deliveryTag, multiple);
         }
     }
 
     @Override
     public void basicNack(long deliveryTag, boolean multiple, boolean requeue) throws IOException {
-        // FIXME no check if deliveryTag = 0 (nack all)
         long realTag = deliveryTag - activeDeliveryTagOffset;
-        // 0 tag means nack all
-        if (realTag >= 0) {
+        // 0 tag means nack all when multiple is set
+        if (realTag > 0 || (multiple && realTag == 0)) {
             transmit(new Basic.Nack(realTag, multiple, requeue));
             metricsCollector.basicNack(this, deliveryTag);
         }

src/test/java/com/rabbitmq/client/test/functional/ConnectionRecovery.java

@@ -559,6 +559,28 @@ public void queueRecovered(String oldName, String newName) {
             // expected
         }
     }
+    
+    @Test public void thatExcludedQueueDoesNotReappearOnRecover() throws IOException, InterruptedException {
+        final String q = "java-client.test.recovery.excludedQueue1";
+        channel.queueDeclare(q, true, false, false, null);
+        // now delete it using the delegate so AutorecoveringConnection and AutorecoveringChannel are not aware of it
+        ((AutorecoveringChannel)channel).getDelegate().queueDelete(q);
+        assertNotNull(((AutorecoveringConnection)connection).getRecordedQueues().get(q));
+        // exclude the queue from recovery
+        ((AutorecoveringConnection)connection).excludeQueueFromRecovery(q, true);
+        // verify its not there
+        assertNull(((AutorecoveringConnection)connection).getRecordedQueues().get(q));
+        // reconnect
+        closeAndWaitForRecovery();
+        expectChannelRecovery(channel);
+        // verify queue was not recreated
+        try {
+            channel.queueDeclarePassive(q);
+            fail("Expected passive declare to fail");
+        } catch (IOException ioe) {
+            // expected
+        }
+    }
 
     @Test public void thatCancelledConsumerDoesNotReappearOnRecover() throws IOException, InterruptedException {
         String q = UUID.randomUUID().toString();