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

Obtain observed timers in the Integration Graph (#3967)

Browse Source 
* Obtain observed timers in the Integration Graph

The `Observation` populates slightly different Micrometer timer names and their tags

* Fix `MicrometerNodeEnhancer` to search timers in the registry according to the `Observation`
convention if the provided component is observed
* For that purpose expose an `IntegrationManagement.isObserved()` option to check
of component is instrumented with an `Observation`
* Improve the logic in the `IntegrationGraphServer` to delegate into a `Function`
to call `micrometerEnhancer.enhance()` instead of `static` property which might not
be OK in the environment where several integration applications are ran in the same JVM

* * Collect times into graph for inbound gateways & producers

* Fix language in javadoc of new `IntegrationManagement.isObserved()`

* Add `@Nullable` to `MicrometerNodeEnhancer.buildTimerStats()` param to satisfy this method contract

Co-authored-by: Gary Russell <grussell@vmware.com>

Co-authored-by: Gary Russell <grussell@vmware.com>
This commit is contained in:
Artem Bilan
2022-12-20 11:46:48 -05:00
committed by GitHub
parent d38e0d8720
commit 49a9b7e314
11 changed files with 231 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java
View File

@@ -244,6 +244,11 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
this.observationRegistry = observationRegistry;
}
@Override
public boolean isObserved() {
return !ObservationRegistry.NOOP.equals(this.observationRegistry);
}
@Override
protected void onInit() {
super.onInit();

5
spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java
View File

@@ -199,6 +199,11 @@ public abstract class MessageProducerSupport extends AbstractEndpoint
this.observationRegistry = observationRegistry;
}
@Override
public boolean isObserved() {
return !ObservationRegistry.NOOP.equals(this.observationRegistry);
}
@Override
public IntegrationPatternType getIntegrationPatternType() {
return IntegrationPatternType.inbound_channel_adapter;

5
spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java
View File

@@ -370,6 +370,11 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
this.observationRegistry = observationRegistry;
}
@Override
public boolean isObserved() {
return !ObservationRegistry.NOOP.equals(this.observationRegistry);
}
public void setObservationConvention(
@Nullable MessageRequestReplyReceiverObservationConvention observationConvention) {

117
spring-integration-core/src/main/java/org/springframework/integration/graph/IntegrationGraphServer.java
View File

@@ -68,13 +68,13 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat
private static final float GRAPH_VERSION = 1.2f;
private static MicrometerNodeEnhancer micrometerEnhancer;
private final NodeFactory nodeFactory = new NodeFactory(this::enhance);
private final NodeFactory nodeFactory = new NodeFactory();
private MicrometerNodeEnhancer micrometerEnhancer;
private ApplicationContext applicationContext;
private Graph graph;
private volatile Graph graph;
private String applicationName;
@@ -158,9 +158,18 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat
return this.applicationContext.getBeansOfType(type, true, false);
}
private <T extends IntegrationNode> T enhance(T node) {
if (this.micrometerEnhancer != null) {
return this.micrometerEnhancer.enhance(node);
}
else {
return node;
}
}
private synchronized Graph buildGraph() {
if (micrometerEnhancer == null && MicrometerMetricsCaptorConfiguration.METER_REGISTRY_PRESENT) {
micrometerEnhancer = new MicrometerNodeEnhancer(this.applicationContext);
if (this.micrometerEnhancer == null && MicrometerMetricsCaptorConfiguration.METER_REGISTRY_PRESENT) {
this.micrometerEnhancer = new MicrometerNodeEnhancer(this.applicationContext);
}
String implementationVersion = IntegrationGraphServer.class.getPackage().getImplementationVersion();
if (implementationVersion == null) {
@@ -354,7 +363,10 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat
private final AtomicInteger nodeId = new AtomicInteger();
NodeFactory() {
private final Function<? super IntegrationNode, ? extends IntegrationNode> micrometerEnhancer;
NodeFactory(Function<IntegrationNode, ? extends IntegrationNode> micrometerEnhancer) {
this.micrometerEnhancer = micrometerEnhancer;
}
MessageChannelNode channelNode(String name, MessageChannel channel) {
@@ -365,10 +377,12 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat
else {
node = new MessageChannelNode(this.nodeId.incrementAndGet(), name, channel);
}
if (IntegrationGraphServer.micrometerEnhancer != null) {
node = IntegrationGraphServer.micrometerEnhancer.enhance(node);
}
return node;
return enhance(node);
}
@SuppressWarnings("unchecked")
private <T extends IntegrationNode> T enhance(T node) {
return (T) this.micrometerEnhancer.apply(node);
}
MessageGatewayNode gatewayNode(String name, MessagingGatewaySupport gateway) {
@@ -379,16 +393,15 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat
@Nullable
private String channelToBeanName(MessageChannel messageChannel) {
return messageChannel instanceof NamedComponent
? ((NamedComponent) messageChannel).getBeanName()
return messageChannel instanceof NamedComponent namedComponent
? namedComponent.getBeanName()
: Objects.toString(messageChannel, null);
}
MessageProducerNode producerNode(String name, MessageProducerSupport producer) {
String errorChannel = channelToBeanName(producer.getErrorChannel());
String outputChannel = channelToBeanName(producer.getOutputChannel());
return new MessageProducerNode(this.nodeId.incrementAndGet(), name, producer,
outputChannel, errorChannel);
return new MessageProducerNode(this.nodeId.incrementAndGet(), name, producer, outputChannel, errorChannel);
}
MessageSourceNode sourceNode(String name, SourcePollingChannelAdapter adapter) {
@@ -401,10 +414,8 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat
}
MessageSourceNode node = new MessageSourceNode(this.nodeId.incrementAndGet(), nameToUse, source,
outputChannel, errorChannel);
if (IntegrationGraphServer.micrometerEnhancer != null) {
node = IntegrationGraphServer.micrometerEnhancer.enhance(node);
}
return node;
return enhance(node);
}
MessageHandlerNode handlerNode(String nameArg, IntegrationConsumer consumer) {
@@ -412,34 +423,30 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat
MessageHandler handler = consumer.getHandler();
MessageHandlerNode node;
String name = nameArg;
if (handler instanceof NamedComponent) {
name = IntegrationUtils.obtainComponentName((NamedComponent) handler);
if (handler instanceof NamedComponent namedComponent) {
name = IntegrationUtils.obtainComponentName(namedComponent);
}
if (handler instanceof CompositeMessageHandler) {
node = compositeHandler(name, consumer, (CompositeMessageHandler) handler, outputChannelName, null,
false);
if (handler instanceof CompositeMessageHandler compositeMessageHandler) {
node = compositeHandler(name, consumer, compositeMessageHandler, outputChannelName, null, false);
}
else if (handler instanceof DiscardingMessageHandler) {
node = discardingHandler(name, consumer, (DiscardingMessageHandler) handler, outputChannelName, null,
false);
else if (handler instanceof DiscardingMessageHandler discardingMessageHandler) {
node = discardingHandler(name, consumer, discardingMessageHandler, outputChannelName, null, false);
}
else if (handler instanceof MappingMessageRouterManagement) {
node = routingHandler(name, consumer, handler, (MappingMessageRouterManagement) handler,
outputChannelName, null, false);
else if (handler instanceof MappingMessageRouterManagement mappingMessageRouterManagement) {
node = routingHandler(name, consumer, handler, mappingMessageRouterManagement, outputChannelName,
null, false);
}
else if (handler instanceof RecipientListRouterManagement) {
node = recipientListRoutingHandler(name, consumer, handler, (RecipientListRouterManagement) handler,
else if (handler instanceof RecipientListRouterManagement recipientListRouterManagement) {
node = recipientListRoutingHandler(name, consumer, handler, recipientListRouterManagement,
outputChannelName, null, false);
}
else {
String inputChannel = channelToBeanName(consumer.getInputChannel());
node = new MessageHandlerNode(this.nodeId.incrementAndGet(), name, handler,
inputChannel, outputChannelName);
node = new MessageHandlerNode(this.nodeId.incrementAndGet(), name, handler, inputChannel,
outputChannelName);
}
if (IntegrationGraphServer.micrometerEnhancer != null) {
node = IntegrationGraphServer.micrometerEnhancer.enhance(node);
}
return node;
return enhance(node);
}
MessageHandlerNode polledHandlerNode(String nameArg, PollingConsumer consumer) {
@@ -448,34 +455,30 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat
MessageHandler handler = consumer.getHandler();
MessageHandlerNode node;
String name = nameArg;
if (handler instanceof NamedComponent) {
name = IntegrationUtils.obtainComponentName((NamedComponent) handler);
if (handler instanceof NamedComponent namedComponent) {
name = IntegrationUtils.obtainComponentName(namedComponent);
}
if (handler instanceof CompositeMessageHandler) {
node = compositeHandler(name, consumer, (CompositeMessageHandler) handler, outputChannelName,
if (handler instanceof CompositeMessageHandler compositeMessageHandler) {
node = compositeHandler(name, consumer, compositeMessageHandler, outputChannelName, errorChannel, true);
}
else if (handler instanceof DiscardingMessageHandler discardMessageHandler) {
node = discardingHandler(name, consumer, discardMessageHandler, outputChannelName, errorChannel, true);
}
else if (handler instanceof MappingMessageRouterManagement mappingMessageRouterManagement) {
node = routingHandler(name, consumer, handler, mappingMessageRouterManagement, outputChannelName,
errorChannel, true);
}
else if (handler instanceof DiscardingMessageHandler) {
node = discardingHandler(name, consumer, (DiscardingMessageHandler) handler, outputChannelName,
errorChannel, true);
}
else if (handler instanceof MappingMessageRouterManagement) {
node = routingHandler(name, consumer, handler, (MappingMessageRouterManagement) handler,
outputChannelName, errorChannel, true);
}
else if (handler instanceof RecipientListRouterManagement) {
node = recipientListRoutingHandler(name, consumer, handler, (RecipientListRouterManagement) handler,
else if (handler instanceof RecipientListRouterManagement recipientListRouterManagement) {
node = recipientListRoutingHandler(name, consumer, handler, recipientListRouterManagement,
outputChannelName, errorChannel, true);
}
else {
String inputChannel = channelToBeanName(consumer.getInputChannel());
node = new ErrorCapableMessageHandlerNode(this.nodeId.incrementAndGet(), name, handler,
inputChannel, outputChannelName, errorChannel);
node = new ErrorCapableMessageHandlerNode(this.nodeId.incrementAndGet(), name, handler, inputChannel,
outputChannelName, errorChannel);
}
if (IntegrationGraphServer.micrometerEnhancer != null) {
node = IntegrationGraphServer.micrometerEnhancer.enhance(node);
}
return node;
return enhance(node);
}
MessageHandlerNode compositeHandler(String name, IntegrationConsumer consumer,

31
spring-integration-core/src/main/java/org/springframework/integration/graph/IntegrationNode.java
View File

@@ -25,6 +25,7 @@ import org.springframework.integration.IntegrationPattern;
import org.springframework.integration.IntegrationPatternType;
import org.springframework.integration.context.ExpressionCapable;
import org.springframework.integration.support.context.NamedComponent;
import org.springframework.integration.support.management.IntegrationManagement;
import org.springframework.lang.Nullable;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
@@ -57,24 +58,34 @@ public abstract class IntegrationNode {
private final Map<String, Object> unmodifiableProperties = Collections.unmodifiableMap(this.properties);
private final transient boolean observed;
protected IntegrationNode(int nodeId, String name, Object nodeObject) {
this.nodeId = nodeId;
this.nodeName = name;
this.componentType =
nodeObject instanceof NamedComponent
? ((NamedComponent) nodeObject).getComponentType()
nodeObject instanceof NamedComponent namedComponent
? namedComponent.getComponentType()
: nodeObject.getClass().getSimpleName();
if (nodeObject instanceof ExpressionCapable) {
Expression expression = ((ExpressionCapable) nodeObject).getExpression();
if (nodeObject instanceof ExpressionCapable expressionCapable) {
Expression expression = expressionCapable.getExpression();
if (expression != null) {
this.properties.put("expression", expression.getExpressionString());
}
}
if (nodeObject instanceof IntegrationManagement integrationManagement) {
this.observed = integrationManagement.isObserved();
}
else {
this.observed = false;
}
IntegrationPatternType patternType = null;
if (nodeObject instanceof IntegrationPattern) {
patternType = ((IntegrationPattern) nodeObject).getIntegrationPatternType(); // NOSONAR
if (nodeObject instanceof IntegrationPattern integrationPattern) {
patternType = integrationPattern.getIntegrationPatternType();
}
else if (nodeObject instanceof MessageHandler) {
patternType = IntegrationPatternType.service_activator;
@@ -140,4 +151,12 @@ public abstract class IntegrationNode {
}
}
/**
* Return true if this component is instrumented with an observation.
* @return true if this component is instrumented with an observation.
* @since 6.0.1
*/
public boolean isObserved() {
return this.observed;
}
}

89
spring-integration-core/src/main/java/org/springframework/integration/graph/MicrometerNodeEnhancer.java
View File

@@ -22,14 +22,22 @@ import java.util.concurrent.TimeUnit;
import io.micrometer.core.instrument.Counter;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Timer;
import io.micrometer.core.instrument.search.Search;
import org.springframework.context.ApplicationContext;
import org.springframework.integration.support.management.IntegrationManagement;
import org.springframework.integration.support.management.observation.DefaultMessageReceiverObservationConvention;
import org.springframework.integration.support.management.observation.DefaultMessageRequestReplyReceiverObservationConvention;
import org.springframework.integration.support.management.observation.DefaultMessageSenderObservationConvention;
import org.springframework.integration.support.management.observation.IntegrationObservation;
import org.springframework.lang.Nullable;
/**
* Add micrometer metrics to the node.
*
* @author Gary Russell
* @author Artem Bilan
*
* @since 5.2
*
*/
@@ -68,7 +76,7 @@ public class MicrometerNodeEnhancer {
if (node instanceof MessageChannelNode) {
enhanceWithTimers(node, "channel");
}
else if (node instanceof MessageHandlerNode) {
else if (node instanceof MessageHandlerNode || node instanceof MessageProducerNode) {
enhanceWithTimers(node, "handler");
}
if (node instanceof PollableChannelNode) {
@@ -77,6 +85,9 @@ public class MicrometerNodeEnhancer {
else if (node instanceof MessageSourceNode) {
enhanceWithCounts(node, "source");
}
else if (node instanceof MessageGatewayNode) {
enhanceWithTimers(node, "gateway");
}
}
return node;
}
@@ -86,35 +97,53 @@ public class MicrometerNodeEnhancer {
}
private <T extends IntegrationNode> SendTimers retrieveTimers(T node, String type) {
Timer successTimer = null;
Timer successTimer = obtainTimer(node, type, true);
Timer failureTimer = obtainTimer(node, type, false);
return new SendTimers(buildTimerStats(successTimer), buildTimerStats(failureTimer));
}
@Nullable
private <T extends IntegrationNode> Timer obtainTimer(T node, String type, boolean success) {
try {
successTimer = this.registry.get(IntegrationManagement.SEND_TIMER_NAME)
.tag(TAG_TYPE, type)
.tag(TAG_NAME, node.getName())
.tag(TAG_RESULT, "success")
if (node.isObserved()) {
return observationTimer(node, type, success);
}
else {
return this.registry.get(IntegrationManagement.SEND_TIMER_NAME)
.tag(TAG_TYPE, type)
.tag(TAG_NAME, node.getName())
.tag(TAG_RESULT, success ? "success" : "failure")
.timer();
}
}
catch (Exception ex) {
return null;
}
}
@Nullable
private <T extends IntegrationNode> Timer observationTimer(T node, String type, boolean success) {
Search timerSearch =
switch (type) {
case "channel" -> this.registry.find(DefaultMessageSenderObservationConvention.INSTANCE.getName())
.tag(IntegrationObservation.ProducerTags.COMPONENT_TYPE.asString(), "producer");
case "handler" -> this.registry.find(DefaultMessageReceiverObservationConvention.INSTANCE.getName())
.tag(IntegrationObservation.HandlerTags.COMPONENT_TYPE.asString(), "handler");
case "gateway" ->
this.registry.find(DefaultMessageRequestReplyReceiverObservationConvention.INSTANCE.getName())
.tag(IntegrationObservation.GatewayTags.COMPONENT_TYPE.asString(), "gateway");
default -> null;
};
if (timerSearch != null) {
return timerSearch
.tag(IntegrationObservation.HandlerTags.COMPONENT_NAME.asString(), node.getName())
.tag("error", value -> success == "none".equals(value))
.timer();
}
catch (@SuppressWarnings(UNUSED) Exception e) { // NOSONAR ignored
// NOSONAR empty
}
Timer failureTimer = null;
try {
failureTimer = this.registry.get(IntegrationManagement.SEND_TIMER_NAME)
.tag(TAG_TYPE, type)
.tag(TAG_NAME, node.getName())
.tag(TAG_RESULT, "failure")
.timer();
}
catch (@SuppressWarnings(UNUSED) Exception e) { // NOSONAR ignored
// NOSONAR empty;
}
TimerStats successes = successTimer == null ? ZERO_TIMER_STATS
: new TimerStats(successTimer.count(), successTimer.mean(TimeUnit.MILLISECONDS),
successTimer.max(TimeUnit.MILLISECONDS));
TimerStats failures = failureTimer == null ? ZERO_TIMER_STATS
: new TimerStats(failureTimer.count(), failureTimer.mean(TimeUnit.MILLISECONDS),
failureTimer.max(TimeUnit.MILLISECONDS));
return new SendTimers(successes, failures);
return null;
}
private <T extends IntegrationNode> void enhanceWithCounts(T node, String type) {
@@ -150,5 +179,11 @@ public class MicrometerNodeEnhancer {
(long) (failures == null ? 0 : failures.count()));
}
private static TimerStats buildTimerStats(@Nullable Timer timer) {
return timer == null
? ZERO_TIMER_STATS
: new TimerStats(timer.count(), timer.mean(TimeUnit.MILLISECONDS), timer.max(TimeUnit.MILLISECONDS));
}
}

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

@@ -61,9 +61,8 @@ public abstract class AbstractMessageHandler extends MessageHandlerSupport
if (isLoggingEnabled()) {
this.logger.debug(() -> this + " received message: " + message);
}
ObservationRegistry observationRegistry = getObservationRegistry();
if (observationRegistry != null) {
handleWithObservation(message, observationRegistry);
if (isObserved()) {
handleWithObservation(message, getObservationRegistry());
}
else {
MetricsCaptor metricsCaptor = getMetricsCaptor();

9
spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerSupport.java
View File

@@ -32,6 +32,7 @@ import org.springframework.integration.support.management.TrackableComponent;
import org.springframework.integration.support.management.metrics.MeterFacade;
import org.springframework.integration.support.management.metrics.MetricsCaptor;
import org.springframework.integration.support.management.metrics.TimerFacade;
import org.springframework.util.Assert;
/**
* Base class for Message handling components that provides basic validation and error
@@ -63,7 +64,7 @@ public abstract class MessageHandlerSupport extends IntegrationObjectSupport
private MetricsCaptor metricsCaptor;
private ObservationRegistry observationRegistry;
private ObservationRegistry observationRegistry = ObservationRegistry.NOOP;
private int order = Ordered.LOWEST_PRECEDENCE;
@@ -95,9 +96,15 @@ public abstract class MessageHandlerSupport extends IntegrationObjectSupport
@Override
public void registerObservationRegistry(ObservationRegistry observationRegistry) {
Assert.notNull(observationRegistry, "'observationRegistry' must not be null");
this.observationRegistry = observationRegistry;
}
@Override
public boolean isObserved() {
return !ObservationRegistry.NOOP.equals(this.observationRegistry);
}
protected ObservationRegistry getObservationRegistry() {
return this.observationRegistry;
}

9
spring-integration-core/src/main/java/org/springframework/integration/support/management/IntegrationManagement.java
View File

@@ -106,6 +106,15 @@ public interface IntegrationManagement extends NamedComponent, DisposableBean {
// no op
}
/**
* True if this implementation is going to deal with a registry other than the {@link ObservationRegistry#NOOP} instance.
* @return true if this implementation is going to deal with a registry other than the {@link ObservationRegistry#NOOP} instance.
* @since 6.0.1
*/
default boolean isObserved() {
return false;
}
@Override
default void destroy() {
// no op

57
spring-integration-core/src/test/java/org/springframework/integration/graph/IntegrationGraphServerTests.java
View File

@@ -26,8 +26,11 @@ import java.util.Map;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializationFeature;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.observation.DefaultMeterObservationHandler;
import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
import io.micrometer.observation.ObservationRegistry;
import net.minidev.json.JSONArray;
import org.assertj.core.api.InstanceOfAssertFactories;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
@@ -49,6 +52,7 @@ import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.config.EnableIntegrationManagement;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.context.IntegrationFlowContext;
import org.springframework.integration.dsl.context.IntegrationFlowContext.IntegrationFlowRegistration;
@@ -117,7 +121,7 @@ public class IntegrationGraphServerTests {
assertThat(map.size()).isEqualTo(3);
List<Map<?, ?>> nodes = (List<Map<?, ?>>) map.get("nodes");
assertThat(nodes).isNotNull();
assertThat(nodes.size()).isEqualTo(34);
assertThat(nodes.size()).isEqualTo(35);
JSONArray jsonArray =
JsonPathUtils.evaluate(baos.toByteArray(), "$..nodes[?(@.componentType == 'gateway')]");
@@ -166,7 +170,7 @@ public class IntegrationGraphServerTests {
assertThat(map.size()).isEqualTo(3);
nodes = (List<Map<?, ?>>) map.get("nodes");
assertThat(nodes).isNotNull();
assertThat(nodes.size()).isEqualTo(34);
assertThat(nodes.size()).isEqualTo(35);
links = (List<Map<?, ?>>) map.get("links");
assertThat(links).isNotNull();
assertThat(links.size()).isEqualTo(35);
@@ -239,20 +243,51 @@ public class IntegrationGraphServerTests {
@Test
void testIncludesDynamic() {
Graph graph = this.server.getGraph();
assertThat(graph.getNodes().size()).isEqualTo(34);
assertThat(graph.getNodes().size()).isEqualTo(35);
IntegrationFlow flow = f -> f.handle(m -> {
});
IntegrationFlowRegistration reg = this.flowContext.registration(flow).register();
graph = this.server.rebuild();
assertThat(graph.getNodes().size()).isEqualTo(36);
assertThat(graph.getNodes().size()).isEqualTo(37);
this.flowContext.remove(reg.getId());
graph = this.server.rebuild();
assertThat(graph.getNodes().size()).isEqualTo(34);
assertThat(graph.getNodes().size()).isEqualTo(35);
}
@Autowired
MessageChannel filterInputChannel;
@Test
void timersViaObservationArePopulated() {
MessagingTemplate messagingTemplate = new MessagingTemplate();
assertThat(messagingTemplate.convertSendAndReceive(this.filterInputChannel, "test", String.class))
.isEqualTo("test");
Graph graph = this.server.getGraph();
IntegrationNode myFilter =
graph.getNodes()
.stream()
.filter(node -> node.getName().equals("myFilter"))
.findFirst()
.get();
assertThat(myFilter)
.asInstanceOf(InstanceOfAssertFactories.type(MessageHandlerNode.class))
.extracting(MessageHandlerNode::getSendTimers)
.isNotNull()
.satisfies(sendTimers -> {
assertThat(sendTimers.getFailures().getCount()).isEqualTo(0);
assertThat(sendTimers.getFailures().getMax()).isEqualTo(0);
assertThat(sendTimers.getFailures().getMean()).isEqualTo(0);
assertThat(sendTimers.getSuccesses().getCount()).isEqualTo(1);
assertThat(sendTimers.getSuccesses().getMax()).isGreaterThan(0);
assertThat(sendTimers.getSuccesses().getMean()).isGreaterThan(0);
});
}
@Configuration
@EnableIntegration
@EnableIntegrationManagement
@EnableIntegrationManagement(observationPatterns = "myFilter")
@IntegrationComponentScan
@ImportResource("org/springframework/integration/graph/integration-graph-context.xml")
public static class Config {
@@ -262,14 +297,20 @@ public class IntegrationGraphServerTests {
return new SimpleMeterRegistry();
}
@Bean
public ObservationRegistry observationRegistry(MeterRegistry meterRegistry) {
ObservationRegistry registry = ObservationRegistry.create();
registry.observationConfig().observationHandler(new DefaultMeterObservationHandler(meterRegistry));
return registry;
}
@Bean
public IntegrationGraphServer server() {
IntegrationGraphServer server = new IntegrationGraphServer();
server.setApplicationName("myAppName:1.0");
server.setAdditionalPropertiesCallback(namedComponent -> {
Map<String, Object> properties = null;
if (namedComponent instanceof SmartLifecycle) {
SmartLifecycle smartLifecycle = (SmartLifecycle) namedComponent;
if (namedComponent instanceof SmartLifecycle smartLifecycle) {
properties = new HashMap<>();
properties.put("auto-startup", smartLifecycle.isAutoStartup());
properties.put("running", smartLifecycle.isRunning());

2
spring-integration-core/src/test/java/org/springframework/integration/graph/integration-graph-context.xml
View File

@@ -15,7 +15,7 @@
<int:service-activator expression="payload" />
</int:chain>
<int:filter id="myFilter" input-channel="filterChannel" discard-channel="three" expression="true" />
<int:filter id="myFilter" input-channel="filterInputChannel" discard-channel="three" expression="true" />
<int:aggregator id="myAggregator" input-channel="polledChannel" discard-channel="three" />