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

Author: acogoluegnes

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

@@ -600,8 +600,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/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);
         }