#354 | call metrics (acks and nacks have to be corrected for multiples)

Author: slayful

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

@@ -38,9 +38,9 @@ public interface MetricsCollector {
 
     void basicPublishFailure(Channel channel, Throwable cause);
 
-    void basicPublishAck(Channel channel);
+    void basicPublishAck(Channel channel, long deliveryTag, boolean multiple);
 
-    void basicPublishNack(Channel channel);
+    void basicPublishNack(Channel channel, long deliveryTag, boolean multiple);
 
     void basicPublishUnrouted(Channel channel);
 

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

@@ -76,12 +76,12 @@ public void basicPublishFailure(Channel channel, Throwable cause) {
     }
 
     @Override
-    public void basicPublishAck(Channel channel) {
+    public void basicPublishAck(Channel channel, long deliveryTag, boolean multiple) {
 
     }
 
     @Override
-    public void basicPublishNack(Channel channel) {
+    public void basicPublishNack(Channel channel, long deliveryTag, boolean multiple) {
 
     }
 

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

@@ -112,7 +112,7 @@ public void basicPublishFailure(Channel channel, Throwable cause) {
     }
 
     @Override
-    public void basicPublishAck(Channel channel) {
+    public void basicPublishAck(Channel channel, long deliveryTag, boolean multiple) {
         try {
             markMessagePublishAcknowledged();
         } catch(Exception e) {
@@ -121,7 +121,7 @@ public void basicPublishAck(Channel channel) {
     }
 
     @Override
-    public void basicPublishNack(Channel channel) {
+    public void basicPublishNack(Channel channel, long deliveryTag, boolean multiple) {
         try {
             markMessagePublishNotAcknowledged();
         } catch(Exception e) {

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

@@ -481,6 +481,8 @@ private void callReturnListeners(Command command, Basic.Return basicReturn) {
             }
         } catch (Throwable ex) {
             getConnection().getExceptionHandler().handleReturnListenerException(this, ex);
+        } finally {
+            metricsCollector.basicPublishUnrouted(this);
         }
     }
 
@@ -491,6 +493,8 @@ private void callConfirmListeners(@SuppressWarnings("unused") Command command, B
             }
         } catch (Throwable ex) {
             getConnection().getExceptionHandler().handleConfirmListenerException(this, ex);
+        } finally {
+            metricsCollector.basicPublishAck(this, ack.getDeliveryTag(), ack.getMultiple());
         }
     }
 
@@ -501,6 +505,8 @@ private void callConfirmListeners(@SuppressWarnings("unused") Command command, B
             }
         } catch (Throwable ex) {
             getConnection().getExceptionHandler().handleConfirmListenerException(this, ex);
+        } finally {
+            metricsCollector.basicPublishNack(this, nack.getDeliveryTag(), nack.getMultiple());
         }
     }