diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/GatewayProxySpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/GatewayProxySpec.java index 67b7d119e1..f9c82ca3b5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/GatewayProxySpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/GatewayProxySpec.java @@ -16,6 +16,7 @@ package org.springframework.integration.dsl; +import java.util.Collections; import java.util.HashMap; import java.util.Map; import java.util.concurrent.Executor; @@ -40,10 +41,10 @@ import org.springframework.messaging.MessageChannel; * declared. * * @author Artem Bilan - * + * @author Gary Russell * @since 5.2 */ -public class GatewayProxySpec { +public class GatewayProxySpec implements ComponentsRegistration { protected static final SpelExpressionParser PARSER = new SpelExpressionParser(); // NOSONAR - final @@ -287,4 +288,10 @@ public class GatewayProxySpec { return this.gatewayProxyFactoryBean; } + @Override + public Map getComponentsToRegister() { + return Collections.singletonMap(this.gatewayProxyFactoryBean, this.gatewayProxyFactoryBean.getBeanName() + + ".gateway"); + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java index 94e946d8dc..00f771f260 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java @@ -63,7 +63,9 @@ import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.expression.ExpressionUtils; import org.springframework.integration.expression.ValueExpression; import org.springframework.integration.support.channel.ChannelResolverUtils; +import org.springframework.integration.support.management.IntegrationManagement; import org.springframework.integration.support.management.TrackableComponent; +import org.springframework.integration.support.management.metrics.MetricsCaptor; import org.springframework.integration.util.JavaUtils; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; @@ -100,7 +102,8 @@ import reactor.core.publisher.Mono; * @author Artem Bilan */ public class GatewayProxyFactoryBean extends AbstractEndpoint - implements TrackableComponent, FactoryBean, MethodInterceptor, BeanClassLoaderAware { + implements TrackableComponent, FactoryBean, MethodInterceptor, BeanClassLoaderAware, + IntegrationManagement { private final Object initializationMonitor = new Object(); @@ -154,6 +157,8 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint private EvaluationContext evaluationContext = new StandardEvaluationContext(); + private MetricsCaptor metricsCaptor; + /** * Create a Factory whose service interface type can be configured by setter injection. * If none is set, it will fall back to the default service interface type, @@ -436,6 +441,12 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint return Collections.unmodifiableMap(this.gatewayMap); } + @Override + public void registerMetricsCaptor(MetricsCaptor metricsCaptorToRegister) { + this.metricsCaptor = metricsCaptorToRegister; + this.gatewayMap.values().forEach(gw -> gw.registerMetricsCaptor(metricsCaptorToRegister)); + } + @Override protected void onInit() { synchronized (this.initializationMonitor) { @@ -856,6 +867,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint gateway.setBeanName(getComponentName()); gateway.setBeanFactory(getBeanFactory()); gateway.setShouldTrack(this.shouldTrack); + gateway.registerMetricsCaptor(this.metricsCaptor); gateway.afterPropertiesSet(); return gateway; @@ -1020,6 +1032,11 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint } } + @Override + public void destroy() { + this.gatewayMap.values().forEach(MethodInvocationGateway::destroy); + } + private static final class MethodInvocationGateway extends MessagingGatewaySupport { private Expression receiveTimeoutExpression; 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 72d7d31ec9..1fff549c77 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 @@ -17,6 +17,8 @@ package org.springframework.integration.gateway; import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; import org.reactivestreams.Publisher; import org.reactivestreams.Subscriber; @@ -47,6 +49,10 @@ import org.springframework.integration.support.MutableMessageBuilder; import org.springframework.integration.support.converter.SimpleMessageConverter; import org.springframework.integration.support.management.IntegrationInboundManagement; import org.springframework.integration.support.management.IntegrationManagedResource; +import org.springframework.integration.support.management.metrics.MeterFacade; +import org.springframework.integration.support.management.metrics.MetricsCaptor; +import org.springframework.integration.support.management.metrics.SampleFacade; +import org.springframework.integration.support.management.metrics.TimerFacade; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; @@ -73,7 +79,6 @@ import reactor.core.publisher.MonoProcessor; * @author Gary Russell * @author Artem Bilan */ -@SuppressWarnings("deprecation") @IntegrationManagedResource public abstract class MessagingGatewaySupport extends AbstractEndpoint implements org.springframework.integration.support.management.TrackableComponent, @@ -90,10 +95,12 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint private final Object replyMessageCorrelatorMonitor = new Object(); - private boolean errorOnTimeout; - private final ManagementOverrides managementOverrides = new ManagementOverrides(); + private final Set timers = ConcurrentHashMap.newKeySet(); + + private boolean errorOnTimeout; + private ErrorMessageStrategy errorMessageStrategy = new DefaultErrorMessageStrategy(); private MessageChannel requestChannel; @@ -118,6 +125,10 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint private String managedName; + private MetricsCaptor metricsCaptor; + + private TimerFacade successTimer; + private volatile AbstractEndpoint replyMessageCorrelator; private volatile boolean initialized; @@ -327,6 +338,11 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint return IntegrationPatternType.inbound_gateway; } + @Override + public void registerMetricsCaptor(MetricsCaptor metricsCaptorToRegister) { + this.metricsCaptor = metricsCaptorToRegister; + } + @Override protected void onInit() { Assert.state(!(this.requestChannelName != null && this.requestChannel != null), @@ -404,11 +420,20 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint MessageChannel channel = getRequestChannel(); Assert.state(channel != null, "send is not supported, because no request channel has been configured"); + SampleFacade sample = null; + if (this.metricsCaptor != null) { + sample = this.metricsCaptor.start(); + } try { - // TODO Micrometer counter this.messagingTemplate.convertAndSend(channel, object, this.historyWritingPostProcessor); + if (sample != null) { + sample.stop(sendTimer()); + } } catch (Exception e) { + if (sample != null) { + sample.stop(buildSendTimer(false, e.getClass().getSimpleName())); + } MessageChannel errorChan = getErrorChannel(); if (errorChan != null) { this.messagingTemplate.send(errorChan, new ErrorMessage(e)); @@ -479,8 +504,11 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint Object reply; Message requestMessage = null; + SampleFacade sample = null; try { - // TODO Micrometer counter + if (this.metricsCaptor != null) { + sample = this.metricsCaptor.start(); + } if (shouldConvert) { reply = this.messagingTemplate.convertSendAndReceive(channel, object, Object.class, this.historyWritingPostProcessor); @@ -496,12 +524,18 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint if (reply == null && this.errorOnTimeout) { throwMessageTimeoutException(object, "No reply received within timeout"); } + if (sample != null) { + sample.stop(sendTimer()); + } } catch (Exception ex) { if (logger.isDebugEnabled()) { logger.debug("failure occurred in gateway sendAndReceive: " + ex.getMessage()); } reply = ex; + if (sample != null) { + sample.stop(buildSendTimer(false, ex.getClass().getSimpleName())); + } } if (reply instanceof Throwable || reply instanceof ErrorMessage) { @@ -644,10 +678,11 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint private Mono> buildReplyMono(Message requestMessage, Mono> reply, boolean error, @Nullable Object originalReplyChannelHeader, @Nullable Object originalErrorChannelHeader) { + MetricsCaptor captor = this.metricsCaptor; return reply .doOnSubscribe(s -> { - if (!error) { - // TODO Micrometer counter + if (!error && captor != null) { + captor.start().stop(sendTimer()); } }) .>map(replyMessage -> { @@ -690,6 +725,25 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint } } + protected TimerFacade sendTimer() { + if (this.successTimer == null) { + this.successTimer = buildSendTimer(true, "none"); + } + return this.successTimer; + } + + protected TimerFacade buildSendTimer(boolean success, String exception) { + TimerFacade timer = this.metricsCaptor.timerBuilder(SEND_TIMER_NAME) + .tag("type", "source") + .tag("name", getComponentName() == null ? "unknown" : getComponentName()) + .tag("result", success ? "success" : "failure") + .tag("exception", exception) + .description("Send processing time") + .build(); + this.timers.add(timer); + return timer; + } + private long sendTimeout(Message requestMessage) { Long sendTimeout = headerToLong(requestMessage.getHeaders() .get(this.messagingTemplate.getSendTimeoutHeader())); @@ -802,6 +856,13 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint } } + @Override + public void destroy() { + this.timers.forEach(MeterFacade::remove); + this.timers.clear(); + super.destroy(); + } + private static class DefaultRequestMapper implements InboundMessageMapper { private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory(); 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 600ca1e52a..7ddbc34694 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 @@ -20,6 +20,7 @@ import org.springframework.beans.factory.DisposableBean; import org.springframework.integration.support.context.NamedComponent; import org.springframework.integration.support.management.metrics.MetricsCaptor; import org.springframework.jmx.export.annotation.ManagedAttribute; +import org.springframework.lang.Nullable; /** * Base interface for Integration managed components. @@ -41,14 +42,17 @@ public interface IntegrationManagement extends NamedComponent, DisposableBean { * @param enabled dalse to disable. */ @ManagedAttribute(description = "Use to disable debug logging during normal message flow") - void setLoggingEnabled(boolean enabled); + default void setLoggingEnabled(boolean enabled) { + } /** * Return whether logging is enabled. * @return true if enabled. */ @ManagedAttribute - boolean isLoggingEnabled(); + default boolean isLoggingEnabled() { + return true; + } default void setManagedName(String managedName) { } @@ -69,7 +73,10 @@ public interface IntegrationManagement extends NamedComponent, DisposableBean { * @return the overrides. * @since 5.0 */ - ManagementOverrides getOverrides(); + @Nullable + default ManagementOverrides getOverrides() { + return null; + } /** * Inject a {@link MetricsCaptor} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/support/management/micrometer/MicrometerMetricsTests.java b/spring-integration-core/src/test/java/org/springframework/integration/support/management/micrometer/MicrometerMetricsTests.java index 6ab50407ac..b211c91fc5 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/support/management/micrometer/MicrometerMetricsTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/support/management/micrometer/MicrometerMetricsTests.java @@ -24,12 +24,14 @@ import java.util.Set; import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.support.BeanDefinitionRegistry; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.integration.annotation.EndpointId; +import org.springframework.integration.annotation.Gateway; import org.springframework.integration.annotation.ServiceActivator; import org.springframework.integration.channel.AbstractMessageChannel; import org.springframework.integration.channel.AbstractPollableChannel; @@ -39,7 +41,11 @@ import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.config.EnableIntegrationManagement; import org.springframework.integration.core.MessageSource; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.integration.endpoint.AbstractMessageSource; +import org.springframework.integration.gateway.GatewayProxyFactoryBean; +import org.springframework.integration.gateway.MessagingGatewaySupport; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; @@ -92,6 +98,13 @@ public class MicrometerMetricsTests { @Autowired private NullChannel nullChannel; + @Autowired + private Gate gates; + + @Autowired + @Qualifier("gatesFlow.gateway") + private Gate gatesFlow; + @SuppressWarnings("unchecked") @Test public void testMicrometerMetrics() { @@ -118,8 +131,8 @@ public class MicrometerMetricsTests { nullChannel.send(message); MeterRegistry registry = this.meterRegistry; - assertThat(registry.get("spring.integration.channels").gauge().value()).isEqualTo(6); - assertThat(registry.get("spring.integration.handlers").gauge().value()).isEqualTo(3); + assertThat(registry.get("spring.integration.channels").gauge().value()).isEqualTo(7); + assertThat(registry.get("spring.integration.handlers").gauge().value()).isEqualTo(4); assertThat(registry.get("spring.integration.sources").gauge().value()).isEqualTo(1); assertThat(registry.get("spring.integration.receive") @@ -175,7 +188,7 @@ public class MicrometerMetricsTests { assertThat(registry.get("spring.integration.send") .tag("name", "nullChannel") .tag("result", "success") - .timer().count()).isEqualTo(1); + .timer().count()).isEqualTo(3); BeanDefinitionRegistry beanFactory = (BeanDefinitionRegistry) this.context.getBeanFactory(); beanFactory.registerBeanDefinition("newChannel", @@ -205,6 +218,23 @@ public class MicrometerMetricsTests { .withStackTraceContaining("A meter with name 'spring.integration.send' was found") .withStackTraceContaining("No meters have a tag 'name' with value 'newChannel'"); + this.gates.oneWay("foo"); + this.gates.twoWay("bar"); + assertThat(registry.get("spring.integration.send") + .tag("name", "gates") + .tag("result", "success") + .timer().count()).isEqualTo(2); + this.gatesFlow.oneWay("foo"); + this.gatesFlow.twoWay("bar"); + assertThat(registry.get("spring.integration.send") + .tag("name", "gatesFlow.gateway") + .tag("result", "success") + .timer().count()).isEqualTo(2); + assertThat(registry.get("spring.integration.send") + .tag("name", "customGw") + .tag("result", "success") + .timer().count()).isEqualTo(2); + this.context.close(); assertThatExceptionOfType(MeterNotFoundException.class) @@ -217,7 +247,6 @@ public class MicrometerMetricsTests { this.channel.destroy(); assertThat(TestUtils.getPropertyValue(this.channel, "meters", Set.class)).hasSize(0); - } @Configuration @@ -289,6 +318,48 @@ public class MicrometerMetricsTests { }; } + @Bean + public GatewayProxyFactoryBean gates() { + GatewayProxyFactoryBean gpfb = new GatewayProxyFactoryBean(Gate.class); + gpfb.setDefaultRequestChannelName("nullChannel"); + return gpfb; + } + + @Bean + IntegrationFlow gatesFlow() { + return IntegrationFlows.from(Gate.class) + .nullChannel(); + } + + @Bean + MessagingGatewaySupport customGw(NullChannel nullChannel) { + return new MessagingGatewaySupport() { + + @Override + protected void onInit() { + setRequestChannel(nullChannel); + setReplyTimeout(0L); + super.onInit(); + } + + @Override + protected void doStart() { + send("foo"); + sendAndReceive("bar"); + } + + }; + } + + } + + public interface Gate { + + void oneWay(String in); + + @Gateway(replyTimeout = 0) + String twoWay(String in); + } }