INT-3831: MessageGatewaySupport Statistics

JIRA: https://jira.spring.io/browse/INT-3831

Implement `MessageSourceMetrics` in MGS.

spring-integration-core/src/main/java/org/springframework/integration/aggregator/BarrierMessageHandler.java

@@ -106,6 +106,11 @@ public class BarrierMessageHandler extends AbstractReplyProducingMessageHandler
 		this.timeout = timeout;
 	}
 
+	@Override
+	public String getComponentType() {
+		return "barrier";
+	}
+
 	@Override
 	protected Object handleRequestMessage(Message<?> requestMessage) {
 		Object key = this.correlationStrategy.getCorrelationKey(requestMessage);

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

@@ -16,6 +16,8 @@
 
 package org.springframework.integration.gateway;
 
+import java.util.concurrent.atomic.AtomicLong;
+
 import org.springframework.integration.MessageTimeoutException;
 import org.springframework.integration.core.MessagingTemplate;
 import org.springframework.integration.endpoint.AbstractEndpoint;
@@ -28,6 +30,7 @@ import org.springframework.integration.mapping.OutboundMessageMapper;
 import org.springframework.integration.support.DefaultMessageBuilderFactory;
 import org.springframework.integration.support.MessageBuilderFactory;
 import org.springframework.integration.support.converter.SimpleMessageConverter;
+import org.springframework.integration.support.management.MessageSourceMetrics;
 import org.springframework.integration.support.management.TrackableComponent;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
@@ -47,7 +50,8 @@ import org.springframework.util.Assert;
  * @author Gary Russell
  * @author Artem Bilan
  */
-public abstract class MessagingGatewaySupport extends AbstractEndpoint implements TrackableComponent {
+public abstract class MessagingGatewaySupport extends AbstractEndpoint
+		implements TrackableComponent, MessageSourceMetrics {
 
 	private static final long DEFAULT_TIMEOUT = 1000L;
 
@@ -62,6 +66,8 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement
 
 	private final boolean errorOnTimeout;
 
+	private final AtomicLong messageCount = new AtomicLong();
+
 	private volatile MessageChannel requestChannel;
 
 	private volatile String requestChannelName;
@@ -83,6 +89,14 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement
 
 	private volatile AbstractEndpoint replyMessageCorrelator;
 
+	private volatile String managedType;
+
+	private volatile String managedName;
+
+	private volatile boolean countsEnabled;
+
+	private volatile boolean loggingEnabled = true;
+
 
 	/**
 	 * Construct an instance that will return null if no reply is received.
@@ -218,11 +232,61 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement
 		this.historyWritingPostProcessor.setShouldTrack(shouldTrack);
 	}
 
+	@Override
+	public int getMessageCount() {
+		return (int) this.messageCount.get();
+	}
+
+	@Override
+	public long getMessageCountLong() {
+		return this.messageCount.get();
+	}
+
+	@Override
+	public void setManagedName(String name) {
+		this.managedName = name;
+	}
+
+	@Override
+	public String getManagedName() {
+		return this.managedName;
+	}
+
+	@Override
+	public void setManagedType(String type) {
+		this.managedType = type;
+	}
+
+	@Override
+	public String getManagedType() {
+		return this.managedType;
+	}
+
 	@Override
 	public String getComponentType() {
 		return "gateway";
 	}
 
+	@Override
+	public void setLoggingEnabled(boolean enabled) {
+		this.loggingEnabled = enabled;
+	}
+
+	@Override
+	public boolean isLoggingEnabled() {
+		return this.loggingEnabled;
+	}
+
+	@Override
+	public void setCountsEnabled(boolean countsEnabled) {
+		this.countsEnabled = countsEnabled;
+	}
+
+	@Override
+	public boolean isCountsEnabled() {
+		return this.countsEnabled;
+	}
+
 	@Override
 	protected void onInit() throws Exception {
 		Assert.state(!(this.requestChannelName != null && this.requestChannel != null),
@@ -292,6 +356,9 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement
 		Assert.state(requestChannel != null,
 				"send is not supported, because no request channel has been configured");
 		try {
+			if (this.countsEnabled) {
+				this.messageCount.incrementAndGet();
+			}
 			this.messagingTemplate.convertAndSend(requestChannel, object, this.historyWritingPostProcessor);
 		}
 		catch (Exception e) {
@@ -336,6 +403,9 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement
 		Object reply = null;
 		Throwable error = null;
 		try {
+			if (this.countsEnabled) {
+				this.messageCount.incrementAndGet();
+			}
 			if (shouldConvert) {
 				reply = this.messagingTemplate.convertSendAndReceive(requestChannel, object, null,
 						this.historyWritingPostProcessor);
@@ -462,6 +532,11 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement
 		}
 	}
 
+	@Override
+	public void reset() {
+		this.messageCount.set(0);
+	}
+
 
 	private static class DefaultRequestMapper implements InboundMessageMapper<Object> {
 

spring-integration-core/src/test/java/org/springframework/integration/gateway/MessagingGatewayTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -70,6 +70,7 @@ public class MessagingGatewayTests {
 		this.messagingGateway.setReplyChannel(replyChannel);
 		TestApplicationContext applicationContext = TestUtils.createTestApplicationContext();
 		this.messagingGateway.setBeanFactory(applicationContext);
+		this.messagingGateway.setCountsEnabled(true);
 		this.messagingGateway.afterPropertiesSet();
 		this.messagingGateway.start();
 		applicationContext.refresh();
@@ -83,6 +84,7 @@ public class MessagingGatewayTests {
 		Mockito.when(requestChannel.send(messageMock, 1000L)).thenReturn(true);
 		this.messagingGateway.send(messageMock);
 		Mockito.verify(requestChannel).send(messageMock, 1000L);
+		assertEquals(1, this.messagingGateway.getMessageCount());
 	}
 
 	@Test(expected=MessageDeliveryException.class)