Merge pull request #360 from slayful/rabbitmq-java-354-ack-nack-unrouted

acogoluegnes · web-flow · commit b248a7a27992 · 2018-06-04T17:13:40.000+02:00
#354 | Metrics for ack, nack & unrouted publishes
diff --git a/src/main/java/com/rabbitmq/client/MetricsCollector.java b/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, long deliveryTag, boolean multiple);
+
+    void basicPublishNack(Channel channel, long deliveryTag, boolean multiple);
+
+    void basicPublishUnrouted(Channel channel);
+
     void consumedMessage(Channel channel, long deliveryTag, boolean autoAck);
 
     void consumedMessage(Channel channel, long deliveryTag, String consumerTag);
diff --git a/src/main/java/com/rabbitmq/client/NoOpMetricsCollector.java b/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, long deliveryTag, boolean multiple) {
+
+    }
+
+    @Override
+    public void basicPublishNack(Channel channel, long deliveryTag, boolean multiple) {
+
+    }
+
+    @Override
+    public void basicPublishUnrouted(Channel channel) {
+
+    }
+
     @Override
     public void consumedMessage(Channel channel, long deliveryTag, boolean autoAck) {
 
diff --git a/src/main/java/com/rabbitmq/client/impl/AbstractMetricsCollector.java b/src/main/java/com/rabbitmq/client/impl/AbstractMetricsCollector.java
@@ -111,6 +111,41 @@ public void basicPublishFailure(Channel channel, Throwable cause) {
         }
     }
 
+    @Override
+    public void basicPublishAck(Channel channel, long deliveryTag, boolean multiple) {
+        if (multiple) {
+            // this is a naive approach, as it does not handle multiple nacks
+            return;
+        }
+        try {
+            markMessagePublishAcknowledged();
+        } catch(Exception e) {
+            LOGGER.info("Error while computing metrics in basicPublishAck: " + e.getMessage());
+        }
+    }
+
+    @Override
+    public void basicPublishNack(Channel channel, long deliveryTag, boolean multiple) {
+        if (multiple) {
+            // this is a naive approach, as it does not handle multiple nacks
+            return;
+        }
+        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 +395,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();
 }
diff --git a/src/main/java/com/rabbitmq/client/impl/ChannelN.java b/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());
         }
     }
 
