diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java index 4f9e880067..0d8a6ab5ef 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java @@ -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(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java index 8ba458d68b..b2022b552f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java @@ -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; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java index 5ecb01511a..e8bf89025e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java @@ -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) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/graph/IntegrationGraphServer.java b/spring-integration-core/src/main/java/org/springframework/integration/graph/IntegrationGraphServer.java index a95324a875..5086fe63f7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/graph/IntegrationGraphServer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/graph/IntegrationGraphServer.java @@ -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 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 micrometerEnhancer; + + NodeFactory(Function 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 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, diff --git a/spring-integration-core/src/main/java/org/springframework/integration/graph/IntegrationNode.java b/spring-integration-core/src/main/java/org/springframework/integration/graph/IntegrationNode.java index 3b74bde5a8..8f6b8cf203 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/graph/IntegrationNode.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/graph/IntegrationNode.java @@ -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 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; + } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/graph/MicrometerNodeEnhancer.java b/spring-integration-core/src/main/java/org/springframework/integration/graph/MicrometerNodeEnhancer.java index 727ee7939c..81bb37b149 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/graph/MicrometerNodeEnhancer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/graph/MicrometerNodeEnhancer.java @@ -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 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 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 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 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)); + } + } 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 7de6914c67..65649a2d4e 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 @@ -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(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerSupport.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerSupport.java index 28b3d9f573..93edaa495c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerSupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerSupport.java @@ -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; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/IntegrationManagement.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/IntegrationManagement.java index f177e97a05..673c99d105 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/IntegrationManagement.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/IntegrationManagement.java @@ -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 diff --git a/spring-integration-core/src/test/java/org/springframework/integration/graph/IntegrationGraphServerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/graph/IntegrationGraphServerTests.java index d24e683283..cab590adb6 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/graph/IntegrationGraphServerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/graph/IntegrationGraphServerTests.java @@ -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> nodes = (List>) 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.get("nodes"); assertThat(nodes).isNotNull(); - assertThat(nodes.size()).isEqualTo(34); + assertThat(nodes.size()).isEqualTo(35); links = (List>) 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 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()); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/graph/integration-graph-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/graph/integration-graph-context.xml index 1bfdd703ad..07c14bc518 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/graph/integration-graph-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/graph/integration-graph-context.xml @@ -15,7 +15,7 @@ - +