Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add failure counter to AbstractMessageSource

Browse Source 
* Clear timers in AMH.destroy()
This commit is contained in:
Gary Russell
2019-09-20 10:53:18 -04:00
committed by Artem Bilan
parent aa29bb15e3
commit 58c0b9885b
3 changed files with 49 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractMessageSource.java
View File

@@ -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<T> extends AbstractExpressionEvaluat
private final ManagementOverrides managementOverrides = new ManagementOverrides();
private final Set<MeterFacade> meters = ConcurrentHashMap.newKeySet();
private Map<String, Expression> headerExpressions;
private String beanName;
@@ -158,7 +163,15 @@ public abstract class AbstractMessageSource<T> extends AbstractExpressionEvaluat
@Override
public final Message<T> 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<T> 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<String, Object> evaluateHeaders() {
return CollectionUtils.isEmpty(this.headerExpressions)
@@ -234,9 +253,8 @@ public abstract class AbstractMessageSource<T> extends AbstractExpressionEvaluat
@Override
public void destroy() {
if (this.receiveCounter != null) {
this.receiveCounter.remove();
}
this.meters.forEach(MeterFacade::remove);
this.meters.clear();
}
}

1
spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java
View File

@@ -347,6 +347,7 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport
@Override
public void destroy() {
this.timers.forEach(MeterFacade::remove);
this.timers.clear();
}
}

24
spring-integration-core/src/test/java/org/springframework/integration/support/management/graph/IntegrationGraphServerTests.java
View File

@@ -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<String, Object> gateway1 = (Map) jsonArray.get(0);
Map<String, Object> gateway1 = (Map<String, Object>) jsonArray.get(0);
Map<String, Object> properties = (Map<String, Object>) 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<String> 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;
}