INT-3211 TCP Interceptor Events; Test Polishing

Intercepted TCP connections did not publish events. TcpReceivingChannelAdapterTests did not close server factories after tests. Suppress 'no publisher' log messages.
2013-11-19 10:34:08 -05:00
parent 945cb14407
commit 5bb6e550aa
8 changed files with 204 additions and 36 deletions
--- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorSupport.java
+++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorSupport.java
@@ -16,6 +16,7 @@

 package org.springframework.integration.ip.tcp.connection;

+import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.core.serializer.Deserializer;
 import org.springframework.core.serializer.Serializer;
 import org.springframework.integration.Message;
@@ -38,15 +39,25 @@ public abstract class TcpConnectionInterceptorSupport extends TcpConnectionSuppo

 	private Boolean realSender;

+	public TcpConnectionInterceptorSupport() {
+		super();
+	}
+
+	public TcpConnectionInterceptorSupport(ApplicationEventPublisher applicationEventPublisher) {
+		super(applicationEventPublisher);
+	}
+
 	@Override
 	public void close() {
 		this.theConnection.close();
 	}

+	@Override
 	public boolean isOpen() {
 		return this.theConnection.isOpen();
 	}

+	@Override
 	public Object getPayload() throws Exception {
 		return this.theConnection.getPayload();
 	}
@@ -61,10 +72,12 @@ public abstract class TcpConnectionInterceptorSupport extends TcpConnectionSuppo
 		return this.theConnection.getHostAddress();
 	}

+	@Override
 	public int getPort() {
 		return this.theConnection.getPort();
 	}

+	@Override
 	public Object getDeserializerStateKey() {
 		return this.theConnection.getDeserializerStateKey();
 	}
@@ -91,6 +104,7 @@ public abstract class TcpConnectionInterceptorSupport extends TcpConnectionSuppo
 		return this.theConnection.isSingleUse();
 	}

+	@Override
 	public void run() {
 		this.theConnection.run();
 	}
@@ -130,6 +144,7 @@ public abstract class TcpConnectionInterceptorSupport extends TcpConnectionSuppo
 		return this.theConnection.isServer();
 	}

