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

Conflicts:
	spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java
	spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java
	spring-integration-core/src/test/java/org/springframework/integration/channel/QueueChannelTests.java
This commit is contained in:
Gary Russell
2015-03-04 11:20:30 -05:00
committed by Artem Bilan
parent 7a4b6b9ffd
commit 1749b0b36a
6 changed files with 108 additions and 24 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

@@ -101,6 +101,9 @@ public class PollableAmqpChannel extends AbstractAmqpChannel implements Pollable
ChannelInterceptorList interceptorList = getInterceptors();
Deque<ChannelInterceptor> interceptorStack = null;
try {
if (logger.isTraceEnabled()) {
logger.trace("preReceive on channel '" + this + "'");
}
if (interceptorList.getInterceptors().size() > 0) {
interceptorStack = new ArrayDeque<ChannelInterceptor>();
@@ -110,6 +113,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;
}
Message<?> message = null;
@@ -119,8 +125,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;

44
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.context.IntegrationContextUtils;
@@ -54,7 +56,7 @@ import org.springframework.util.StringUtils;
public abstract class AbstractMessageChannel extends IntegrationObjectSupport
implements MessageChannel, TrackableComponent, ChannelInterceptorAware {
private final ChannelInterceptorList interceptors = new ChannelInterceptorList();
private final ChannelInterceptorList interceptors;
private final Comparator<Object> orderComparator = new OrderComparator();
@@ -66,6 +68,10 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
private volatile MessageConverter messageConverter;
public AbstractMessageChannel() {
this.interceptors = new ChannelInterceptorList(logger);
}
@Override
public String getComponentType() {
return "channel";
@@ -267,7 +273,10 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
if (this.datatypes.length > 0) {
message = this.convertPayloadIfNecessary(message);
}
if (this.interceptors.getInterceptors().size() > 0) {
if (logger.isDebugEnabled()) {
logger.debug("preSend on channel '" + this + "', message: " + message);
}
if (interceptors.getInterceptors().size() > 0) {
interceptorStack = new ArrayDeque<ChannelInterceptor>();
message = this.interceptors.preSend(message, this, interceptorStack);
if (message == null) {
@@ -275,9 +284,13 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
}
}
sent = this.doSend(message, timeout);
this.interceptors.postSend(message, this, sent);
if (logger.isDebugEnabled()) {
logger.debug("postSend (sent=" + sent + ") on channel '" + this + "', message: " + message);
}
if (interceptorStack != null) {
this.interceptors.afterSendCompletion(message, this, sent, null, interceptorStack);
interceptors.postSend(message, this, sent);
interceptors.afterSendCompletion(message, this, sent, null, interceptorStack);
}
return sent;
}
@@ -337,10 +350,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();
@@ -358,9 +377,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);
@@ -379,9 +395,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);
@@ -403,9 +416,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)) {
@@ -419,12 +429,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

@@ -61,6 +61,9 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel imp
ChannelInterceptorList interceptorList = this.getInterceptors();
Deque<ChannelInterceptor> interceptorStack = null;
try {
if (logger.isTraceEnabled()) {
logger.trace("preReceive on channel '" + this + "'");
}
if (interceptorList.getInterceptors().size() > 0) {
interceptorStack = new ArrayDeque<ChannelInterceptor>();
@@ -69,8 +72,14 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel imp
}
}
Message<?> message = this.doReceive(timeout);
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

35
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,12 +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;
@@ -44,6 +55,7 @@ import reactor.queue.spec.PersistentQueueSpec;
/**
* @author Mark Fisher
* @author Artem Bilan
* @author Gary Russell
*/
public class QueueChannelTests {
@@ -52,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) {
@@ -65,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
@@ -75,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) {
@@ -84,6 +108,7 @@ public class QueueChannelTests {
}
};
Runnable sendTask = new Runnable() {
@Override
public void run() {
channel.send(new GenericMessage<String>("testing"));
}
@@ -93,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) {
@@ -112,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);
@@ -132,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);
@@ -167,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);
@@ -188,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);
@@ -257,6 +287,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) {
@@ -273,6 +304,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);
@@ -295,6 +327,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);
@@ -311,6 +344,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);
@@ -342,6 +376,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

@@ -47,6 +47,9 @@ public class PollableJmsChannel extends AbstractJmsChannel implements PollableCh
ChannelInterceptorList interceptorList = getInterceptors();
Deque<ChannelInterceptor> interceptorStack = null;
try {
if (logger.isTraceEnabled()) {
logger.trace("preReceive on channel '" + this + "'");
}
if (interceptorList.getInterceptors().size() > 0) {
interceptorStack = new ArrayDeque<ChannelInterceptor>();
@@ -63,6 +66,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;
}
Message<?> message = null;
@@ -72,8 +78,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;