Fix PollableChannels for Micrometer

* Add Micrometer metrics capture for the
`PollableAmqpChannel` and `PollableJmsChannel`
* Polishing and optimization for the `AbstractPollableChannel.receive()`

**Cherry-pick to 5.0.x**

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java

@@ -30,6 +30,8 @@ import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.integration.amqp.support.AmqpHeaderMapper;
 import org.springframework.integration.channel.ExecutorChannelInterceptorAware;
 import org.springframework.integration.support.management.PollableChannelManagement;
+import org.springframework.integration.support.management.metrics.CounterFacade;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.support.ChannelInterceptor;
@@ -54,6 +56,8 @@ public class PollableAmqpChannel extends AbstractAmqpChannel
 
 	private Queue queue;
 
+	private CounterFacade receiveCounter;
+
 	private volatile int executorInterceptorsSize;
 
 	private volatile boolean declared;
@@ -161,11 +165,13 @@ public class PollableAmqpChannel extends AbstractAmqpChannel
 	}
 
 	@Override
+	@Nullable
 	public Message<?> receive() {
 		return doReceive(null);
 	}
 
 	@Override
+	@Nullable
 	public Message<?> receive(long timeout) {
 		return doReceive(timeout);
 	}
@@ -196,6 +202,9 @@ public class PollableAmqpChannel extends AbstractAmqpChannel
 			}
 			else {
 				if (countsEnabled) {
+					if (getMetricsCaptor() != null) {
+						incrementReceiveCounter();
+					}
 					getMetrics().afterReceive();
 					counted = true;
 				}
@@ -211,6 +220,7 @@ public class PollableAmqpChannel extends AbstractAmqpChannel
 					logger.debug("postReceive on channel '" + this + "', message: " + message);
 				}
 			}
+
 			if (interceptorStack != null) {
 				if (message != null) {
 					message = interceptorList.postReceive(message, this);
@@ -221,6 +231,16 @@ public class PollableAmqpChannel extends AbstractAmqpChannel
 		}
 		catch (RuntimeException e) {
 			if (countsEnabled && !counted) {
+				if (getMetricsCaptor() != null) {
+					getMetricsCaptor().counterBuilder(RECEIVE_COUNTER_NAME)
+							.tag("name", getComponentName() == null ? "unknown" : getComponentName())
+							.tag("type", "channel")
+							.tag("result", "failure")
+							.tag("exception", e.getClass().getSimpleName())
+							.description("Messages received")
+							.build()
+							.increment();
+				}
 				getMetrics().afterError();
 			}
 			if (interceptorStack != null) {
@@ -269,6 +289,20 @@ public class PollableAmqpChannel extends AbstractAmqpChannel
 		}
 	}
 
+	private void incrementReceiveCounter() {
+		if (this.receiveCounter == null) {
+			this.receiveCounter = getMetricsCaptor().counterBuilder(RECEIVE_COUNTER_NAME)
+					.tag("name", getComponentName())
+					.tag("type", "channel")
+					.tag("result", "success")
+					.tag("exception", "none")
+					.description("Messages received")
+					.build();
+		}
+		this.receiveCounter.increment();
+	}
+
+
 	@Override
 	public void setInterceptors(List<ChannelInterceptor> interceptors) {
 		super.setInterceptors(interceptors);