Fix recovery channel metrics that are sent with realTag without offset

Author: tsotnikov

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

@@ -93,7 +93,7 @@ public class ChannelN extends AMQChannel implements com.rabbitmq.client.Channel
     /** Whether any nacks have been received since the last waitForConfirms(). */
     private volatile boolean onlyAcksReceived = true;
 
-    private final MetricsCollector metricsCollector;
+    protected final MetricsCollector metricsCollector;
 
     /**
      * Construct a new channel on the given connection with the given

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

@@ -22,6 +22,7 @@
 import com.rabbitmq.client.impl.AMQImpl;
 import com.rabbitmq.client.impl.ChannelN;
 import com.rabbitmq.client.impl.ConsumerWorkService;
+import com.rabbitmq.client.impl.AMQImpl.Basic;
 
 import java.io.IOException;
 
@@ -86,7 +87,8 @@ public void basicAck(long deliveryTag, boolean multiple) throws IOException {
         long realTag = deliveryTag - activeDeliveryTagOffset;
         // 0 tag means ack all
         if (realTag >= 0) {
-            super.basicAck(realTag, multiple);
+            transmit(new Basic.Ack(deliveryTag, multiple));
+            metricsCollector.basicAck(this, deliveryTag, multiple);
         }
     }
 
@@ -96,15 +98,17 @@ public void basicNack(long deliveryTag, boolean multiple, boolean requeue) throw
         long realTag = deliveryTag - activeDeliveryTagOffset;
         // 0 tag means nack all
         if (realTag >= 0) {
-            super.basicNack(realTag, multiple, requeue);
+            transmit(new Basic.Nack(realTag, multiple, requeue));
+            metricsCollector.basicNack(this, deliveryTag);
         }
     }
 
     @Override
     public void basicReject(long deliveryTag, boolean requeue) throws IOException {
         long realTag = deliveryTag - activeDeliveryTagOffset;
         if (realTag > 0) {
-            super.basicReject(realTag, requeue);
+            transmit(new Basic.Reject(realTag, requeue));
+            metricsCollector.basicReject(this, deliveryTag);
         }
     }