#354 | add implementation for ack, nack and unrouted publishes

Author: slayful

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

@@ -38,6 +38,12 @@ public interface MetricsCollector {
 
     void basicPublishFailure(Channel channel, Throwable cause);
 
+    void basicPublishAck(Channel channel);
+
+    void basicPublishNack(Channel channel);
+
+    void basicPublishUnrouted(Channel channel);
+
     void consumedMessage(Channel channel, long deliveryTag, boolean autoAck);
 
     void consumedMessage(Channel channel, long deliveryTag, String consumerTag);

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

@@ -75,6 +75,21 @@ public void basicPublishFailure(Channel channel, Throwable cause) {
 
     }
 
+    @Override
+    public void basicPublishAck(Channel channel) {
+
+    }
+
+    @Override
+    public void basicPublishNack(Channel channel) {
+
+    }
+
+    @Override
+    public void basicPublishUnrouted(Channel channel) {
+
+    }
+
     @Override
     public void consumedMessage(Channel channel, long deliveryTag, boolean autoAck) {
 

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

@@ -111,6 +111,33 @@ public void basicPublishFailure(Channel channel, Throwable cause) {
         }
     }
 
+    @Override
+    public void basicPublishAck(Channel channel) {
+        try {
+            markMessagePublishAcknowledged();
+        } catch(Exception e) {
+            LOGGER.info("Error while computing metrics in basicPublishAck: " + e.getMessage());
+        }
+    }
+
+    @Override
+    public void basicPublishNack(Channel channel) {
+        try {
+            markMessagePublishNotAcknowledged();
+        } catch(Exception e) {
+            LOGGER.info("Error while computing metrics in basicPublishNack: " + e.getMessage());
+        }
+    }
+
+    @Override
+    public void basicPublishUnrouted(Channel channel) {
+        try {
+            markPublishedMessageNotRouted();
+        } catch(Exception e) {
+            LOGGER.info("Error while computing metrics in markPublishedMessageNotRouted: " + e.getMessage());
+        }
+    }
+
     @Override
     public void basicConsume(Channel channel, String consumerTag, boolean autoAck) {
         try {
@@ -360,6 +387,17 @@ private ChannelState(Channel channel) {
      */
     protected abstract void markRejectedMessage();
 
+    /**
+     * Marks the event of a message publishing acknowledgement.
+     */
+    protected abstract void markMessagePublishAcknowledged();
 
-
+    /**
+     * Marks the event of a message publishing not being acknowledged.
+     */
+    protected abstract void markMessagePublishNotAcknowledged();
+    /**
+     * Marks the event of a published message not being routed.
+     */
+    protected abstract void markPublishedMessageNotRouted();
 }

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

@@ -51,6 +51,12 @@ public class MicrometerMetricsCollector extends AbstractMetricsCollector {
 
     private final Counter failedToPublishMessages;
 
+    private final Counter ackedPublishedMessages;
+
+    private final Counter nackedPublishedMessages;
+
+    private final Counter unroutedPublishedMessages;
+
     private final Counter consumedMessages;
 
     private final Counter acknowledgedMessages;
@@ -81,6 +87,9 @@ public MicrometerMetricsCollector(Function<Metrics, Object> metricsCreator) {
         this.acknowledgedMessages = (Counter) metricsCreator.apply(ACKNOWLEDGED_MESSAGES);
         this.rejectedMessages = (Counter) metricsCreator.apply(REJECTED_MESSAGES);
         this.failedToPublishMessages = (Counter) metricsCreator.apply(FAILED_TO_PUBLISH_MESSAGES);
+        this.ackedPublishedMessages = (Counter) metricsCreator.apply(ACKED_PUBLISHED_MESSAGES);
+        this.nackedPublishedMessages = (Counter) metricsCreator.apply(NACKED_PUBLISHED_MESSAGES);
+        this.unroutedPublishedMessages = (Counter) metricsCreator.apply(UNROUTED_PUBLISHED_MESSAGES);
     }
 
     @Override
@@ -128,6 +137,21 @@ protected void markRejectedMessage() {
         rejectedMessages.increment();
     }
 
+    @Override
+    protected void markMessagePublishAcknowledged() {
+        ackedPublishedMessages.increment();
+    }
+
+    @Override
+    protected void markMessagePublishNotAcknowledged() {
+        nackedPublishedMessages.increment();
+    }
+
+    @Override
+    protected void markPublishedMessageNotRouted() {
+        unroutedPublishedMessages.increment();
+    }
+
     public AtomicLong getConnections() {
         return connections;
     }
@@ -144,6 +168,18 @@ public Counter getFailedToPublishMessages() {
         return failedToPublishMessages;
     }
 
+    public Counter getAckedPublishedMessages() {
+        return ackedPublishedMessages;
+    }
+
+    public Counter getNackedPublishedMessages() {
+        return nackedPublishedMessages;
+    }
+
+    public Counter getUnroutedPublishedMessages() {
+        return unroutedPublishedMessages;
+    }
+
     public Counter getConsumedMessages() {
         return consumedMessages;
     }
@@ -198,6 +234,24 @@ Object create(MeterRegistry registry, String prefix, Iterable<Tag> tags) {
             Object create(MeterRegistry registry, String prefix, Iterable<Tag> tags) {
                 return registry.counter(prefix + ".failed_to_publish", tags);
             }
+        },
+        ACKED_PUBLISHED_MESSAGES {
+            @Override
+            Object create(MeterRegistry registry, String prefix, Iterable<Tag> tags) {
+                return registry.counter(prefix + ".acknowledged_published", tags);
+            }
+        },
+        NACKED_PUBLISHED_MESSAGES {
+            @Override
+            Object create(MeterRegistry registry, String prefix, Iterable<Tag> tags) {
+                return registry.counter(prefix + ".not_acknowledged_published", tags);
+            }
+        },
+        UNROUTED_PUBLISHED_MESSAGES {
+            @Override
+            Object create(MeterRegistry registry, String prefix, Iterable<Tag> tags) {
+                return registry.counter(prefix + ".unrouted_published", tags);
+            }
         };
 
         /**

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

@@ -42,6 +42,9 @@ public class StandardMetricsCollector extends AbstractMetricsCollector {
     private final Meter acknowledgedMessages;
     private final Meter rejectedMessages;
     private final Meter failedToPublishMessages;
+    private final Meter publishAcknowledgedMessages;
+    private final Meter publishNacknowledgedMessages;
+    private final Meter publishUnroutedMessages;
 
 
     public StandardMetricsCollector(MetricRegistry registry, String metricsPrefix) {
@@ -50,6 +53,9 @@ public StandardMetricsCollector(MetricRegistry registry, String metricsPrefix) {
         this.channels = registry.counter(metricsPrefix+".channels");
         this.publishedMessages = registry.meter(metricsPrefix+".published");
         this.failedToPublishMessages = registry.meter(metricsPrefix+".failed_to_publish");
+        this.publishAcknowledgedMessages = registry.meter(metricsPrefix+".publish_ack");
+        this.publishNacknowledgedMessages = registry.meter(metricsPrefix+".publish_nack");
+        this.publishUnroutedMessages = registry.meter(metricsPrefix+".publish_unrouted");
         this.consumedMessages = registry.meter(metricsPrefix+".consumed");
         this.acknowledgedMessages = registry.meter(metricsPrefix+".acknowledged");
         this.rejectedMessages = registry.meter(metricsPrefix+".rejected");
@@ -108,8 +114,21 @@ protected void markRejectedMessage() {
         rejectedMessages.mark();
     }
 
+    @Override
+    protected void markMessagePublishAcknowledged() {
+        acknowledgedMessages.mark();
+    }
+
+    @Override
+    protected void markMessagePublishNotAcknowledged() {
+        publishNacknowledgedMessages.mark();
+    }
+
+    @Override
+    protected void markPublishedMessageNotRouted() {
+        publishUnroutedMessages.mark();
+    }
 
-    
     public MetricRegistry getMetricRegistry() {
         return registry;
     }
@@ -141,4 +160,17 @@ public Meter getRejectedMessages() {
     public Meter getFailedToPublishMessages() {
         return failedToPublishMessages;
     }
+
+    public Meter getPublishAcknowledgedMessages() {
+        return publishAcknowledgedMessages;
+    }
+
+    public Meter getPublishNotAcknowledgedMessages() {
+        return publishNacknowledgedMessages;
+    }
+
+    public Meter getPublishUnroutedMessages() {
+        return publishUnroutedMessages;
+    }
+
 }