+	@Override
 	public boolean onMessage(Message<?> message) {
 		if (this.tcpListener == null) {
 			if (message instanceof ErrorMessage) {
@@ -142,6 +157,7 @@ public abstract class TcpConnectionInterceptorSupport extends TcpConnectionSuppo
 		return this.tcpListener.onMessage(message);
 	}

+	@Override
 	public void send(Message<?> message) throws Exception {
 		this.theConnection.send(message);
 	}
@@ -170,12 +186,14 @@ public abstract class TcpConnectionInterceptorSupport extends TcpConnectionSuppo
 		return tcpListener;
 	}

+	@Override
 	public void addNewConnection(TcpConnection connection) {
 		if (this.tcpSender != null) {
 			this.tcpSender.addNewConnection(this);
 		}
 	}

+	@Override
 	public void removeDeadConnection(TcpConnection connection) {
 		if (this.tcpSender != null) {
 			this.tcpSender.removeDeadConnection(this);
--- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java
+++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java
@@ -89,8 +89,12 @@ public abstract class TcpConnectionSupport implements TcpConnection {
 	private volatile boolean noReadErrorOnClose;

 	public TcpConnectionSupport() {
+		this(null);
+	}
+
+	public TcpConnectionSupport(ApplicationEventPublisher applicationEventPublisher) {
 		this.server = false;
-		this.applicationEventPublisher = null;
+		this.applicationEventPublisher = applicationEventPublisher;
 	}

 	/**
@@ -150,6 +154,7 @@ public abstract class TcpConnectionSupport implements TcpConnection {
 	/**
 	 * Closes this connection.
 	 */
+	@Override
 	public void close() {
 		if (this.sender != null) {
 			this.sender.removeDeadConnection(this);
@@ -203,6 +208,7 @@ public abstract class TcpConnectionSupport implements TcpConnection {
 	 *
 	 * @return the deserializer
 	 */
+	@Override
 	public Deserializer<?> getDeserializer() {
 		return this.deserializer;
 	}
@@ -218,6 +224,7 @@ public abstract class TcpConnectionSupport implements TcpConnection {
 	 *
 	 * @return the serializer
 	 */
+	@Override
 	public Serializer<?> getSerializer() {
 		return this.serializer;
 	}
@@ -266,6 +273,7 @@ public abstract class TcpConnectionSupport implements TcpConnection {
 	/**
 	 * @return the listener
 	 */
+	@Override
 	public TcpListener getListener() {
 		return this.listener;
 	}
@@ -289,26 +297,32 @@ public abstract class TcpConnectionSupport implements TcpConnection {
 	 *
 	 * @return True if connection is used once.
 	 */
+	@Override
 	public boolean isSingleUse() {
 		return this.singleUse;
 	}

+	@Override
 	public boolean isServer() {
 		return server;
 	}

+	@Override
 	public long incrementAndGetConnectionSequence() {
 		return this.sequence.incrementAndGet();
 	}

+	@Override
 	public String getHostAddress() {
 		return this.hostAddress;
 	}

+	@Override
 	public String getHostName() {
 		return this.hostName;
 	}

+	@Override
 	public String getConnectionId() {
 		return this.connectionId;
 	}
--- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/AbstractTcpChannelAdapterTests.java
+++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/AbstractTcpChannelAdapterTests.java
@@ -0,0 +1,48 @@
+/*
+ * Copyright 2013 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.ip.tcp;
+
+import org.springframework.context.ApplicationEvent;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory;
+import org.springframework.integration.ip.tcp.connection.HelloWorldInterceptorFactory;
+
+/**
+ * @author Gary Russell
+ * @since 3.0
+ *
+ */
+public class AbstractTcpChannelAdapterTests {
+
+	private static final ApplicationEventPublisher NOOP_PUBLISHER = new ApplicationEventPublisher() {
+
+		@Override
+		public void publishEvent(ApplicationEvent event) {
+		}
+	};
+
+	protected HelloWorldInterceptorFactory newInterceptorFactory() {
+		HelloWorldInterceptorFactory factory = new HelloWorldInterceptorFactory();
+		factory.setApplicationEventPublisher(NOOP_PUBLISHER);
+		return factory;
+	}
+
+	protected void noopPublisher(AbstractConnectionFactory connectionFactory) {
+		connectionFactory.setApplicationEventPublisher(NOOP_PUBLISHER);
+	}
+
+
+}
--- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java
+++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java
@@ -40,6 +40,7 @@ import javax.net.ServerSocketFactory;
 import javax.net.SocketFactory;

 import org.junit.Test;
+
 import org.springframework.core.serializer.DefaultDeserializer;
 import org.springframework.core.serializer.DefaultSerializer;
 import org.springframework.integration.Message;
@@ -49,7 +50,6 @@ import org.springframework.integration.core.SubscribableChannel;
 import org.springframework.integration.handler.ServiceActivatingHandler;
 import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory;
 import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory;
-import org.springframework.integration.ip.tcp.connection.HelloWorldInterceptorFactory;
 import org.springframework.integration.ip.tcp.connection.TcpConnectionInterceptorFactory;
 import org.springframework.integration.ip.tcp.connection.TcpConnectionInterceptorFactoryChain;
 import org.springframework.integration.ip.tcp.connection.TcpNetClientConnectionFactory;
@@ -63,12 +63,13 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
 /**
 * @author Gary Russell
 */
-public class TcpReceivingChannelAdapterTests {
+public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTests {

 	@Test
 	public void testNet() throws Exception {
 		final int port = SocketUtils.findAvailableServerSocket();
 		AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port);
+		noopPublisher(scf);
 		ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
 		scf.setSerializer(serializer);
 		scf.setDeserializer(serializer);
@@ -88,6 +89,7 @@ public class TcpReceivingChannelAdapterTests {
 		message = channel.receive(10000);
 		assertNotNull(message);
 		assertEquals("Test2", new String((byte[]) message.getPayload()));
+		scf.stop();
 	}

 	@Test
@@ -97,6 +99,7 @@ public class TcpReceivingChannelAdapterTests {
 		final CountDownLatch latch2 = new CountDownLatch(1);
 		final AtomicBoolean done = new AtomicBoolean();
 		Executors.newSingleThreadExecutor().execute(new Runnable() {
+			@Override
 			public void run() {
 				try {
 					ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port, 10);
@@ -115,6 +118,7 @@ public class TcpReceivingChannelAdapterTests {
 			}
 		});
 		AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
+		noopPublisher(ccf);
 		ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
 		ccf.setSerializer(serializer);
 		ccf.setDeserializer(serializer);
@@ -142,12 +146,14 @@ public class TcpReceivingChannelAdapterTests {
 		adapter.start();
 		adapter.stop();
 		latch2.countDown();
+		ccf.stop();
 	}

 	@Test
 	public void testNio() throws Exception {
 		final int port = SocketUtils.findAvailableServerSocket();
 		TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port);
+		noopPublisher(scf);
 		ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
 		scf.setSerializer(serializer);
 		scf.setDeserializer(serializer);
@@ -171,12 +177,14 @@ public class TcpReceivingChannelAdapterTests {
 		for (int i = 0; i < 1000; i++) {
 			assertTrue(results.remove("Test" + i));
 		}
+		scf.stop();
 	}

 	@Test
 	public void testNetShared() throws Exception {
 		final int port = SocketUtils.findAvailableServerSocket();
 		AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port);
+		noopPublisher(scf);
 		ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
 		scf.setSerializer(serializer);
 		scf.setDeserializer(serializer);
@@ -203,12 +211,14 @@ public class TcpReceivingChannelAdapterTests {
 		assertEquals("Test\r\n", new String(b));
 		readFully(socket.getInputStream(), b);
 		assertEquals("Test\r\n", new String(b));
+		scf.stop();
 	}

 	@Test
 	public void testNioShared() throws Exception {
 		final int port = SocketUtils.findAvailableServerSocket();
 		TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port);
+		noopPublisher(scf);
 		ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
 		scf.setSerializer(serializer);
 		scf.setDeserializer(serializer);
@@ -235,12 +245,14 @@ public class TcpReceivingChannelAdapterTests {
 		assertEquals("Test\r\n", new String(b));
 		readFully(socket.getInputStream(), b);
 		assertEquals("Test\r\n", new String(b));
+		scf.stop();
 	}

 	@Test
 	public void testNetSingleNoOutbound() throws Exception {
 		final int port = SocketUtils.findAvailableServerSocket();
 		AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port);
+		noopPublisher(scf);
 		ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
 		scf.setSerializer(serializer);
 		scf.setDeserializer(serializer);
@@ -265,12 +277,14 @@ public class TcpReceivingChannelAdapterTests {
 		results.add(new String((byte[]) message.getPayload()));
 		assertTrue(results.contains("Test1"));
 		assertTrue(results.contains("Test2"));
+		scf.stop();
 	}

 	@Test
 	public void testNioSingleNoOutbound() throws Exception {
 		final int port = SocketUtils.findAvailableServerSocket();
 		TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port);
+		noopPublisher(scf);
 		ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
 		scf.setSerializer(serializer);
 		scf.setDeserializer(serializer);
@@ -295,6 +309,7 @@ public class TcpReceivingChannelAdapterTests {
 		results.add(new String((byte[]) message.getPayload()));
 		assertTrue(results.contains("Test1"));
 		assertTrue(results.contains("Test2"));
+		scf.stop();
 	}

 	/**
@@ -311,6 +326,7 @@ public class TcpReceivingChannelAdapterTests {
 	public void testNetSingleShared() throws Exception {
 		final int port = SocketUtils.findAvailableServerSocket();
 		AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port);
+		noopPublisher(scf);
 		ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
 		scf.setSerializer(serializer);
 		scf.setDeserializer(serializer);
@@ -340,12 +356,14 @@ public class TcpReceivingChannelAdapterTests {
 		assertEquals("Test1\r\n", new String(b));
 		readFully(socket2.getInputStream(), b);
 		assertEquals("Test2\r\n", new String(b));
+		scf.stop();
 	}

 	@Test
 	public void testNioSingleShared() throws Exception {
 		final int port = SocketUtils.findAvailableServerSocket();
 		TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port);
+		noopPublisher(scf);
 		ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
 		scf.setSerializer(serializer);
 		scf.setDeserializer(serializer);
@@ -375,12 +393,14 @@ public class TcpReceivingChannelAdapterTests {
 		assertEquals("Test1\r\n", new String(b));
 		readFully(socket2.getInputStream(), b);
 		assertEquals("Test2\r\n", new String(b));
+		scf.stop();
 	}

 	@Test
 	public void testNioSingleSharedMany() throws Exception {
 		final int port = SocketUtils.findAvailableServerSocket();
 		TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port);
+		noopPublisher(scf);
 		ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
 		scf.setSerializer(serializer);
 		scf.setDeserializer(serializer);
@@ -413,48 +433,61 @@ public class TcpReceivingChannelAdapterTests {
 			readFully(sockets.remove(0).getInputStream(), b);
 			assertEquals("Test" + i + "\r\n", new String(b));
 		}
+		scf.stop();
 	}

 	@Test
 	public void testNetInterceptors() throws Exception {
 		final int port = SocketUtils.findAvailableServerSocket();
 		AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port);
+		noopPublisher(scf);
 		interceptorsGuts(port, scf);
+		scf.stop();
 	}

 	@Test
 	public void testNetSingleNoOutboundInterceptors() throws Exception {
 		final int port = SocketUtils.findAvailableServerSocket();
 		AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port);
+		noopPublisher(scf);
 		singleNoOutboundInterceptorsGuts(port, scf);
+		scf.stop();
 	}

 	@Test
 	public void testNetSingleSharedInterceptors() throws Exception {
 		final int port = SocketUtils.findAvailableServerSocket();
 		AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port);
+		noopPublisher(scf);
 		singleSharedInterceptorsGuts(port, scf);
+		scf.stop();
 	}

 	@Test
 	public void testNioInterceptors() throws Exception {
 		final int port = SocketUtils.findAvailableServerSocket();
 		AbstractServerConnectionFactory scf = new TcpNioServerConnectionFactory(port);
+		noopPublisher(scf);
 		interceptorsGuts(port, scf);
+		scf.stop();
 	}

 	@Test
 	public void testNioSingleNoOutboundInterceptors() throws Exception {
 		final int port = SocketUtils.findAvailableServerSocket();
 		AbstractServerConnectionFactory scf = new TcpNioServerConnectionFactory(port);
+		noopPublisher(scf);
 		singleNoOutboundInterceptorsGuts(port, scf);
+		scf.stop();
 	}

 	@Test
 	public void testNioSingleSharedInterceptors() throws Exception {
 		final int port = SocketUtils.findAvailableServerSocket();
 		AbstractServerConnectionFactory scf = new TcpNioServerConnectionFactory(port);
+		noopPublisher(scf);
 		singleSharedInterceptorsGuts(port, scf);
+		scf.stop();
 	}

 	private void interceptorsGuts(final int port, AbstractServerConnectionFactory scf) throws Exception {
@@ -464,9 +497,10 @@ public class TcpReceivingChannelAdapterTests {
 		TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
 		adapter.setConnectionFactory(scf);
 		TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain();
-		fc.setInterceptors(new TcpConnectionInterceptorFactory[]
-            		     {new HelloWorldInterceptorFactory(),
-               		      new HelloWorldInterceptorFactory()});
+		fc.setInterceptors(new TcpConnectionInterceptorFactory[] {
+				newInterceptorFactory(),
+				newInterceptorFactory()
+		});
 		scf.setInterceptorFactoryChain(fc);
 		scf.setSoTimeout(10000);
 		scf.start();
@@ -498,9 +532,10 @@ public class TcpReceivingChannelAdapterTests {
 		scf.setSingleUse(true);
 		scf.setSoTimeout(10000);
 		TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain();
-		fc.setInterceptors(new TcpConnectionInterceptorFactory[]
-            		     {new HelloWorldInterceptorFactory(),
-               		      new HelloWorldInterceptorFactory()});
+		fc.setInterceptors(new TcpConnectionInterceptorFactory[] {
+				newInterceptorFactory(),
+				newInterceptorFactory()
+		});
 		scf.setInterceptorFactoryChain(fc);
 		TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
 		adapter.setConnectionFactory(scf);
@@ -540,9 +575,10 @@ public class TcpReceivingChannelAdapterTests {
 		scf.setSingleUse(true);
 		scf.setSoTimeout(60000);
 		TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain();
-		fc.setInterceptors(new TcpConnectionInterceptorFactory[]
-            		     {new HelloWorldInterceptorFactory(),
-               		      new HelloWorldInterceptorFactory()});
+		fc.setInterceptors(new TcpConnectionInterceptorFactory[] {
+				newInterceptorFactory(),
+				newInterceptorFactory()
+		});
 		scf.setInterceptorFactoryChain(fc);
 		TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
 		handler.setConnectionFactory(scf);
@@ -583,6 +619,7 @@ public class TcpReceivingChannelAdapterTests {
 	public void testException() throws Exception {
 		final int port = SocketUtils.findAvailableServerSocket();
 		AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port);
+		noopPublisher(scf);
 		ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
 		scf.setSerializer(serializer);
 		scf.setDeserializer(serializer);
@@ -605,6 +642,7 @@ public class TcpReceivingChannelAdapterTests {
 		message = errorChannel.receive(10000);
 		assertNotNull(message);
 		assertEquals("Failed", ((Exception) message.getPayload()).getCause().getMessage());
+		scf.stop();
 	}

 	private class FailingService {
--- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java
+++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2012 the original author or authors.
+ * Copyright 2002-2013 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.
@@ -48,7 +48,8 @@ import org.junit.Test;
 import org.mockito.Mockito;
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
-import org.springframework.context.ApplicationContext;
+
+import org.springframework.context.support.AbstractApplicationContext;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
 import org.springframework.core.serializer.DefaultDeserializer;
 import org.springframework.core.serializer.DefaultSerializer;
@@ -60,7 +61,6 @@ import org.springframework.integration.core.PollableChannel;
 import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory;
 import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory;
 import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory;
-import org.springframework.integration.ip.tcp.connection.HelloWorldInterceptorFactory;
 import org.springframework.integration.ip.tcp.connection.TcpConnectionInterceptorFactory;
 import org.springframework.integration.ip.tcp.connection.TcpConnectionInterceptorFactoryChain;
 import org.springframework.integration.ip.tcp.connection.TcpNetClientConnectionFactory;
@@ -80,7 +80,7 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
 * @author Artem Bilan
 * @since 2.0
 */
-public class TcpSendingMessageHandlerTests {
+public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTests {

 	private static final Log logger = LogFactory.getLog(TcpSendingMessageHandlerTests.class);

@@ -97,6 +97,7 @@ public class TcpSendingMessageHandlerTests {
 		final CountDownLatch latch = new CountDownLatch(1);
 		final AtomicBoolean done = new AtomicBoolean();
 		Executors.newSingleThreadExecutor().execute(new Runnable() {
+			@Override
 			public void run() {
 				try {
 					ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
@@ -118,6 +119,7 @@ public class TcpSendingMessageHandlerTests {
 			}
 		});
 		AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
+		noopPublisher(ccf);
 		ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
 		ccf.setSerializer(serializer);
 		ccf.setDeserializer(serializer);
@@ -148,6 +150,7 @@ public class TcpSendingMessageHandlerTests {
 		final CountDownLatch latch = new CountDownLatch(1);
 		final AtomicBoolean done = new AtomicBoolean();
 		Executors.newSingleThreadExecutor().execute(new Runnable() {
+			@Override
 			public void run() {
 				try {
 					ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
@@ -169,6 +172,7 @@ public class TcpSendingMessageHandlerTests {
 			}
 		});
 		AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
+		noopPublisher(ccf);
 		ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
 		ccf.setSerializer(serializer);
 		ccf.setDeserializer(serializer);
@@ -201,6 +205,7 @@ public class TcpSendingMessageHandlerTests {
 		handler.stop();
 		handler.start();
 		handler.stop();
+		adapter.stop();
 	}

 	@Test
@@ -209,6 +214,7 @@ public class TcpSendingMessageHandlerTests {
 		final CountDownLatch latch = new CountDownLatch(1);
 		final AtomicBoolean done = new AtomicBoolean();
 		Executors.newSingleThreadExecutor().execute(new Runnable() {
+			@Override
 			public void run() {
 				try {
 					ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
@@ -229,6 +235,7 @@ public class TcpSendingMessageHandlerTests {
 			}
 		});
 		AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port);
+		noopPublisher(ccf);
 		ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
 		ccf.setSerializer(serializer);
 		ccf.setDeserializer(serializer);
@@ -262,6 +269,7 @@ public class TcpSendingMessageHandlerTests {
 		final CountDownLatch latch = new CountDownLatch(1);
 		final AtomicBoolean done = new AtomicBoolean();
 		Executors.newSingleThreadExecutor().execute(new Runnable() {
+			@Override
 			public void run() {
 				try {
 					ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
@@ -282,6 +290,7 @@ public class TcpSendingMessageHandlerTests {
 			}
 		});
 		AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
+		noopPublisher(ccf);
 		ByteArrayStxEtxSerializer serializer = new ByteArrayStxEtxSerializer();
 		ccf.setSerializer(serializer);
 		ccf.setDeserializer(serializer);
@@ -312,6 +321,7 @@ public class TcpSendingMessageHandlerTests {
 		final CountDownLatch latch = new CountDownLatch(1);
 		final AtomicBoolean done = new AtomicBoolean();
 		Executors.newSingleThreadExecutor().execute(new Runnable() {
+			@Override
 			public void run() {
 				try {
 					ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
@@ -332,6 +342,7 @@ public class TcpSendingMessageHandlerTests {
 			}
 		});
 		AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port);
+		noopPublisher(ccf);
 		ByteArrayStxEtxSerializer serializer = new ByteArrayStxEtxSerializer();
 		ccf.setSerializer(serializer);
 		ccf.setDeserializer(serializer);
@@ -365,6 +376,7 @@ public class TcpSendingMessageHandlerTests {
 		final CountDownLatch latch = new CountDownLatch(1);
 		final AtomicBoolean done = new AtomicBoolean();
 		Executors.newSingleThreadExecutor().execute(new Runnable() {
+			@Override
 			public void run() {
 				try {
 					ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
@@ -388,6 +400,7 @@ public class TcpSendingMessageHandlerTests {
 			}
 		});
 		AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
+		noopPublisher(ccf);
 		ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer();
 		ccf.setSerializer(serializer);
 		ccf.setDeserializer(serializer);
@@ -418,6 +431,7 @@ public class TcpSendingMessageHandlerTests {
 		final CountDownLatch latch = new CountDownLatch(1);
 		final AtomicBoolean done = new AtomicBoolean();
 		Executors.newSingleThreadExecutor().execute(new Runnable() {
+			@Override
 			public void run() {
 				try {
 					ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
@@ -441,6 +455,7 @@ public class TcpSendingMessageHandlerTests {
 			}
 		});
 		AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port);
+		noopPublisher(ccf);
 		ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer();
 		ccf.setSerializer(serializer);
 		ccf.setDeserializer(serializer);
@@ -474,6 +489,7 @@ public class TcpSendingMessageHandlerTests {
 		final CountDownLatch latch = new CountDownLatch(1);
 		final AtomicBoolean done = new AtomicBoolean();
 		Executors.newSingleThreadExecutor().execute(new Runnable() {
+			@Override
 			public void run() {
 				try {
 					ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
@@ -494,6 +510,7 @@ public class TcpSendingMessageHandlerTests {
 			}
 		});
 		AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
+		noopPublisher(ccf);
 		ccf.setSerializer(new DefaultSerializer());
 		ccf.setDeserializer(new DefaultDeserializer());
 		ccf.setSoTimeout(10000);
@@ -523,6 +540,7 @@ public class TcpSendingMessageHandlerTests {
 		final CountDownLatch latch = new CountDownLatch(1);
 		final AtomicBoolean done = new AtomicBoolean();
 		Executors.newSingleThreadExecutor().execute(new Runnable() {
+			@Override
 			public void run() {
 				try {
 					ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
@@ -543,6 +561,7 @@ public class TcpSendingMessageHandlerTests {
 			}
 		});
 		AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port);
+		noopPublisher(ccf);
 		ccf.setSerializer(new DefaultSerializer());
 		ccf.setDeserializer(new DefaultDeserializer());
 		ccf.setSoTimeout(10000);
@@ -576,6 +595,7 @@ public class TcpSendingMessageHandlerTests {
 		final Semaphore semaphore = new Semaphore(0);
 		final AtomicBoolean done = new AtomicBoolean();
 		Executors.newSingleThreadExecutor().execute(new Runnable() {
+			@Override
 			public void run() {
 				try {
 					ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
@@ -597,6 +617,7 @@ public class TcpSendingMessageHandlerTests {
 			}
 		});
 		AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
+		noopPublisher(ccf);
 		ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
 		ccf.setSerializer(serializer);
 		ccf.setDeserializer(serializer);
@@ -620,6 +641,7 @@ public class TcpSendingMessageHandlerTests {
 		final Semaphore semaphore = new Semaphore(0);
 		final AtomicBoolean done = new AtomicBoolean();
 		Executors.newSingleThreadExecutor().execute(new Runnable() {
+			@Override
 			public void run() {
 				try {
 					ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
@@ -641,6 +663,7 @@ public class TcpSendingMessageHandlerTests {
 			}
 		});
 		AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port);
+		noopPublisher(ccf);
 		ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
 		ccf.setSerializer(serializer);
 		ccf.setDeserializer(serializer);
@@ -664,6 +687,7 @@ public class TcpSendingMessageHandlerTests {
 		final Semaphore semaphore = new Semaphore(0);
 		final AtomicBoolean done = new AtomicBoolean();
 		Executors.newSingleThreadExecutor().execute(new Runnable() {
+			@Override
 			public void run() {
 				try {
 					ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
@@ -686,6 +710,7 @@ public class TcpSendingMessageHandlerTests {
 			}
 		});
 		AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
+		noopPublisher(ccf);
 		ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
 		ccf.setSerializer(serializer);
 		ccf.setDeserializer(serializer);
@@ -721,6 +746,7 @@ public class TcpSendingMessageHandlerTests {
 		final Semaphore semaphore = new Semaphore(0);
 		final AtomicBoolean done = new AtomicBoolean();
 		Executors.newSingleThreadExecutor().execute(new Runnable() {
+			@Override
 			public void run() {
 				try {
 					ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
@@ -743,6 +769,7 @@ public class TcpSendingMessageHandlerTests {
 			}
 		});
 		AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port);
+		noopPublisher(ccf);
 		ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
 		ccf.setSerializer(serializer);
 		ccf.setDeserializer(serializer);
@@ -779,6 +806,7 @@ public class TcpSendingMessageHandlerTests {
 		final AtomicBoolean done = new AtomicBoolean();
 		final List<Socket> serverSockets = new ArrayList<Socket>();
 		Executors.newSingleThreadExecutor().execute(new Runnable() {
+			@Override
 			public void run() {
 				try {
 					ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port, 100);
@@ -802,6 +830,7 @@ public class TcpSendingMessageHandlerTests {
 			}
 		});
 		AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port);
+		noopPublisher(ccf);
 		ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
 		ccf.setSerializer(serializer);
 		ccf.setDeserializer(serializer);
@@ -845,6 +874,7 @@ public class TcpSendingMessageHandlerTests {
 		final CountDownLatch latch = new CountDownLatch(1);
 		final AtomicBoolean done = new AtomicBoolean();
 		Executors.newSingleThreadExecutor().execute(new Runnable() {
+			@Override
 			public void run() {
 				try {
 					ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
@@ -879,13 +909,15 @@ public class TcpSendingMessageHandlerTests {
 			}
 		});
 		AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
+		noopPublisher(ccf);
 		ccf.setSerializer(new DefaultSerializer());
 		ccf.setDeserializer(new DefaultDeserializer());
 		ccf.setSoTimeout(10000);
 		TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain();
-		fc.setInterceptors(new TcpConnectionInterceptorFactory[]
-		     {new HelloWorldInterceptorFactory(),
-		      new HelloWorldInterceptorFactory()});
+		fc.setInterceptors(new TcpConnectionInterceptorFactory[] {
+				newInterceptorFactory(),
+				newInterceptorFactory()
+		});
 		ccf.setInterceptorFactoryChain(fc);
 		ccf.start();
 		TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
@@ -913,6 +945,7 @@ public class TcpSendingMessageHandlerTests {
 		final CountDownLatch latch = new CountDownLatch(1);
 		final AtomicBoolean done = new AtomicBoolean();
 		Executors.newSingleThreadExecutor().execute(new Runnable() {
+			@Override
 			public void run() {
 				try {
 					ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
@@ -942,11 +975,12 @@ public class TcpSendingMessageHandlerTests {
 			}
 		});
 		AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port);
+		noopPublisher(ccf);
 		ccf.setSerializer(new DefaultSerializer());
 		ccf.setDeserializer(new DefaultDeserializer());
 		ccf.setSoTimeout(10000);
 		TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain();
-		fc.setInterceptors(new TcpConnectionInterceptorFactory[] {new HelloWorldInterceptorFactory()});
+		fc.setInterceptors(new TcpConnectionInterceptorFactory[] {newInterceptorFactory()});
 		ccf.setInterceptorFactoryChain(fc);
 		ccf.start();
 		TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
@@ -979,6 +1013,7 @@ public class TcpSendingMessageHandlerTests {
 		final CountDownLatch latch = new CountDownLatch(1);
 		final AtomicBoolean done = new AtomicBoolean();
 		Executors.newSingleThreadExecutor().execute(new Runnable() {
+			@Override
 			public void run() {
 				try {
 					ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
@@ -1013,13 +1048,15 @@ public class TcpSendingMessageHandlerTests {
 			}
 		});
 		AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
+		noopPublisher(ccf);
 		ccf.setSerializer(new DefaultSerializer());
 		ccf.setDeserializer(new DefaultDeserializer());
 		ccf.setSoTimeout(10000);
 		TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain();
-		fc.setInterceptors(new TcpConnectionInterceptorFactory[]
-            		     {new HelloWorldInterceptorFactory(),
-               		      new HelloWorldInterceptorFactory()});
+		fc.setInterceptors(new TcpConnectionInterceptorFactory[] {
+				newInterceptorFactory(),
+				newInterceptorFactory()
+		});
 		ccf.setInterceptorFactoryChain(fc);
 		ccf.setSingleUse(true);
 		ccf.start();
@@ -1037,6 +1074,7 @@ public class TcpSendingMessageHandlerTests {
 		final CountDownLatch latch = new CountDownLatch(1);
 		final AtomicBoolean done = new AtomicBoolean();
 		Executors.newSingleThreadExecutor().execute(new Runnable() {
+			@Override
 			public void run() {
 				int i = 0;
 				try {
@@ -1070,13 +1108,15 @@ public class TcpSendingMessageHandlerTests {
 			}
 		});
 		AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port);
+		noopPublisher(ccf);
 		ccf.setSerializer(new DefaultSerializer());
 		ccf.setDeserializer(new DefaultDeserializer());
 		ccf.setSoTimeout(10000);
 		TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain();
-		fc.setInterceptors(new TcpConnectionInterceptorFactory[]
-            		     {new HelloWorldInterceptorFactory(),
-               		      new HelloWorldInterceptorFactory()});
+		fc.setInterceptors(new TcpConnectionInterceptorFactory[] {
+				newInterceptorFactory(),
+				newInterceptorFactory()
+		});
 		ccf.setInterceptorFactoryChain(fc);
 		ccf.setSingleUse(true);
 		ccf.start();
@@ -1090,7 +1130,7 @@ public class TcpSendingMessageHandlerTests {

 	@Test
 	public void testOutboundChannelAdapterWithinChain() throws Exception {
-		ApplicationContext ctx = new ClassPathXmlApplicationContext(
+		AbstractApplicationContext ctx = new ClassPathXmlApplicationContext(
 				"TcpOutboundChannelAdapterWithinChainTests-context.xml", this.getClass());
 		AbstractServerConnectionFactory scf = ctx.getBean(AbstractServerConnectionFactory.class);
 		TestingUtilities.waitListening(scf, null);
@@ -1101,6 +1141,7 @@ public class TcpSendingMessageHandlerTests {
 		Message<?> m = inbound.receive(1000);
 		assertNotNull(m);
 		assertEquals(testPayload, new String((byte[]) m.getPayload()));
+		ctx.destroy();
 	}

 	@Test
@@ -1109,6 +1150,7 @@ public class TcpSendingMessageHandlerTests {
 		AbstractConnectionFactory mockCcf = mock(AbstractClientConnectionFactory.class);
 		Mockito.doAnswer(new Answer<Object>() {

+			@Override
 			public Object answer(InvocationOnMock invocation) throws Throwable {
 				throw new SocketException("Failed to connect");
 			}
--- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/HelloWorldInterceptor.java
+++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/HelloWorldInterceptor.java
@@ -20,6 +20,8 @@ import java.util.concurrent.TimeUnit;

 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+
+import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.integration.Message;
 import org.springframework.integration.MessagingException;
 import org.springframework.integration.support.MessageBuilder;
@@ -50,12 +52,8 @@ public class HelloWorldInterceptor extends TcpConnectionInterceptorSupport {
 	public HelloWorldInterceptor() {
 	}

-	/**
-	 * @param hello
-	 * @param world
-	 */
-	public HelloWorldInterceptor(String hello, String world) {
-		super();
+	public HelloWorldInterceptor(String hello, String world, ApplicationEventPublisher applicationEventPublisher) {
+		super(applicationEventPublisher);
 		this.hello = hello;
 		this.world = world;
 	}
--- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/HelloWorldInterceptorFactory.java
+++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/HelloWorldInterceptorFactory.java
@@ -15,6 +15,9 @@
 */
 package org.springframework.integration.ip.tcp.connection;

+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.context.ApplicationEventPublisherAware;
+

 /**
 * @author Gary Russell
@@ -22,12 +25,14 @@ package org.springframework.integration.ip.tcp.connection;
 *
 */
 public class HelloWorldInterceptorFactory implements
-		TcpConnectionInterceptorFactory {
+		TcpConnectionInterceptorFactory, ApplicationEventPublisherAware {

 	private String hello = "Hello";

 	private String world = "world!";

+	private volatile ApplicationEventPublisher applicationEventPublisher;
+
 	public HelloWorldInterceptorFactory() {
 	}

@@ -40,9 +45,14 @@ public class HelloWorldInterceptorFactory implements
 		this.world = world;
 	}

+	@Override
+	public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
+		this.applicationEventPublisher = applicationEventPublisher;
+	}

+	@Override
 	public TcpConnectionInterceptorSupport getInterceptor() {
-		return new HelloWorldInterceptor(hello, world);
+		return new HelloWorldInterceptor(this.hello, this.world, this.applicationEventPublisher);
 	}


--- a/spring-integration-ip/src/test/resources/log4j.properties
+++ b/spring-integration-ip/src/test/resources/log4j.properties
@@ -2,7 +2,7 @@ log4j.rootCategory=WARN, stdout

 log4j.appender.stdout=org.apache.log4j.ConsoleAppender
 log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
-log4j.appender.stdout.layout.ConversionPattern=%c{1}: %m%n
+log4j.appender.stdout.layout.ConversionPattern=%c{1} [%t] : %m%n

 log4j.category.org.springframework.integration=WARN
 log4j.category.org.springframework.integration.ip=WARN