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

Polishing pollable channels around meters

Browse Source
This commit is contained in:
Artem Bilan
2019-02-28 13:22:14 -05:00
committed by Gary Russell
parent 7efb14cf60
commit 3d87ac6463
3 changed files with 13 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -364,12 +364,4 @@ public class PollableAmqpChannel extends AbstractAmqpChannel
return this.executorInterceptorsSize > 0;
}
@Override
public void destroy() {
super.destroy();
if (this.receiveCounter != null) {
this.receiveCounter.remove();
}
}
}

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

@@ -40,7 +40,7 @@ import org.springframework.messaging.support.ExecutorChannelInterceptor;
public abstract class AbstractPollableChannel extends AbstractMessageChannel
implements PollableChannel, PollableChannelManagement, ExecutorChannelInterceptorAware {
private volatile int executorInterceptorsSize;
private int executorInterceptorsSize;
private CounterFacade receiveCounter;
@@ -67,7 +67,6 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel
/**
* Receive the first available message from this channel. If the channel
* contains no messages, this method will block.
*
* @return the first available message or <code>null</code> if the
* receiving thread is interrupted.
*/
@@ -83,9 +82,7 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel
* elapses. If the specified timeout is 0, the method will return
* immediately. If less than zero, it will block indefinitely (see
* {@link #receive()}).
*
* @param timeout the timeout in milliseconds
*
* @return the first available message or <code>null</code> if no message
* is available within the allotted time or the receiving thread is
* interrupted.
@@ -97,8 +94,9 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel
Deque<ChannelInterceptor> interceptorStack = null;
boolean counted = false;
boolean countsEnabled = isCountsEnabled();
boolean traceEnabled = isLoggingEnabled() && logger.isTraceEnabled();
try {
if (isLoggingEnabled() && logger.isTraceEnabled()) {
if (traceEnabled) {
logger.trace("preReceive on channel '" + this + "'");
}
if (interceptorList.getSize() > 0) {
@@ -110,7 +108,7 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel
}
Message<?> message = doReceive(timeout);
if (message == null) {
if (isLoggingEnabled() && logger.isTraceEnabled()) {
if (traceEnabled) {
logger.trace("postReceive on channel '" + this + "', message is null");
}
}
@@ -121,10 +119,9 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel
counted = true;
}
if (isLoggingEnabled() && logger.isDebugEnabled()) {
if (traceEnabled) {
logger.debug("postReceive on channel '" + this + "', message: " + message);
}
}
if (interceptorStack != null && message != null) {
@@ -235,12 +232,4 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel
@Nullable
protected abstract Message<?> doReceive(long timeout);
@Override
public void destroy() {
super.destroy();
if (this.receiveCounter != null) {
this.receiveCounter.remove();
}
}
}

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

@@ -251,13 +251,14 @@ public class NullChannel implements PollableChannel, MessageChannelMetrics,
private TimerFacade sendTimer() {
if (this.successTimer == null) {
this.successTimer = this.metricsCaptor.timerBuilder(SEND_TIMER_NAME)
.tag("type", "channel")
.tag("name", getComponentName() == null ? "nullChannel" : getComponentName())
.tag("result", "success")
.tag("exception", "none")
.description("Subflow process time")
.build();
this.successTimer =
this.metricsCaptor.timerBuilder(SEND_TIMER_NAME)
.tag("type", "channel")
.tag("name", getComponentName() == null ? "nullChannel" : getComponentName())
.tag("result", "success")
.tag("exception", "none")
.description("Subflow process time")
.build();
}
return this.successTimer;
}