INT-4418: Micrometer docs and polishing

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

Add docs for the rework and optimize `Meter` creation.

Polishing

* Polishing according PR comments

* Fix typo in `dsl.adoc`

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

@@ -92,6 +92,10 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
 
 	private MeterRegistry meterRegistry;
 
+	private Timer successTimer;
+
+	private Timer failureTimer;
+
 	public AbstractMessageChannel() {
 		this.interceptors = new ChannelInterceptorList(logger);
 	}
@@ -445,7 +449,7 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
 				logger.debug("preSend on channel '" + this + "', message: " + message);
 			}
 			if (interceptors.getSize() > 0) {
-				interceptorStack = new ArrayDeque<ChannelInterceptor>();
+				interceptorStack = new ArrayDeque<>();
 				message = interceptors.preSend(message, this, interceptorStack);
 				if (message == null) {
 					return false;
@@ -458,13 +462,7 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
 				}
 				sent = doSend(message, timeout);
 				if (sample != null) {
-					sample.stop(Timer.builder(SEND_TIMER_NAME)
-							.tag("type", "channel")
-							.tag("name", getComponentName() == null ? "unknown" : getComponentName())
-							.tag("result", sent ? "success" : "failure")
-							.tag("exception", "none")
-							.description("Subflow process time")
-							.register(this.meterRegistry));
+					sample.stop(sendTimer(sent));
 				}
 				channelMetrics.afterSend(metrics, sent);
 				metricsProcessed = true;
@@ -485,13 +483,7 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
 		catch (Exception e) {
 			if (countsEnabled && !metricsProcessed) {
 				if (sample != null) {
-					sample.stop(Timer.builder(SEND_TIMER_NAME)
-							.tag("type", "channel")
-							.tag("name", getComponentName() == null ? "unknown" : getComponentName())
-							.tag("result", "failure")
-							.tag("exception", e.getClass().getSimpleName())
-							.description("Subflow process time")
-							.register(this.meterRegistry));
+					sample.stop(buildSendTimer(false, e.getClass().getSimpleName()));
 				}
 				channelMetrics.afterSend(metrics, false);
 			}
@@ -506,6 +498,31 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
 		}
 	}
 