diff --git a/src/main/java/com/rabbitmq/client/impl/MicrometerMetricsCollector.java b/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);
+            }
         };
 
         /**
diff --git a/src/main/java/com/rabbitmq/client/impl/StandardMetricsCollector.java b/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() {
+        publishAcknowledgedMessages.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;
+    }
+
 }
diff --git a/src/test/java/com/rabbitmq/client/test/MetricsCollectorTest.java b/src/test/java/com/rabbitmq/client/test/MetricsCollectorTest.java
@@ -149,6 +149,62 @@ public void basicGetAndAck() {
         assertThat(publishedMessages(metrics), is(2L));
     }
 
+    @Test public void publishingAcknowledgements() {
+        long anyDeliveryTag = 123L;
+        AbstractMetricsCollector metrics = factory.create();
+        Channel channel = mock(Channel.class);
+        // begins with no messages acknowledged
+        assertThat(publishAck(metrics), is(0L));
+        // first acknowledgement gets tracked
+        metrics.basicPublishAck(channel, anyDeliveryTag, false);
+        assertThat(publishAck(metrics), is(1L));
+        // second acknowledgement gets tracked
+        metrics.basicPublishAck(channel, anyDeliveryTag, false);
+        assertThat(publishAck(metrics), is(2L));
+        // multiple deliveries aren't tracked
+        metrics.basicPublishAck(channel, anyDeliveryTag, true);
+        assertThat(publishAck(metrics), is(2L));
+        // cleaning stale state doesn't affect the metric
+        metrics.cleanStaleState();
+        assertThat(publishAck(metrics), is(2L));
+    }
+
+    @Test public void publishingNotAcknowledgements() {
+        long anyDeliveryTag = 123L;
+        AbstractMetricsCollector metrics = factory.create();
+        Channel channel = mock(Channel.class);
+        // begins with no messages not-acknowledged
+        assertThat(publishNack(metrics), is(0L));
+        // first not-acknowledgement gets tracked
+        metrics.basicPublishNack(channel, anyDeliveryTag, false);
+        assertThat(publishNack(metrics), is(1L));
+        // second not-acknowledgement gets tracked
+        metrics.basicPublishNack(channel, anyDeliveryTag, false);
+        assertThat(publishNack(metrics), is(2L));
+        // multiple deliveries aren't tracked
+        metrics.basicPublishNack(channel, anyDeliveryTag, true);
+        assertThat(publishNack(metrics), is(2L));
+        // cleaning stale state doesn't affect the metric
+        metrics.cleanStaleState();
+        assertThat(publishNack(metrics), is(2L));
+    }
+
+    @Test public void publishingUnrouted() {
+        AbstractMetricsCollector metrics = factory.create();
+        Channel channel = mock(Channel.class);
+        // begins with no messages not-acknowledged
+        assertThat(publishUnrouted(metrics), is(0L));
+        // first unrouted gets tracked
+        metrics.basicPublishUnrouted(channel);
+        assertThat(publishUnrouted(metrics), is(1L));
+        // second unrouted gets tracked
+        metrics.basicPublishUnrouted(channel);
+        assertThat(publishUnrouted(metrics), is(2L));
+        // cleaning stale state doesn't affect the metric
+        metrics.cleanStaleState();
+        assertThat(publishUnrouted(metrics), is(2L));
+    }
+
     @Test public void cleanStaleState() {
         AbstractMetricsCollector metrics = factory.create();
         Connection openConnection = mock(Connection.class);
@@ -189,6 +245,31 @@ public void basicGetAndAck() {
         assertThat(channels(metrics), is(1L));
     }
 
+
+    long publishAck(MetricsCollector metrics) {
+        if (metrics instanceof StandardMetricsCollector) {
+            return ((StandardMetricsCollector) metrics).getPublishAcknowledgedMessages().getCount();
+        } else {
+            return (long) ((MicrometerMetricsCollector) metrics).getAckedPublishedMessages().count();
+        }
+    }
+
+    long publishNack(MetricsCollector metrics) {
+        if (metrics instanceof StandardMetricsCollector) {
+            return ((StandardMetricsCollector) metrics).getPublishNotAcknowledgedMessages().getCount();
+        } else {
+            return (long) ((MicrometerMetricsCollector) metrics).getNackedPublishedMessages().count();
+        }
+    }
+
+    long publishUnrouted(MetricsCollector metrics) {
+        if (metrics instanceof StandardMetricsCollector) {
+            return ((StandardMetricsCollector) metrics).getPublishUnroutedMessages().getCount();
+        } else {
+            return (long) ((MicrometerMetricsCollector) metrics).getUnroutedPublishedMessages().count();
+        }
+    }
+
     long publishedMessages(MetricsCollector metrics) {
         if (metrics instanceof StandardMetricsCollector) {
             return ((StandardMetricsCollector) metrics).getPublishedMessages().getCount();
diff --git a/src/test/java/com/rabbitmq/client/test/functional/Metrics.java b/src/test/java/com/rabbitmq/client/test/functional/Metrics.java

Original file line number	Diff line number	Diff line change
`@@ -481,6 +481,8 @@ private void callReturnListeners(Command command, Basic.Return basicReturn) {`
`481`	`481`	`}`
`482`	`482`	`} catch (Throwable ex) {`
`483`	`483`	`getConnection().getExceptionHandler().handleReturnListenerException(this, ex);`
	`484`	`+ } finally {`
	`485`	`+ metricsCollector.basicPublishUnrouted(this);`
`484`	`486`	`}`
`485`	`487`	`}`
`486`	`488`
`@@ -491,6 +493,8 @@ private void callConfirmListeners(@SuppressWarnings("unused") Command command, B`
`491`	`493`	`}`
`492`	`494`	`} catch (Throwable ex) {`
`493`	`495`	`getConnection().getExceptionHandler().handleConfirmListenerException(this, ex);`
	`496`	`+ } finally {`
	`497`	`+ metricsCollector.basicPublishAck(this, ack.getDeliveryTag(), ack.getMultiple());`
`494`	`498`	`}`
`495`	`499`	`}`
`496`	`500`
`@@ -501,6 +505,8 @@ private void callConfirmListeners(@SuppressWarnings("unused") Command command, B`
`501`	`505`	`}`
`502`	`506`	`} catch (Throwable ex) {`
`503`	`507`	`getConnection().getExceptionHandler().handleConfirmListenerException(this, ex);`
	`508`	`+ } finally {`
	`509`	`+ metricsCollector.basicPublishNack(this, nack.getDeliveryTag(), nack.getMultiple());`
`504`	`510`	`}`
`505`	`511`	`}`
`506`	`512`