diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java index 28e806a47a..1a6ad45734 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java @@ -29,7 +29,9 @@ import org.springframework.amqp.rabbit.core.RabbitAdmin; 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.CounterFacade; import org.springframework.integration.support.management.PollableChannelManagement; +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 volatile 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); } @@ -194,11 +200,14 @@ public class PollableAmqpChannel extends AbstractAmqpChannel } return null; } + Message message; if (countsEnabled) { + if (getMetricsCaptor() != null) { + incrementReceiveCounter(); + } getMetrics().afterReceive(); counted = true; } - Message message; if (object instanceof Message) { message = (Message) object; } @@ -210,6 +219,7 @@ public class PollableAmqpChannel extends AbstractAmqpChannel if (isLoggingEnabled() && logger.isDebugEnabled()) { logger.debug("postReceive on channel '" + this + "', message: " + message); } + if (interceptorStack != null) { message = interceptorList.postReceive(message, this); interceptorList.afterReceiveCompletion(message, this, null, interceptorStack); @@ -218,6 +228,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) { @@ -266,6 +286,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 interceptors) { super.setInterceptors(interceptors); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java index 6ffaab08e2..eb3b4e65b6 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java @@ -22,11 +22,11 @@ import java.util.List; import org.springframework.integration.support.management.CounterFacade; import org.springframework.integration.support.management.PollableChannelManagement; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.ChannelInterceptor; import org.springframework.messaging.support.ExecutorChannelInterceptor; -import org.springframework.util.CollectionUtils; /** * Base class for all pollable channels. @@ -95,7 +95,7 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel boolean counted = false; boolean countsEnabled = isCountsEnabled(); try { - if (logger.isTraceEnabled()) { + if (isLoggingEnabled() && logger.isTraceEnabled()) { logger.trace("preReceive on channel '" + this + "'"); } if (interceptorList.getSize() > 0) { @@ -105,21 +105,28 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel return null; } } - Message message = this.doReceive(timeout); - if (countsEnabled && message != null) { - if (getMetricsCaptor() != null) { - incrementReceiveCounter(); + Message message = doReceive(timeout); + if (message == null) { + if (isLoggingEnabled() && logger.isTraceEnabled()) { + logger.trace("postReceive on channel '" + this + "', message is null"); } - getMetrics().afterReceive(); - counted = true; } - if (message != null && logger.isDebugEnabled()) { - logger.debug("postReceive on channel '" + this + "', message: " + message); + else { + if (countsEnabled) { + if (getMetricsCaptor() != null) { + incrementReceiveCounter(); + } + getMetrics().afterReceive(); + counted = true; + } + + if (isLoggingEnabled() && logger.isDebugEnabled()) { + logger.debug("postReceive on channel '" + this + "', message: " + message); + } + } - else if (logger.isTraceEnabled()) { - logger.trace("postReceive on channel '" + this + "', message is null"); - } - if (!CollectionUtils.isEmpty(interceptorStack)) { + + if (interceptorStack != null) { message = interceptorList.postReceive(message, this); interceptorList.afterReceiveCompletion(message, this, null, interceptorStack); } @@ -139,7 +146,7 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel } getMetrics().afterError(); } - if (!CollectionUtils.isEmpty(interceptorStack)) { + if (interceptorStack != null) { interceptorList.afterReceiveCompletion(null, this, e, interceptorStack); } throw e; @@ -214,10 +221,10 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel * return immediately with or without success). A negative timeout value * indicates that the method should block until either a message is * available or the blocking thread is interrupted. - * * @param timeout The timeout. * @return The message, or null. */ + @Nullable protected abstract Message doReceive(long timeout); } diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/PollableJmsChannel.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/PollableJmsChannel.java index b54509fa79..d60a216328 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/PollableJmsChannel.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/PollableJmsChannel.java @@ -21,8 +21,10 @@ import java.util.Deque; import java.util.List; import org.springframework.integration.channel.ExecutorChannelInterceptorAware; +import org.springframework.integration.support.management.CounterFacade; import org.springframework.integration.support.management.PollableChannelManagement; import org.springframework.jms.core.JmsTemplate; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.ChannelInterceptor; @@ -40,6 +42,8 @@ public class PollableJmsChannel extends AbstractJmsChannel private volatile String messageSelector; + private CounterFacade receiveCounter; + private volatile int executorInterceptorsSize; public PollableJmsChannel(JmsTemplate jmsTemplate) { @@ -71,17 +75,30 @@ public class PollableJmsChannel extends AbstractJmsChannel } @Override + @Nullable + public Message receive(long timeout) { + try { + DynamicJmsTemplateProperties.setReceiveTimeout(timeout); + return receive(); + } + finally { + DynamicJmsTemplateProperties.clearReceiveTimeout(); + } + } + + @Override + @Nullable public Message receive() { ChannelInterceptorList interceptorList = getInterceptors(); Deque interceptorStack = null; boolean counted = false; boolean countsEnabled = isCountsEnabled(); try { - if (logger.isTraceEnabled()) { + if (isLoggingEnabled() && logger.isTraceEnabled()) { logger.trace("preReceive on channel '" + this + "'"); } if (interceptorList.getInterceptors().size() > 0) { - interceptorStack = new ArrayDeque(); + interceptorStack = new ArrayDeque<>(); if (!interceptorList.preReceive(this, interceptorStack)) { return null; @@ -96,23 +113,28 @@ public class PollableJmsChannel extends AbstractJmsChannel } if (object == null) { - if (logger.isTraceEnabled()) { + if (isLoggingEnabled() && logger.isTraceEnabled()) { logger.trace("postReceive on channel '" + this + "', message is null"); } return null; } + Message message = null; if (countsEnabled) { + if (getMetricsCaptor() != null) { + incrementReceiveCounter(); + } getMetrics().afterReceive(); counted = true; } - Message message = null; if (object instanceof Message) { message = (Message) object; } else { - message = getMessageBuilderFactory().withPayload(object).build(); + message = getMessageBuilderFactory() + .withPayload(object) + .build(); } - if (logger.isDebugEnabled()) { + if (isLoggingEnabled() && logger.isDebugEnabled()) { logger.debug("postReceive on channel '" + this + "', message: " + message); } if (interceptorStack != null) { @@ -123,6 +145,16 @@ public class PollableJmsChannel extends AbstractJmsChannel } 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) { @@ -132,15 +164,17 @@ public class PollableJmsChannel extends AbstractJmsChannel } } - @Override - public Message receive(long timeout) { - try { - DynamicJmsTemplateProperties.setReceiveTimeout(timeout); - return this.receive(); - } - finally { - DynamicJmsTemplateProperties.clearReceiveTimeout(); + 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