+	private Timer sendTimer(boolean sent) {
+		if (sent) {
+			if (this.successTimer == null) {
+				this.successTimer = buildSendTimer(true, "none");
+			}
+			return this.successTimer;
+		}
+		else {
+			if (this.failureTimer == null) {
+				this.failureTimer = buildSendTimer(false, "none");
+			}
+			return this.failureTimer;
+		}
+	}
+
+	private Timer buildSendTimer(boolean success, String exception) {
+		return Timer.builder(SEND_TIMER_NAME)
+				.tag("type", "channel")
+				.tag("name", getComponentName() == null ? "unknown" : getComponentName())
+				.tag("result", success ? "success" : "failure")
+				.tag("exception", exception)
+				.description("Send processing time")
+				.register(this.meterRegistry);
+	}
+
 	private Message<?> convertPayloadIfNecessary(Message<?> message) {
 		// first pass checks if the payload type already matches any of the datatypes
 		for (Class<?> datatype : this.datatypes) {

spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java

@@ -42,6 +42,8 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel
 
 	private volatile int executorInterceptorsSize;
 
+	private Counter receiveCounter;
+
 	@Override
 	public int getReceiveCount() {
 		return getMetrics().getReceiveCount();
@@ -107,13 +109,7 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel
 			Message<?> message = this.doReceive(timeout);
 			if (countsEnabled && message != null) {
 				if (getMeterRegistry() != null) {
-					Counter.builder(RECEIVE_COUNTER_NAME)
-						.tag("name", getComponentName())
-						.tag("type", "channel")
-						.tag("result", "success")
-						.tag("exception", "none")
-						.description("Messages received")
-						.register(getMeterRegistry()).increment();
+					incrementReceiveCounter();
 				}
 				getMetrics().afterReceive();
 				counted = true;
@@ -134,12 +130,13 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel
 			if (countsEnabled && !counted) {
 				if (getMeterRegistry() != null) {
 					Counter.builder(RECEIVE_COUNTER_NAME)
-						.tag("name", getComponentName() == null ? "unknown" : getComponentName())
-						.tag("type", "channel")
-						.tag("result", "failure")
-						.tag("exception", e.getClass().getSimpleName())
-						.description("Messages received")
-						.register(getMeterRegistry()).increment();
+							.tag("name", getComponentName() == null ? "unknown" : getComponentName())
+							.tag("type", "channel")
+							.tag("result", "failure")
+							.tag("exception", e.getClass().getSimpleName())
+							.description("Messages received")
+							.register(getMeterRegistry())
+							.increment();
 				}
 				getMetrics().afterError();
 			}
@@ -150,6 +147,19 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel
 		}
 	}
 
+	private void incrementReceiveCounter() {
+		if (this.receiveCounter == null) {
+			this.receiveCounter = Counter.builder(RECEIVE_COUNTER_NAME)
+					.tag("name", getComponentName())
+					.tag("type", "channel")
+					.tag("result", "success")
+					.tag("exception", "none")
+					.description("Messages received")
+					.register(getMeterRegistry());
+		}
+		this.receiveCounter.increment();
+	}
+
 	@Override
 	public void setInterceptors(List<ChannelInterceptor> interceptors) {
 		super.setInterceptors(interceptors);

spring-integration-core/src/main/java/org/springframework/integration/channel/NullChannel.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -45,6 +45,7 @@ import io.micrometer.core.instrument.Timer;
  *
  * @author Mark Fisher
  * @author Gary Russell
+ * @author Artyem Bilan
  */
 @IntegrationManagedResource
 public class NullChannel implements PollableChannel, MessageChannelMetrics,
@@ -66,6 +67,8 @@ public class NullChannel implements PollableChannel, MessageChannelMetrics,
 
 	private MeterRegistry meterRegistry;
 
+	private Timer successTimer;
+
 	@Override
 	public void setBeanName(String beanName) {
 		this.beanName = beanName;
@@ -221,6 +224,11 @@ public class NullChannel implements PollableChannel, MessageChannelMetrics,
 		return this.managementOverrides;
 	}
 
+	@Override
+	public boolean send(Message<?> message, long timeout) {
+		return send(message);
+	}
+
 	@Override
 	public boolean send(Message<?> message) {
 		if (this.loggingEnabled && this.logger.isDebugEnabled()) {
@@ -228,22 +236,24 @@ public class NullChannel implements PollableChannel, MessageChannelMetrics,
 		}
 		if (this.countsEnabled) {
 			if (this.meterRegistry != null) {
-				Timer.builder(SEND_TIMER_NAME)
-						.tag("type", "channel")
-						.tag("name", getComponentName() == null ? "unknown" : getComponentName())
-						.tag("result", "success")
-						.tag("exception", "none")
-						.description("Subflow process time")
-						.register(this.meterRegistry).record(0, TimeUnit.MILLISECONDS);
+				sendTimer().record(0, TimeUnit.MILLISECONDS);
 			}
 			this.channelMetrics.afterSend(this.channelMetrics.beforeSend(), true);
 		}
 		return true;
 	}
 
-	@Override
-	public boolean send(Message<?> message, long timeout) {
-		return this.send(message);
+	private Timer sendTimer() {
+		if (this.successTimer == null) {
+			this.successTimer = Timer.builder(SEND_TIMER_NAME)
+					.tag("type", "channel")
+					.tag("name", getComponentName() == null ? "unknown" : getComponentName())
+					.tag("result", "success")
+					.tag("exception", "none")
+					.description("Subflow process time")
+					.register(this.meterRegistry);
+		}
+		return this.successTimer;
 	}
 
 	@Override

spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractMessageSource.java

@@ -60,14 +60,14 @@ public abstract class AbstractMessageSource<T> extends AbstractExpressionEvaluat
 
 	private String managedName;
 
-	private Counter counter;
-
 	private volatile boolean countsEnabled;
 
 	private volatile boolean loggingEnabled = true;
 
 	private MeterRegistry meterRegistry;
 
+	private Counter receiveCounter;
+
 	public void setHeaderExpressions(Map<String, Expression> headerExpressions) {
 		this.headerExpressions = (headerExpressions != null)
 				? headerExpressions : Collections.emptyMap();
@@ -130,11 +130,6 @@ public abstract class AbstractMessageSource<T> extends AbstractExpressionEvaluat
 		this.managementOverrides.loggingConfigured = true;
 	}
 
-	@Override
-	public void setCounter(Counter counter) {
-		this.counter = counter;
-	}
-
 	@Override
 	public void reset() {
 		this.messageCount.set(0);
@@ -200,19 +195,26 @@ public abstract class AbstractMessageSource<T> extends AbstractExpressionEvaluat
 		}
 		if (this.countsEnabled && message != null) {
 			if (this.meterRegistry != null) {
-				Counter.builder(RECEIVE_COUNTER_NAME)
-					.tag("name", getComponentName() == null ? "unknown" : getComponentName())
-					.tag("type", "source")
-					.tag("result", "success")
-					.tag("exception", "none")
-					.description("Messages received")
-					.register(this.meterRegistry).increment();
+				incrementReceiveCounter();
 			}
 			this.messageCount.incrementAndGet();
 		}
 		return message;
 	}
 
+	private void incrementReceiveCounter() {
+		if (this.receiveCounter == null) {
+			this.receiveCounter = Counter.builder(RECEIVE_COUNTER_NAME)
+				.tag("name", getComponentName() == null ? "unknown" : getComponentName())
+				.tag("type", "source")
+				.tag("result", "success")
+				.tag("exception", "none")
+				.description("Messages received")
+				.register(this.meterRegistry);
+		}
+		this.receiveCounter.increment();
+	}
+
 	private Map<String, Object> evaluateHeaders() {
 		Map<String, Object> results = new HashMap<>();
 		for (Map.Entry<String, Expression> entry : this.headerExpressions.entrySet()) {

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

@@ -50,11 +50,12 @@ import reactor.core.CoreSubscriber;
  * @author Mark Fisher
  * @author Oleg Zhurakousky
  * @author Gary Russell
+ * @author Artem Bilan
  */
 @IntegrationManagedResource
-public abstract class AbstractMessageHandler extends IntegrationObjectSupport implements MessageHandler,
-		MessageHandlerMetrics, ConfigurableMetricsAware<AbstractMessageHandlerMetrics>, TrackableComponent, Orderable,
-		CoreSubscriber<Message<?>> {
+public abstract class AbstractMessageHandler extends IntegrationObjectSupport
+		implements MessageHandler, MessageHandlerMetrics, ConfigurableMetricsAware<AbstractMessageHandlerMetrics>,
+		TrackableComponent, Orderable, CoreSubscriber<Message<?>> {
 
 	private final ManagementOverrides managementOverrides = new ManagementOverrides();
 
@@ -76,6 +77,8 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im
 
 	private MeterRegistry meterRegistry;
 
+	private Timer successTimer;
+
 	@Override
 	public boolean isLoggingEnabled() {
 		return this.loggingEnabled;
@@ -147,19 +150,13 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im
 		}
 		try {
 			if (this.shouldTrack) {
-				message = MessageHistory.write(message, this, this.getMessageBuilderFactory());
+				message = MessageHistory.write(message, this, getMessageBuilderFactory());
 			}
 			if (countsEnabled) {
 				start = handlerMetrics.beforeHandle();
 				handleMessageInternal(message);
 				if (this.meterRegistry != null) {
-					sample.stop(Timer.builder(SEND_TIMER_NAME)
-							.tag("type", "handler")
-							.tag("name", getComponentName() == null ? "unknown" : getComponentName())
-							.tag("result", "success")
-							.tag("exception", "none")
-							.description("Subflow process time")
-							.register(this.meterRegistry));
+					sample.stop(sendTimer());
 				}
 				handlerMetrics.afterHandle(start, true);
 			}
@@ -169,13 +166,7 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im
 		}
 		catch (Exception e) {
 			if (sample != null) {
-				sample.stop(Timer.builder(SEND_TIMER_NAME)
-						.tag("type", "handler")
-						.tag("name", getComponentName() == null ? "unknown" : getComponentName())
-						.tag("result", "failure")
-						.tag("exception", e.getClass().getSimpleName())
-						.description("Subflow process time")
-						.register(this.meterRegistry));
+				sample.stop(buildSendTimer(false, e.getClass().getSimpleName()));
 			}
 			if (countsEnabled) {
 				handlerMetrics.afterHandle(start, false);
@@ -187,6 +178,23 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im
 		}
 	}
 
+	private Timer sendTimer() {
+		if (this.successTimer == null) {
+			this.successTimer = buildSendTimer(true, "none");
+		}
+		return this.successTimer;
+	}
+
+	private Timer buildSendTimer(boolean success, String exception) {
+		return Timer.builder(SEND_TIMER_NAME)
+				.tag("type", "handler")
+				.tag("name", getComponentName() == null ? "unknown" : getComponentName())
+				.tag("result", success ? "success" : "failure")
+				.tag("exception", exception)
+				.description("Send processing time")
+				.register(this.meterRegistry);
+	}
+
 	@Override
 	public void onSubscribe(Subscription subscription) {
 		Assert.notNull(subscription, "'subscription' must not be null");

spring-integration-core/src/main/java/org/springframework/integration/support/management/LifecycleMessageSourceMetrics.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -20,14 +20,14 @@ import org.springframework.context.Lifecycle;
 import org.springframework.jmx.export.annotation.ManagedAttribute;
 import org.springframework.jmx.export.annotation.ManagedOperation;
 
-import io.micrometer.core.instrument.Counter;
-
 /**
- * A {@link MessageSourceMetrics} that exposes in addition the {@link Lifecycle} interface. The lifecycle methods can
- * be used to start and stop polling endpoints, for instance, in a live system.
+ * A {@link MessageSourceMetrics} that exposes in addition the {@link Lifecycle} interface.
+ * The lifecycle methods can be used to start and stop polling endpoints, for instance, in a live system.
  *
  * @author Dave Syer
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 2.0
  */
 @IntegrationManagedResource
@@ -126,9 +126,4 @@ public class LifecycleMessageSourceMetrics implements MessageSourceMetrics, Life
 		return this.delegate.getOverrides();
 	}
 
-	@Override
-	public void setCounter(Counter counter) {
-		this.delegate.setCounter(counter);
-	}
-
 }

spring-integration-core/src/main/java/org/springframework/integration/support/management/MessageSourceMetrics.java

@@ -24,6 +24,8 @@ import io.micrometer.core.instrument.Counter;
 /**
  * @author Dave Syer
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 2.0
  */
 public interface MessageSourceMetrics extends IntegrationManagement {
@@ -53,7 +55,10 @@ public interface MessageSourceMetrics extends IntegrationManagement {
 	 * Set a micrometer counter to count messages produced.
 	 * @param counter the counter.
 	 * @since 5.0.2
+	 * @deprecated in favor of built-in counter registration via {@code MeterRegistry} callbacks.
+	 * Will be remove in the next release.
 	 */
+	@Deprecated
 	default void setCounter(Counter counter) {
 		// no op
 	}

spring-integration-core/src/test/java/org/springframework/integration/support/management/micrometer/MicrometerMetricsTests.java

@@ -32,6 +32,7 @@ import org.springframework.integration.annotation.ServiceActivator;
 import org.springframework.integration.channel.AbstractMessageChannel;
 import org.springframework.integration.channel.AbstractPollableChannel;
 import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.channel.NullChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.config.EnableIntegrationManagement;
@@ -75,6 +76,9 @@ public class MicrometerMetricsTests {
 	@Autowired
 	private PollableChannel badPoll;
 
+	@Autowired
+	private NullChannel nullChannel;
+
 	@Test
 	public void testSend() {
 		GenericMessage<String> message = new GenericMessage<>("foo");
@@ -97,6 +101,7 @@ public class MicrometerMetricsTests {
 		catch (RuntimeException e) {
 			assertThat(e.getMessage()).isEqualTo("badPoll");
 		}
+		nullChannel.send(message);
 		MeterRegistry registry = this.meterRegistry;
 		assertThat(registry.get("spring.integration.channels").gauge().value()).isEqualTo(5);
 		assertThat(registry.get("spring.integration.handlers").gauge().value()).isEqualTo(2);
@@ -137,6 +142,11 @@ public class MicrometerMetricsTests {
 				.tag("result", "success")
 				.counter().count()).isEqualTo(1);
 
+		assertThat(registry.get("spring.integration.send")
+				.tag("name", "nullChannel")
+				.tag("result", "success")
+				.timer().count()).isEqualTo(1);
+
 		BeanDefinitionRegistry beanFactory = (BeanDefinitionRegistry) this.context.getBeanFactory();
 		beanFactory.registerBeanDefinition("newChannel",
 				BeanDefinitionBuilder.genericBeanDefinition(DirectChannel.class).getRawBeanDefinition());