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

INT-3656: Restore PreSend Log; Avoid Access Method

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3656

The optimization (when no interceptors) inadvertently removed
the `preSend` DEBUG log.

Reinstate the DEBUG log.

Also change `ChannelInterceptorList` to a `static` class.

This avoids the access method invocation each time the channel's
`logger` is used. The only `AMC` field used was the `logger`; pass
it into the `CIL` via a constructor.

Old bytecode:

       104: aload_0
       105: getfield      #14                 // Field this$0:Lorg/springframework/integration/channel/AbstractMessageChannel;
       108: invokestatic  #60                 // Method org/springframework/integration/channel/AbstractMessageChannel.access$1:(Lorg/springframework/integration/channel/AbstractMessageChannel;)Lorg/apache/commons/logging/Log;
       111: invokeinterface #66,  1           // InterfaceMethod org/apache/commons/logging/Log.isDebugEnabled:()Z
       116: ifeq          157

New bytecode:

        98: aload_0
        99: getfield      #22                 // Field logger:Lorg/apache/commons/logging/Log;
       102: invokeinterface #60,  1           // InterfaceMethod org/apache/commons/logging/Log.isDebugEnabled:()Z
This commit is contained in:
Gary Russell
2015-03-04 11:20:30 -05:00
parent 4511b42da8
commit b7ba238860
6 changed files with 104 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -127,6 +127,9 @@ public class PollableAmqpChannel extends AbstractAmqpChannel implements Pollable
boolean counted = false;
boolean countsEnabled = isCountsEnabled();
try {
if (logger.isTraceEnabled()) {
logger.trace("preReceive on channel '" + this + "'");
}
if (interceptorList.getInterceptors().size() > 0) {
interceptorStack = new ArrayDeque<ChannelInterceptor>();
@@ -136,6 +139,9 @@ public class PollableAmqpChannel extends AbstractAmqpChannel implements Pollable
}
Object object = getAmqpTemplate().receiveAndConvert(this.queueName);
if (object == null) {
if (logger.isTraceEnabled()) {
logger.trace("postReceive on channel '" + this + "', message is null");
}
return null;
}
if (countsEnabled) {
@@ -149,8 +155,11 @@ public class PollableAmqpChannel extends AbstractAmqpChannel implements Pollable
else {
message = getMessageBuilderFactory().withPayload(object).build();
}
message = interceptorList.postReceive(message, this);
if (logger.isDebugEnabled()) {
logger.debug("postReceive on channel '" + this + "', message: " + message);
}
if (interceptorStack != null) {
message = interceptorList.postReceive(message, this);
interceptorList.afterReceiveCompletion(message, this, null, interceptorStack);
}
return message;

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

@@ -24,6 +24,8 @@ import java.util.Iterator;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import org.apache.commons.logging.Log;
import org.springframework.core.OrderComparator;
import org.springframework.core.convert.ConversionService;
import org.springframework.integration.channel.management.AbstractMessageChannelMetrics;
@@ -63,7 +65,7 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
implements MessageChannel, TrackableComponent, ChannelInterceptorAware, MessageChannelMetrics,
ConfigurableMetricsAware<AbstractMessageChannelMetrics> {
private final ChannelInterceptorList interceptors = new ChannelInterceptorList();
private final ChannelInterceptorList interceptors;
private final Comparator<Object> orderComparator = new OrderComparator();
@@ -81,6 +83,10 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
private volatile AbstractMessageChannelMetrics channelMetrics = new DefaultMessageChannelMetrics();
public AbstractMessageChannel() {
this.interceptors = new ChannelInterceptorList(logger);
}
@Override
public String getComponentType() {
return "channel";
@@ -407,6 +413,9 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
if (this.datatypes.length > 0) {
message = this.convertPayloadIfNecessary(message);
}
if (logger.isDebugEnabled()) {
logger.debug("preSend on channel '" + this + "', message: " + message);
}
if (interceptors.getInterceptors().size() > 0) {
interceptorStack = new ArrayDeque<ChannelInterceptor>();
message = interceptors.preSend(message, this, interceptorStack);
@@ -422,8 +431,11 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
channelMetrics.afterSend(metrics, sent);
metricsProcessed = true;
}
interceptors.postSend(message, this, sent);
if (logger.isDebugEnabled()) {
logger.debug("postSend (sent=" + sent + ") on channel '" + this + "', message: " + message);
}
if (interceptorStack != null) {
interceptors.postSend(message, this, sent);
interceptors.afterSendCompletion(message, this, sent, null, interceptorStack);
}
return sent;
@@ -490,10 +502,16 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
/**
* A convenience wrapper class for the list of ChannelInterceptors.
*/
protected class ChannelInterceptorList {
protected static class ChannelInterceptorList {
private final Log logger;
private final List<ChannelInterceptor> interceptors = new CopyOnWriteArrayList<ChannelInterceptor>();
public ChannelInterceptorList(Log logger) {
this.logger = logger;
}
public boolean set(List<ChannelInterceptor> interceptors) {
synchronized (this.interceptors) {
this.interceptors.clear();
@@ -511,9 +529,6 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
public Message<?> preSend(Message<?> message, MessageChannel channel,
Deque<ChannelInterceptor> interceptorStack) {
if (logger.isDebugEnabled()) {
logger.debug("preSend on channel '" + channel + "', message: " + message);
}
if (this.interceptors.size() > 0) {
for (ChannelInterceptor interceptor : this.interceptors) {
message = interceptor.preSend(message, channel);
@@ -532,9 +547,6 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
}
public void postSend(Message<?> message, MessageChannel channel, boolean sent) {
if (logger.isDebugEnabled()) {
logger.debug("postSend (sent=" + sent + ") on channel '" + channel + "', message: " + message);
}
if (this.interceptors.size() > 0) {
for (ChannelInterceptor interceptor : interceptors) {
interceptor.postSend(message, channel, sent);
@@ -556,9 +568,6 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
}
public boolean preReceive(MessageChannel channel, Deque<ChannelInterceptor> interceptorStack) {
if (logger.isTraceEnabled()) {
logger.trace("preReceive on channel '" + channel + "'");
}
if (this.interceptors.size() > 0) {
for (ChannelInterceptor interceptor : interceptors) {
if (!interceptor.preReceive(channel)) {
@@ -572,12 +581,6 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
}
public Message<?> postReceive(Message<?> message, MessageChannel channel) {
if (message != null && logger.isDebugEnabled()) {
logger.debug("postReceive on channel '" + channel + "', message: " + message);
}
else if (logger.isTraceEnabled()) {
logger.trace("postReceive on channel '" + channel + "', message is null");
}
if (this.interceptors.size() > 0) {
for (ChannelInterceptor interceptor : interceptors) {
message = interceptor.postReceive(message, channel);

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

@@ -87,6 +87,9 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel imp
boolean counted = false;
boolean countsEnabled = isCountsEnabled();
try {
if (logger.isTraceEnabled()) {
logger.trace("preReceive on channel '" + this + "'");
}
if (interceptorList.getInterceptors().size() > 0) {
interceptorStack = new ArrayDeque<ChannelInterceptor>();
@@ -99,8 +102,14 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel imp
getMetrics().afterReceive();
counted = true;
}
message = interceptorList.postReceive(message, this);
if (message != null && logger.isDebugEnabled()) {
logger.debug("postReceive on channel '" + this + "', message: " + message);
}
else if (logger.isTraceEnabled()) {
logger.trace("postReceive on channel '" + this + "', message is null");
}
if (interceptorStack != null) {
message = interceptorList.postReceive(message, this);
interceptorList.afterReceiveCompletion(message, this, null, interceptorStack);
}
return message;

20
spring-integration-core/src/test/java/org/springframework/integration/channel/DirectChannelTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 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.
@@ -16,16 +16,25 @@
package org.springframework.integration.channel;
import static org.hamcrest.Matchers.startsWith;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.lang.reflect.Method;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.logging.Log;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.context.support.ClassPathXmlApplicationContext;
@@ -50,6 +59,9 @@ public class DirectChannelTests {
@Test
public void testSend() {
DirectChannel channel = new DirectChannel();
Log logger = spy(TestUtils.getPropertyValue(channel, "logger", Log.class));
when(logger.isDebugEnabled()).thenReturn(true);
new DirectFieldAccessor(channel).setPropertyValue("logger", logger);
ThreadNameExtractingTestTarget target = new ThreadNameExtractingTestTarget();
channel.subscribe(target);
GenericMessage<String> message = new GenericMessage<String>("test");
@@ -60,6 +72,12 @@ public class DirectChannelTests {
DirectFieldAccessor dispatcherAccessor = new DirectFieldAccessor(dispatcher);
Object loadBalancingStrategy = dispatcherAccessor.getPropertyValue("loadBalancingStrategy");
assertTrue(loadBalancingStrategy instanceof RoundRobinLoadBalancingStrategy);
ArgumentCaptor<String> captor = ArgumentCaptor.forClass(String.class);
verify(logger, times(2)).debug(captor.capture());
List<String> logs = captor.getAllValues();
assertEquals(2, logs.size());
assertThat(logs.get(0), startsWith("preSend"));
assertThat(logs.get(1), startsWith("postSend"));
}
@Test

34
spring-integration-core/src/test/java/org/springframework/integration/channel/QueueChannelTests.java
View File

@@ -16,10 +16,16 @@
package org.springframework.integration.channel;
import static org.hamcrest.Matchers.startsWith;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.util.List;
import java.util.concurrent.CountDownLatch;
@@ -28,13 +34,17 @@ import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.commons.logging.Log;
import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.mockito.ArgumentCaptor;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.integration.selector.UnexpiredMessageSelector;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.GenericMessage;
@@ -45,6 +55,7 @@ import reactor.io.queue.spec.PersistentQueueSpec;
/**
* @author Mark Fisher
* @author Artem Bilan
* @author Gary Russell
*/
public class QueueChannelTests {
@@ -53,7 +64,12 @@ public class QueueChannelTests {
final AtomicBoolean messageReceived = new AtomicBoolean(false);
final CountDownLatch latch = new CountDownLatch(1);
final QueueChannel channel = new QueueChannel();
Log logger = spy(TestUtils.getPropertyValue(channel, "logger", Log.class));
when(logger.isDebugEnabled()).thenReturn(true);
when(logger.isTraceEnabled()).thenReturn(true);
new DirectFieldAccessor(channel).setPropertyValue("logger", logger);
new Thread(new Runnable() {
@Override
public void run() {
Message<?> message = channel.receive();
if (message != null) {
@@ -66,6 +82,12 @@ public class QueueChannelTests {
channel.send(new GenericMessage<String>("testing"));
latch.await(1000, TimeUnit.MILLISECONDS);
assertTrue(messageReceived.get());
ArgumentCaptor<String> preCaptor = ArgumentCaptor.forClass(String.class);
ArgumentCaptor<String> postCaptor = ArgumentCaptor.forClass(String.class);
verify(logger).trace(preCaptor.capture());
verify(logger, times(3)).debug(postCaptor.capture());
assertThat(preCaptor.getValue(), startsWith("preReceive"));
assertThat(postCaptor.getValue(), startsWith("postReceive"));
}
@Test
@@ -76,6 +98,7 @@ public class QueueChannelTests {
final CountDownLatch latch2 = new CountDownLatch(1);
Executor singleThreadExecutor = Executors.newSingleThreadExecutor();
Runnable receiveTask1 = new Runnable() {
@Override
public void run() {
Message<?> message = channel.receive(0);
if (message != null) {
@@ -85,6 +108,7 @@ public class QueueChannelTests {
}
};
Runnable sendTask = new Runnable() {
@Override
public void run() {
channel.send(new GenericMessage<String>("testing"));
}
@@ -94,6 +118,7 @@ public class QueueChannelTests {
singleThreadExecutor.execute(sendTask);
assertFalse(messageReceived.get());
Runnable receiveTask2 = new Runnable() {
@Override
public void run() {
Message<?> message = channel.receive(0);
if (message != null) {
@@ -113,6 +138,7 @@ public class QueueChannelTests {
final AtomicBoolean receiveInterrupted = new AtomicBoolean(false);
final CountDownLatch latch = new CountDownLatch(1);
Thread t = new Thread(new Runnable() {
@Override
public void run() {
Message<?> message = channel.receive();
receiveInterrupted.set(true);
@@ -133,6 +159,7 @@ public class QueueChannelTests {
final AtomicBoolean receiveInterrupted = new AtomicBoolean(false);
final CountDownLatch latch = new CountDownLatch(1);
Thread t = new Thread(new Runnable() {
@Override
public void run() {
Message<?> message = channel.receive(10000);
receiveInterrupted.set(true);
@@ -168,6 +195,7 @@ public class QueueChannelTests {
final AtomicBoolean sendInterrupted = new AtomicBoolean(false);
final CountDownLatch latch = new CountDownLatch(1);
Thread t = new Thread(new Runnable() {
@Override
public void run() {
channel.send(new GenericMessage<String>("test-2"));
sendInterrupted.set(true);
@@ -189,6 +217,7 @@ public class QueueChannelTests {
final AtomicBoolean sendInterrupted = new AtomicBoolean(false);
final CountDownLatch latch = new CountDownLatch(1);
Thread t = new Thread(new Runnable() {
@Override
public void run() {
channel.send(new GenericMessage<String>("test-2"), 10000);
sendInterrupted.set(true);
@@ -259,6 +288,7 @@ public class QueueChannelTests {
.get();
final QueueChannel channel = new QueueChannel(queue);
new Thread(new Runnable() {
@Override
public void run() {
Message<?> message = channel.receive();
if (message != null) {
@@ -275,6 +305,7 @@ public class QueueChannelTests {
final CountDownLatch latch1 = new CountDownLatch(2);
Thread thread = new Thread(new Runnable() {
@Override
public void run() {
while (true) {
Message<?> message = channel.receive(100);
@@ -297,6 +328,7 @@ public class QueueChannelTests {
final AtomicBoolean receiveInterrupted = new AtomicBoolean(false);
final CountDownLatch latch2 = new CountDownLatch(1);
Thread t = new Thread(new Runnable() {
@Override
public void run() {
Message<?> message = channel.receive(10000);
receiveInterrupted.set(true);
@@ -313,6 +345,7 @@ public class QueueChannelTests {
receiveInterrupted.set(false);
final CountDownLatch latch3 = new CountDownLatch(1);
t = new Thread(new Runnable() {
@Override
public void run() {
Message<?> message = channel.receive();
receiveInterrupted.set(true);
@@ -344,6 +377,7 @@ public class QueueChannelTests {
// Distributed scenario
final CountDownLatch latch4 = new CountDownLatch(1);
new Thread(new Runnable() {
@Override
public void run() {
Message<?> message = channel.receive();
if (message != null) {

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

@@ -72,6 +72,9 @@ public class PollableJmsChannel extends AbstractJmsChannel implements PollableCh
boolean counted = false;
boolean countsEnabled = isCountsEnabled();
try {
if (logger.isTraceEnabled()) {
logger.trace("preReceive on channel '" + this + "'");
}
if (interceptorList.getInterceptors().size() > 0) {
interceptorStack = new ArrayDeque<ChannelInterceptor>();
@@ -88,6 +91,9 @@ public class PollableJmsChannel extends AbstractJmsChannel implements PollableCh
}
if (object == null) {
if (logger.isTraceEnabled()) {
logger.trace("postReceive on channel '" + this + "', message is null");
}
return null;
}
if (countsEnabled) {
@@ -101,8 +107,11 @@ public class PollableJmsChannel extends AbstractJmsChannel implements PollableCh
else {
message = getMessageBuilderFactory().withPayload(object).build();
}
message = interceptorList.postReceive(message, this);
if (logger.isDebugEnabled()) {
logger.debug("postReceive on channel '" + this + "', message: " + message);
}
if (interceptorStack != null) {
message = interceptorList.postReceive(message, this);
interceptorList.afterReceiveCompletion(message, this, null, interceptorStack);
}
return message;