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 a431398371..9e29fae6f3 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 @@ -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(); @@ -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; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java index 56d150a187..93ac634f9d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java @@ -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 { - private final ChannelInterceptorList interceptors = new ChannelInterceptorList(); + private final ChannelInterceptorList interceptors; private final Comparator 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(); 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 interceptors = new CopyOnWriteArrayList(); + public ChannelInterceptorList(Log logger) { + this.logger = logger; + } + public boolean set(List 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 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 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); 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 6ae74c4b5d..6798a64a33 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 @@ -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(); @@ -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; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/DirectChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/DirectChannelTests.java index 9346afc210..0b84430b03 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/DirectChannelTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/DirectChannelTests.java @@ -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 message = new GenericMessage("test"); @@ -60,6 +72,12 @@ public class DirectChannelTests { DirectFieldAccessor dispatcherAccessor = new DirectFieldAccessor(dispatcher); Object loadBalancingStrategy = dispatcherAccessor.getPropertyValue("loadBalancingStrategy"); assertTrue(loadBalancingStrategy instanceof RoundRobinLoadBalancingStrategy); + ArgumentCaptor captor = ArgumentCaptor.forClass(String.class); + verify(logger, times(2)).debug(captor.capture()); + List logs = captor.getAllValues(); + assertEquals(2, logs.size()); + assertThat(logs.get(0), startsWith("preSend")); + assertThat(logs.get(1), startsWith("postSend")); } @Test diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/QueueChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/QueueChannelTests.java index f58951d63e..c9cd4a95be 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/QueueChannelTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/QueueChannelTests.java @@ -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("testing")); latch.await(1000, TimeUnit.MILLISECONDS); assertTrue(messageReceived.get()); + ArgumentCaptor preCaptor = ArgumentCaptor.forClass(String.class); + ArgumentCaptor 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("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("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("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) { 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 10f3b37702..0754e418f7 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 @@ -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(); @@ -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;