Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix PollableChannels for Micrometer

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

**Cherry-pick to 5.0.x**
This commit is contained in:
Artem Bilan
2018-08-30 16:12:12 -04:00
committed by Gary Russell
parent 6934690947
commit 35daaae632
3 changed files with 106 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<ChannelInterceptor> interceptors) {
super.setInterceptors(interceptors);

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

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

62
spring-integration-jms/src/main/java/org/springframework/integration/jms/PollableJmsChannel.java
View File

@@ -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<ChannelInterceptor> 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<ChannelInterceptor>();
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