From 58c0b9885b9362dd98a218ba72edb0d829799c41 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 20 Sep 2019 10:53:18 -0400 Subject: [PATCH] Add failure counter to AbstractMessageSource * Clear timers in AMH.destroy() --- .../endpoint/AbstractMessageSource.java | 40 ++++++++++++++----- .../handler/AbstractMessageHandler.java | 1 + .../graph/IntegrationGraphServerTests.java | 24 ++++++++--- 3 files changed, 49 insertions(+), 16 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractMessageSource.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractMessageSource.java index 0ccb496fa2..81d99a6383 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractMessageSource.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractMessageSource.java @@ -18,6 +18,8 @@ package org.springframework.integration.endpoint; import java.util.HashMap; import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicLong; import org.springframework.beans.factory.BeanNameAware; @@ -28,6 +30,7 @@ import org.springframework.integration.support.AbstractIntegrationMessageBuilder import org.springframework.integration.support.context.NamedComponent; import org.springframework.integration.support.management.IntegrationManagedResource; import org.springframework.integration.support.management.metrics.CounterFacade; +import org.springframework.integration.support.management.metrics.MeterFacade; import org.springframework.integration.support.management.metrics.MetricsCaptor; import org.springframework.integration.util.AbstractExpressionEvaluator; import org.springframework.lang.Nullable; @@ -52,6 +55,8 @@ public abstract class AbstractMessageSource extends AbstractExpressionEvaluat private final ManagementOverrides managementOverrides = new ManagementOverrides(); + private final Set meters = ConcurrentHashMap.newKeySet(); + private Map headerExpressions; private String beanName; @@ -158,7 +163,15 @@ public abstract class AbstractMessageSource extends AbstractExpressionEvaluat @Override public final Message receive() { - return buildMessage(doReceive()); + try { + return buildMessage(doReceive()); + } + catch (RuntimeException ex) { + if (this.metricsCaptor != null) { + createCounter(false, ex.getClass().getSimpleName()).increment(); + } + throw ex; + } } @SuppressWarnings("unchecked") @@ -203,17 +216,23 @@ public abstract class AbstractMessageSource extends AbstractExpressionEvaluat private void incrementReceiveCounter() { if (this.receiveCounter == null) { - this.receiveCounter = this.metricsCaptor.counterBuilder(RECEIVE_COUNTER_NAME) - .tag("name", getComponentName() == null ? "unknown" : getComponentName()) - .tag("type", "source") - .tag("result", "success") - .tag("exception", "none") - .description("Messages received") - .build(); + this.receiveCounter = createCounter(true, "none"); } this.receiveCounter.increment(); } + private CounterFacade createCounter(boolean success, String exception) { + CounterFacade counter = this.metricsCaptor.counterBuilder(RECEIVE_COUNTER_NAME) + .tag("name", getComponentName() == null ? "unknown" : getComponentName()) + .tag("type", "source") + .tag("result", success ? "success" : "failure") + .tag("exception", exception) + .description("Messages received") + .build(); + this.meters.add(counter); + return counter; + } + @Nullable private Map evaluateHeaders() { return CollectionUtils.isEmpty(this.headerExpressions) @@ -234,9 +253,8 @@ public abstract class AbstractMessageSource extends AbstractExpressionEvaluat @Override public void destroy() { - if (this.receiveCounter != null) { - this.receiveCounter.remove(); - } + this.meters.forEach(MeterFacade::remove); + this.meters.clear(); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java index 16249ee13c..719f60e2e6 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java @@ -347,6 +347,7 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport @Override public void destroy() { this.timers.forEach(MeterFacade::remove); + this.timers.clear(); } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/support/management/graph/IntegrationGraphServerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/support/management/graph/IntegrationGraphServerTests.java index 892745d263..e35273fd7e 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/support/management/graph/IntegrationGraphServerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/support/management/graph/IntegrationGraphServerTests.java @@ -17,6 +17,7 @@ package org.springframework.integration.support.management.graph; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatIllegalStateException; import java.io.ByteArrayOutputStream; import java.util.Arrays; @@ -123,7 +124,7 @@ public class IntegrationGraphServerTests { assertThat(jsonArray.size()).isEqualTo(3); - Map gateway1 = (Map) jsonArray.get(0); + Map gateway1 = (Map) jsonArray.get(0); Map properties = (Map) gateway1.get("properties"); @@ -168,7 +169,7 @@ public class IntegrationGraphServerTests { assertThat(toRouterJson).contains("\"sendTimers\":{\"successes\":{\"count\":4"); jsonArray = JsonPathUtils.evaluate(baos.toByteArray(), "$..nodes[?(@.name == 'testSource')]"); String sourceJson = jsonArray.toJSONString(); - assertThat(sourceJson).contains("\"receiveCounters\":{\"successes\":1"); + assertThat(sourceJson).contains("\"receiveCounters\":{\"successes\":1,\"failures\":0"); // stats refresh without rebuild() this.testSource.receive(); @@ -177,7 +178,15 @@ public class IntegrationGraphServerTests { objectMapper.writeValue(baos, graph); jsonArray = JsonPathUtils.evaluate(baos.toByteArray(), "$..nodes[?(@.name == 'testSource')]"); sourceJson = jsonArray.toJSONString(); - assertThat(sourceJson).contains("\"receiveCounters\":{\"successes\":2"); + assertThat(sourceJson).contains("\"receiveCounters\":{\"successes\":2,\"failures\":0"); + + assertThatIllegalStateException().isThrownBy(() -> this.testSource.receive()); + baos = new ByteArrayOutputStream(); + objectMapper.enable(SerializationFeature.INDENT_OUTPUT); + objectMapper.writeValue(baos, graph); + jsonArray = JsonPathUtils.evaluate(baos.toByteArray(), "$..nodes[?(@.name == 'testSource')]"); + sourceJson = jsonArray.toJSONString(); + assertThat(sourceJson).contains("\"receiveCounters\":{\"successes\":2,\"failures\":1"); } @Test @@ -343,6 +352,8 @@ public class IntegrationGraphServerTests { return new QueueChannel(); } + int sourceCount; + @Bean @InboundChannelAdapter(channel = "fizChannel", autoStartup = "false") public MessageSource testSource() { @@ -355,6 +366,9 @@ public class IntegrationGraphServerTests { @Override protected Object doReceive() { + if (++sourceCount > 2) { + throw new IllegalStateException(); + } return new GenericMessage<>("foo"); } @@ -371,11 +385,11 @@ public class IntegrationGraphServerTests { } @ServiceActivator(inputChannel = "polledChannel") - public void bar(String foo) { + public void bar(@SuppressWarnings("unused") String foo) { } @Filter(inputChannel = "filterChannel") - public boolean filter(String payload) { + public boolean filter(@SuppressWarnings("unused") String payload) { return false; }