GH-3359: Add Timer to MessagingGatewaySupport

Resolves https://github.com/spring-projects/spring-integration/issues/3359

- replace removed simple counter with a `Timer`
- register DSL GPFB as beans so they can be discovered for wiring metrics

* Apply suggestions from code review
* Remove left over deprecation suppression
* Fix import

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<Object, String> getComponentsToRegister() {
+		return Collections.singletonMap(this.gatewayProxyFactoryBean, this.gatewayProxyFactoryBean.getBeanName()
+				+ ".gateway");
+	}
+
 }

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<Object>, MethodInterceptor, BeanClassLoaderAware {
+		implements TrackableComponent, FactoryBean<Object>, 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;

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<TimerFacade> 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<Message<?>> buildReplyMono(Message<?> requestMessage, Mono<Message<?>> 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());
 					}
 				})
 				.<Message<?>>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<Object> {
 
 		private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();

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}

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);
+
 	}
 
 }