From 5bb6e550aacf0ad749488241056cc895c89a057b Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 19 Nov 2013 10:34:08 -0500 Subject: [PATCH] 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. --- .../TcpConnectionInterceptorSupport.java | 18 +++++ .../tcp/connection/TcpConnectionSupport.java | 16 ++++- .../tcp/AbstractTcpChannelAdapterTests.java | 48 +++++++++++++ .../tcp/TcpReceivingChannelAdapterTests.java | 60 +++++++++++++--- .../ip/tcp/TcpSendingMessageHandlerTests.java | 72 +++++++++++++++---- .../tcp/connection/HelloWorldInterceptor.java | 10 ++- .../HelloWorldInterceptorFactory.java | 14 +++- .../src/test/resources/log4j.properties | 2 +- 8 files changed, 204 insertions(+), 36 deletions(-) create mode 100644 spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/AbstractTcpChannelAdapterTests.java diff --git 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 index e46ef42ec5..f542cb384c 100644 --- 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); diff --git 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 index 05f2a675be..3c42c0f2c1 100644 --- 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; } diff --git 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 new file mode 100644 index 0000000000..a4610afd09 --- /dev/null +++ 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); + } + + +} diff --git 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 index 578cdd72e2..7721ad499b 100644 --- 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 { diff --git 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 index ae853eec5f..f9b791aeef 100644 --- 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 serverSockets = new ArrayList(); 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() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { throw new SocketException("Failed to connect"); } diff --git 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 index dbc251bf97..1c82d99ff8 100644 --- 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; } diff --git 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 index 23e8093a15..6fd83e3fd9 100644 --- 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); } diff --git a/spring-integration-ip/src/test/resources/log4j.properties b/spring-integration-ip/src/test/resources/log4j.properties index 8349c76351..9a0731f0e7 100644 --- 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