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

Conflicts:
	spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java
	spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionEventTests.java
	src/reference/docbook/whats-new.xml
This commit is contained in:
Gary Russell
2015-02-18 20:16:51 +02:00
committed by Artem Bilan
parent 6854cb8227
commit c45a050e2b
7 changed files with 225 additions and 68 deletions

View File

@@ -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));
}
}

View File

@@ -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() + "]";
}
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -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;
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -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();

View File

@@ -15,25 +15,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
@@ -43,7 +67,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<TcpConnectionEvent> theEvent = new ArrayList<TcpConnectionEvent>();
TcpNetConnection conn = new TcpNetConnection(socket, false, false, new ApplicationEventPublisher() {
@@ -77,7 +101,67 @@ 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<TcpConnectionServerExceptionEvent> theEvent =
new AtomicReference<TcpConnectionServerExceptionEvent>();
final CountDownLatch latch = new CountDownLatch(1);
factory.setApplicationEventPublisher(new ApplicationEventPublisher() {
@Override
public void publishEvent(ApplicationEvent event) {
theEvent.set((TcpConnectionServerExceptionEvent) event);
latch.countDown();
}
});
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<String> reasonCaptor = ArgumentCaptor.forClass(String.class);
ArgumentCaptor<Throwable> 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();
}
}

View File

@@ -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

View File

@@ -508,6 +508,11 @@
point the exception occurred. Applications can use a normal <interfacename>ApplicationListener</interfacename>,
or see <xref linkend="applicationevent-inbound"/>, to capture these events, allowing analysis of the problem.
</para>
<para>
Starting with <emphasis>versions 4.0.7, 4.1.3</emphasis> <classname>TcpConnectionServerExceptionEvent</classname>s
are published whenever an unexpected exception occurs on a server socket (such as a <classname>BindException</classname>
when the server socket is in use). These events have a reference to the connection factory and the cause.
</para>
</section>
<section id="tcp-adapters">
<title>TCP Adapters</title>