From f4e775ec18ffba9082cc873711f5ce08a3fd06fc Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 18 Feb 2015 20:16:51 +0200 Subject: [PATCH] INT-3646: Add TCP Server Exception Events JIRA: https://jira.spring.io/browse/INT-3646 Publish `TcpConnectionServerExceptionEvent`s when unexpected exceptions occur on server sockets. INT-3646: Polishing Fix Reactor tests --- .../gateway/AsyncGatewayTests.java | 38 +++----- .../AbstractServerConnectionFactory.java | 30 +++--- .../TcpConnectionServerExceptionEvent.java | 53 ++++++++++ .../TcpNetServerConnectionFactory.java | 59 +++++------ .../TcpNioServerConnectionFactory.java | 55 ++++++----- .../tcp/connection/ConnectionEventTests.java | 97 ++++++++++++++++++- .../src/test/resources/log4j.properties | 3 +- src/reference/docbook/ip.xml | 5 + src/reference/docbook/whats-new.xml | 8 ++ 9 files changed, 251 insertions(+), 97 deletions(-) create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionServerExceptionEvent.java diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java index beca06f0d9..552df90c63 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java @@ -18,6 +18,7 @@ package org.springframework.integration.gateway; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotEquals; +import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; @@ -84,7 +85,7 @@ public class AsyncGatewayTests { Object result = f.get(1000, TimeUnit.MILLISECONDS); long elapsed = System.currentTimeMillis() - start; assertTrue(elapsed >= 200); - assertTrue(result instanceof Message); + assertNotNull(result); assertEquals("foobar", ((Message) result).getPayload()); } @@ -149,7 +150,6 @@ public class AsyncGatewayTests { long elapsed = System.currentTimeMillis() - start; assertTrue(elapsed >= 200); assertEquals("foobar", result.get().getPayload()); - Object thread = result.get().getHeaders().get("thread"); assertNotEquals(Thread.currentThread(), thread); } @@ -169,7 +169,6 @@ public class AsyncGatewayTests { CustomFuture f = service.returnCustomFuture("foo"); String result = f.get(1000, TimeUnit.MILLISECONDS); assertEquals("foobar", result); - assertEquals(Thread.currentThread(), f.thread); } @@ -184,14 +183,13 @@ public class AsyncGatewayTests { proxyFactory.setBeanName("testGateway"); proxyFactory.setBeanFactory(mock(BeanFactory.class)); - proxyFactory.setAsyncExecutor(null); // Not async - user flow returns Future + proxyFactory.setAsyncExecutor(null); // Not async - user flow returns Future proxyFactory.afterPropertiesSet(); TestEchoService service = (TestEchoService) proxyFactory.getObject(); CustomFuture f = (CustomFuture) service.returnCustomFutureWithTypeFuture("foo"); String result = f.get(1000, TimeUnit.MILLISECONDS); assertEquals("foobar", result); - assertEquals(Thread.currentThread(), f.thread); } @@ -204,6 +202,7 @@ public class AsyncGatewayTests { .setHeader("thread", Thread.currentThread()) .build(); } + }); } @@ -222,9 +221,8 @@ public class AsyncGatewayTests { long start = System.currentTimeMillis(); Object result = f.get(1000, TimeUnit.MILLISECONDS); long elapsed = System.currentTimeMillis() - start; - assertTrue(elapsed >= 200 - safety); - assertTrue(result instanceof String); + assertNotNull(result); assertEquals("foobar", result); } @@ -243,7 +241,6 @@ public class AsyncGatewayTests { long start = System.currentTimeMillis(); Object result = f.get(1000, TimeUnit.MILLISECONDS); long elapsed = System.currentTimeMillis() - start; - assertTrue(elapsed >= 200 - safety); assertTrue(result instanceof String); assertEquals("foobar", result); @@ -263,10 +260,7 @@ public class AsyncGatewayTests { proxyFactory.afterPropertiesSet(); TestEchoService service = (TestEchoService) proxyFactory.getObject(); Promise> promise = service.returnMessagePromise("foo"); - long start = System.currentTimeMillis(); Object result = promise.await(1, TimeUnit.SECONDS); - long elapsed = System.currentTimeMillis() - start; - assertTrue(elapsed <= 200); assertEquals("foobar", ((Message) result).getPayload()); } @@ -283,11 +277,7 @@ public class AsyncGatewayTests { proxyFactory.afterPropertiesSet(); TestEchoService service = (TestEchoService) proxyFactory.getObject(); Promise promise = service.returnStringPromise("foo"); - long start = System.currentTimeMillis(); Object result = promise.await(1, TimeUnit.SECONDS); - long elapsed = System.currentTimeMillis() - start; - - assertTrue(elapsed <= 200 - safety); assertEquals("foobar", result); } @@ -304,12 +294,8 @@ public class AsyncGatewayTests { proxyFactory.afterPropertiesSet(); TestEchoService service = (TestEchoService) proxyFactory.getObject(); Promise promise = service.returnSomethingPromise("foo"); - long start = System.currentTimeMillis(); Object result = promise.await(1, TimeUnit.SECONDS); - long elapsed = System.currentTimeMillis() - start; - - assertTrue(elapsed <= 200 - safety); - assertTrue(result instanceof String); + assertNotNull(result); assertEquals("foobar", result); } @@ -326,7 +312,6 @@ public class AsyncGatewayTests { proxyFactory.afterPropertiesSet(); TestEchoService service = (TestEchoService) proxyFactory.getObject(); Promise promise = service.returnStringPromise("foo"); - long start = System.currentTimeMillis(); final AtomicReference result = new AtomicReference(); final CountDownLatch latch = new CountDownLatch(1); @@ -340,9 +325,6 @@ public class AsyncGatewayTests { }); latch.await(1, TimeUnit.SECONDS); - long elapsed = System.currentTimeMillis() - start; - - assertTrue(elapsed <= 200 - safety); assertEquals("foobar", result.get()); } @@ -366,6 +348,7 @@ public class AsyncGatewayTests { private static void startResponder(final PollableChannel requestChannel) { new Thread(new Runnable() { + @Override public void run() { Message input = requestChannel.receive(); @@ -389,11 +372,12 @@ public class AsyncGatewayTests { } ((MessageChannel) input.getHeaders().getReplyChannel()).send(reply); } + }).start(); } - static interface TestEchoService { + interface TestEchoService { Future returnString(String s); @@ -403,10 +387,10 @@ public class AsyncGatewayTests { ListenableFuture> returnMessageListenable(String s); - @Gateway(headers=@GatewayHeader(name="method", expression="#gatewayMethod.name")) + @Gateway(headers = @GatewayHeader(name = "method", expression = "#gatewayMethod.name")) CustomFuture returnCustomFuture(String s); - @Gateway(headers=@GatewayHeader(name="method", expression="#gatewayMethod.name")) + @Gateway(headers = @GatewayHeader(name = "method", expression = "#gatewayMethod.name")) Future returnCustomFutureWithTypeFuture(String s); Promise returnStringPromise(String s); diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java index 1ca506025f..408a679c70 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2001-2014 the original author or authors. + * Copyright 2001-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. @@ -57,10 +57,10 @@ public abstract class AbstractServerConnectionFactory @Override public void start() { synchronized (this.lifecycleMonitor) { - if (!this.isActive()) { + if (!isActive()) { this.setActive(true); this.shuttingDown = false; - this.getTaskExecutor().execute(this); + getTaskExecutor().execute(this); } } super.start(); @@ -102,22 +102,22 @@ public abstract class AbstractServerConnectionFactory * @param socket The new socket. */ protected void initializeConnection(TcpConnectionSupport connection, Socket socket) { - TcpListener listener = this.getListener(); + TcpListener listener = getListener(); if (listener != null) { connection.registerListener(listener); } - connection.registerSender(this.getSender()); - connection.setMapper(this.getMapper()); - connection.setDeserializer(this.getDeserializer()); - connection.setSerializer(this.getSerializer()); - connection.setSingleUse(this.isSingleUse()); + connection.registerSender(getSender()); + connection.setMapper(getMapper()); + connection.setDeserializer(getDeserializer()); + connection.setSerializer(getSerializer()); + connection.setSingleUse(isSingleUse()); /* * If we are configured * for single use; need to enforce a timeout on the socket so we will close * if the client connects, but sends nothing. (Protect against DoS). * Behavior can be overridden by explicitly setting the timeout to zero. */ - if (this.isSingleUse() && this.getSoTimeout() < 0) { + if (isSingleUse() && getSoTimeout() < 0) { try { socket.setSoTimeout(DEFAULT_REPLY_TIMEOUT); } catch (SocketException e) { @@ -128,7 +128,7 @@ public abstract class AbstractServerConnectionFactory } protected void postProcessServerSocket(ServerSocket serverSocket) { - this.getTcpSocketSupport().postProcessServerSocket(serverSocket); + getTcpSocketSupport().postProcessServerSocket(serverSocket); } /** @@ -154,7 +154,7 @@ public abstract class AbstractServerConnectionFactory * @return The backlog. */ public int getBacklog() { - return backlog; + return this.backlog; } /** @@ -175,8 +175,12 @@ public abstract class AbstractServerConnectionFactory @Override public int afterShutdown() { - this.stop(); + stop(); return 0; } + protected void publishServerExceptionEvent(Exception e) { + getApplicationEventPublisher().publishEvent(new TcpConnectionServerExceptionEvent(this, e)); + } + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionServerExceptionEvent.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionServerExceptionEvent.java new file mode 100644 index 0000000000..fae213f2ca --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionServerExceptionEvent.java @@ -0,0 +1,53 @@ +/* + * Copyright 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. + * 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.connection; + +import org.springframework.integration.ip.event.IpIntegrationEvent; +import org.springframework.util.Assert; + +/** + * {@link IpIntegrationEvent} representing exceptions on a TCP server socket/channel. + * + * @author Gary Russell + * @since 4.0.7 + */ +@SuppressWarnings("serial") +public class TcpConnectionServerExceptionEvent extends IpIntegrationEvent { + + public TcpConnectionServerExceptionEvent(AbstractServerConnectionFactory connectionFactory, Throwable cause) { + super(connectionFactory, cause); + Assert.notNull(cause, "'cause' cannot be null"); + Assert.notNull(connectionFactory, "'connectionFactory' cannot be null"); + } + + /** + * The connection factory that experienced the exception; examine it to determine the port etc. + * @return the connection factory. + */ + public AbstractServerConnectionFactory getConnectionFactory() { + return (AbstractServerConnectionFactory) getSource(); + } + + @Override + public String toString() { + return super.toString() + + ", [factory=" + getConnectionFactory().getComponentType() + + ":" + getConnectionFactory().getComponentName() + + ", port=" + getConnectionFactory().getPort() + "]"; + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java index 3159dbaf9f..b35779acc6 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.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. @@ -30,6 +30,7 @@ import org.springframework.util.Assert; /** * Implements a server connection factory that produces {@link TcpNetConnection}s using * a {@link ServerSocket}. Must have a {@link TcpListener} registered. + * * @author Gary Russell * @since 2.0 * @@ -48,6 +49,11 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto super(port); } + @Override + public String getComponentType() { + return "tcp-net-server-connection-factory"; + } + /** * If no listener registers, exits. * Accepts incoming connections and creates TcpConnections for each new connection. @@ -58,22 +64,22 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto @Override public void run() { ServerSocket theServerSocket = null; - if (this.getListener() == null) { + if (getListener() == null) { logger.info("No listener bound to server connection factory; will not read; exiting..."); return; } try { - if (this.getLocalAddress() == null) { - theServerSocket = createServerSocket(this.getPort(), this.getBacklog(), null); + if (getLocalAddress() == null) { + theServerSocket = createServerSocket(getPort(), getBacklog(), null); } else { - InetAddress whichNic = InetAddress.getByName(this.getLocalAddress()); - theServerSocket = createServerSocket(this.getPort(), this.getBacklog(), whichNic); + InetAddress whichNic = InetAddress.getByName(getLocalAddress()); + theServerSocket = createServerSocket(getPort(), getBacklog(), whichNic); } - this.getTcpSocketSupport().postProcessServerSocket(theServerSocket); + getTcpSocketSupport().postProcessServerSocket(theServerSocket); this.serverSocket = theServerSocket; - this.setListening(true); - logger.info("Listening on port " + this.getPort()); + setListening(true); + logger.info("Listening on port " + getPort()); while (true) { final Socket socket; /* @@ -89,7 +95,7 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto } continue; } - if (this.isShuttingDown()) { + if (isShuttingDown()) { if (logger.isInfoEnabled()) { logger.info("New connection from " + socket.getInetAddress().getHostAddress() + " rejected; the server is in the process of shutting down."); @@ -101,12 +107,12 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto logger.debug("Accepted connection from " + socket.getInetAddress().getHostAddress()); } setSocketAttributes(socket); - TcpConnectionSupport connection = new TcpNetConnection(socket, true, this.isLookupHost(), - this.getApplicationEventPublisher(), this.getComponentName()); + TcpConnectionSupport connection = new TcpNetConnection(socket, true, isLookupHost(), + getApplicationEventPublisher(), getComponentName()); connection = wrapConnection(connection); - this.initializeConnection(connection, socket); - this.getTaskExecutor().execute(connection); - this.harvestClosedConnections(); + initializeConnection(connection, socket); + getTaskExecutor().execute(connection); + harvestClosedConnections(); connection.publishConnectionOpenEvent(); } } @@ -115,20 +121,21 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto // don't log an error if we had a good socket once and now it's closed if (e instanceof SocketException && theServerSocket != null) { logger.warn("Server Socket closed"); - } else if (this.isActive()) { - logger.error("Error on ServerSocket", e); + } + else if (isActive()) { + logger.error("Error on ServerSocket; port = " + getPort(), e); + publishServerExceptionEvent(e); } } finally { - this.setListening(false); - this.setActive(false); + setListening(false); + setActive(false); } } /** * Create a new {@link ServerSocket}. This default implementation uses the default * {@link ServerSocketFactory}. Override to use some other mechanism - * * @param port The port. * @param backlog The server socket backlog. * @param whichNic An InetAddress if binding to a specific network interface. Set to @@ -139,11 +146,10 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto protected ServerSocket createServerSocket(int port, int backlog, InetAddress whichNic) throws IOException { ServerSocketFactory serverSocketFactory = this.tcpSocketFactorySupport.getServerSocketFactory(); if (whichNic == null) { - return serverSocketFactory.createServerSocket(port, - Math.abs(backlog)); - } else { - return serverSocketFactory.createServerSocket(port, - Math.abs(backlog), whichNic); + return serverSocketFactory.createServerSocket(port, Math.abs(backlog)); + } + else { + return serverSocketFactory.createServerSocket(port, Math.abs(backlog), whichNic); } } @@ -171,8 +177,7 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto return tcpSocketFactorySupport; } - public void setTcpSocketFactorySupport( - TcpSocketFactorySupport tcpSocketFactorySupport) { + public void setTcpSocketFactorySupport(TcpSocketFactorySupport tcpSocketFactorySupport) { Assert.notNull(tcpSocketFactorySupport, "TcpSocketFactorySupport may not be null"); this.tcpSocketFactorySupport = tcpSocketFactorySupport; } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java index 1d53e73d60..a4069c96a1 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.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. @@ -61,6 +61,11 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto super(port); } + @Override + public String getComponentType() { + return "tcp-nio-server-connection-factory"; + } + /** * If no listener registers, exits. * Accepts incoming connections and creates TcpConnections for each new connection. @@ -70,39 +75,41 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto */ @Override public void run() { - if (this.getListener() == null) { + if (getListener() == null) { logger.info("No listener bound to server connection factory; will not read; exiting..."); return; } try { this.serverChannel = ServerSocketChannel.open(); - int port = this.getPort(); - this.getTcpSocketSupport().postProcessServerSocket(this.serverChannel.socket()); + int port = getPort(); + getTcpSocketSupport().postProcessServerSocket(this.serverChannel.socket()); if (logger.isInfoEnabled()) { logger.info("Listening on port " + port); } this.serverChannel.configureBlocking(false); - if (this.getLocalAddress() == null) { - this.serverChannel.socket().bind(new InetSocketAddress(port), - Math.abs(this.getBacklog())); + if (getLocalAddress() == null) { + this.serverChannel.socket().bind(new InetSocketAddress(port), Math.abs(getBacklog())); } else { - InetAddress whichNic = InetAddress.getByName(this.getLocalAddress()); - this.serverChannel.socket().bind(new InetSocketAddress(whichNic, port), - Math.abs(this.getBacklog())); + InetAddress whichNic = InetAddress.getByName(getLocalAddress()); + this.serverChannel.socket().bind(new InetSocketAddress(whichNic, port), Math.abs(getBacklog())); } final Selector selector = Selector.open(); this.serverChannel.register(selector, SelectionKey.OP_ACCEPT); - this.setListening(true); + setListening(true); this.selector = selector; doSelect(this.serverChannel, selector); } catch (IOException e) { - this.stop(); + if (isActive()) { + logger.error("Error on ServerChannel; port = " + getPort(), e); + publishServerExceptionEvent(e); + } + stop(); } finally { - this.setListening(false); + setListening(false); } } @@ -119,8 +126,8 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto * @throws IOException */ private void doSelect(ServerSocketChannel server, final Selector selector) throws IOException { - while (this.isActive()) { - int soTimeout = this.getSoTimeout(); + while (isActive()) { + int soTimeout = getSoTimeout(); int selectionCount = 0; try { long timeout = soTimeout < 0 ? 0 : soTimeout; @@ -131,7 +138,7 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto logger.trace("Delayed reads:" + getDelayedReads().size() + " timeout " + timeout); } selectionCount = selector.select(timeout); - this.processNioSelections(selectionCount, selector, server, this.channelMap); + processNioSelections(selectionCount, selector, server, this.channelMap); } catch (CancelledKeyException cke) { if (logger.isDebugEnabled()) { @@ -139,8 +146,9 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto } } catch (ClosedSelectorException cse) { - if (this.isActive()) { + if (isActive()) { logger.error("Selector closed", cse); + publishServerExceptionEvent(cse); break; } } @@ -157,7 +165,7 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto protected void doAccept(final Selector selector, ServerSocketChannel server, long now) throws IOException { logger.debug("New accept"); SocketChannel channel = server.accept(); - if (this.isShuttingDown()) { + if (isShuttingDown()) { if (logger.isInfoEnabled()) { logger.info("New connection from " + channel.socket().getInetAddress().getHostAddress() + " rejected; the server is in the process of shutting down."); @@ -173,7 +181,7 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto if (connection == null) { return; } - connection.setTaskExecutor(this.getTaskExecutor()); + connection.setTaskExecutor(getTaskExecutor()); connection.setLastRead(now); this.channelMap.put(channel, connection); channel.register(selector, SelectionKey.OP_READ, connection); @@ -188,12 +196,11 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto private TcpNioConnection createTcpNioConnection(SocketChannel socketChannel) { try { - TcpNioConnection connection = this.tcpNioConnectionSupport - .createNewConnection(socketChannel, true, - this.isLookupHost(), this.getApplicationEventPublisher(), this.getComponentName()); + TcpNioConnection connection = this.tcpNioConnectionSupport.createNewConnection(socketChannel, true, + isLookupHost(), getApplicationEventPublisher(), getComponentName()); connection.setUsingDirectBuffers(this.usingDirectBuffers); TcpConnectionSupport wrappedConnection = wrapConnection(connection); - this.initializeConnection(wrappedConnection, socketChannel.socket()); + initializeConnection(wrappedConnection, socketChannel.socket()); return connection; } catch (Exception e) { @@ -204,7 +211,7 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto @Override public void stop() { - this.setActive(false); + setActive(false); if (this.selector != null) { try { this.selector.close(); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionEventTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionEventTests.java index 02d9a197b5..7858a63255 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionEventTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionEventTests.java @@ -16,25 +16,49 @@ package org.springframework.integration.ip.tcp.connection; +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.endsWith; +import static org.hamcrest.Matchers.instanceOf; +import static org.hamcrest.Matchers.startsWith; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import static org.mockito.Matchers.any; +import static org.mockito.Matchers.anyString; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; import java.io.OutputStream; +import java.net.BindException; +import java.net.ServerSocket; import java.net.Socket; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import javax.net.ServerSocketFactory; + +import org.apache.commons.logging.Log; import org.junit.Test; +import org.mockito.ArgumentCaptor; import org.mockito.Mockito; +import org.mockito.internal.stubbing.answers.DoesNothing; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.context.ApplicationEvent; import org.springframework.context.ApplicationEventPublisher; import org.springframework.core.serializer.Serializer; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.messaging.Message; import org.springframework.messaging.support.GenericMessage; +import org.springframework.util.SocketUtils; /** * @author Gary Russell @@ -44,7 +68,7 @@ import org.springframework.messaging.support.GenericMessage; public class ConnectionEventTests { @Test - public void test() throws Exception { + public void testConnectionEvents() throws Exception { Socket socket = mock(Socket.class); final List theEvent = new ArrayList(); TcpNetConnection conn = new TcpNetConnection(socket, false, false, new ApplicationEventPublisher() { @@ -56,9 +80,9 @@ public class ConnectionEventTests { @Override public void publishEvent(Object event) { - + } - + }, "foo"); /* * Open is not published by the connection itself; the factory publishes it after initialization. @@ -85,7 +109,72 @@ public class ConnectionEventTests { assertSame(toBeThrown, event.getCause()); assertTrue(theEvent.size() > 1); assertNotNull(theEvent.get(1)); - assertTrue(theEvent.get(1).toString().endsWith("[factory=foo, connectionId=" + conn.getConnectionId() + "] **CLOSED**")); + assertTrue(theEvent.get(1).toString() + .endsWith("[factory=foo, connectionId=" + conn.getConnectionId() + "] **CLOSED**")); + } + + @Test + public void testNetServerExceptionEvent() throws Exception { + int port = SocketUtils.findAvailableTcpPort(); + AbstractServerConnectionFactory factory = new TcpNetServerConnectionFactory(port); + testServerExceptionGuts(port, factory); + } + + @Test + public void testNioServerExceptionEvent() throws Exception { + int port = SocketUtils.findAvailableTcpPort(); + AbstractServerConnectionFactory factory = new TcpNioServerConnectionFactory(port); + testServerExceptionGuts(port, factory); + } + + private void testServerExceptionGuts(int port, AbstractServerConnectionFactory factory) throws Exception { + ServerSocket ss = ServerSocketFactory.getDefault().createServerSocket(port); + final AtomicReference theEvent = + new AtomicReference(); + final CountDownLatch latch = new CountDownLatch(1); + factory.setApplicationEventPublisher(new ApplicationEventPublisher() { + + @Override + public void publishEvent(ApplicationEvent event) { + theEvent.set((TcpConnectionServerExceptionEvent) event); + latch.countDown(); + } + + @Override + public void publishEvent(Object event) { + + } + + }); + factory.setBeanName("sf"); + factory.registerListener(new TcpListener() { + + @Override + public boolean onMessage(Message message) { + return false; + } + }); + Log logger = spy(TestUtils.getPropertyValue(factory, "logger", Log.class)); + doAnswer(new DoesNothing()).when(logger).error(anyString(), any(Throwable.class)); + new DirectFieldAccessor(factory).setPropertyValue("logger", logger); + + factory.start(); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + String actual = theEvent.toString(); + assertThat(actual, containsString("cause=java.net.BindException")); + assertThat(actual, containsString("[factory=" + + (factory instanceof TcpNetServerConnectionFactory + ? "tcp-net-server-connection-factory" + : "tcp-nio-server-connection-factory") + + ":sf, port=" + port + "]")); + + ArgumentCaptor reasonCaptor = ArgumentCaptor.forClass(String.class); + ArgumentCaptor throwableCaptor = ArgumentCaptor.forClass(Throwable.class); + verify(logger).error(reasonCaptor.capture(), throwableCaptor.capture()); + assertThat(reasonCaptor.getValue(), startsWith("Error on Server")); + assertThat(reasonCaptor.getValue(), endsWith("; port = " + port)); + assertThat(throwableCaptor.getValue(), instanceOf(BindException.class)); + ss.close(); } } diff --git a/spring-integration-ip/src/test/resources/log4j.properties b/spring-integration-ip/src/test/resources/log4j.properties index c38d3b65db..31e0d7ed7e 100644 --- a/spring-integration-ip/src/test/resources/log4j.properties +++ b/spring-integration-ip/src/test/resources/log4j.properties @@ -2,7 +2,6 @@ log4j.rootCategory=WARN, stdout log4j.appender.stdout=org.apache.log4j.ConsoleAppender log4j.appender.stdout.layout=org.apache.log4j.PatternLayout -log4j.appender.stdout.layout.ConversionPattern=%d %c{1} [%t] : %m%n +log4j.appender.stdout.layout.ConversionPattern=%d %5p %c{1} [%t] : %m%n log4j.category.org.springframework.integration=WARN -log4j.category.org.springframework.integration.ip=WARN diff --git a/src/reference/docbook/ip.xml b/src/reference/docbook/ip.xml index 249bf4365d..513ee48411 100644 --- a/src/reference/docbook/ip.xml +++ b/src/reference/docbook/ip.xml @@ -508,6 +508,11 @@ point the exception occurred. Applications can use a normal ApplicationListener, or see , to capture these events, allowing analysis of the problem. + + Starting with versions 4.0.7, 4.1.3 TcpConnectionServerExceptionEvents + are published whenever an unexpected exception occurs on a server socket (such as a BindException + when the server socket is in use). These events have a reference to the connection factory and the cause. +
TCP Adapters diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index b57d3440f5..a0f395d235 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -39,5 +39,13 @@ org.springframework.integration.scattergather.
+
+ Server Socket Exceptions + + TcpConnectionServerExceptionEvents are now published whenever an + unexpected exception occurs on a TCP server socket (also added to 4.1.3, 4.0.7). + See for more information. + +