Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-1854 Shutdown task executor (if internal), close all sockets on close()

Browse Source
This commit is contained in:
Gary Russell
2011-04-26 21:10:46 -04:00
parent 54e7a7b994
commit 8b0b8a76ae
13 changed files with 140 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java
View File

@@ -58,9 +58,6 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection
connection.registerListener(this.listener);
}
if (this.listener != null || this.singleUse) {
if (this.taskExecutor == null) {
this.taskExecutor = Executors.newFixedThreadPool(this.poolSize);
}
if (this.soTimeout <= 0) {
try {
socket.setSoTimeout(DEFAULT_REPLY_TIMEOUT);

108
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java
View File

@@ -25,10 +25,15 @@ import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -74,8 +79,10 @@ public abstract class AbstractConnectionFactory
private int soTrafficClass = -1; // don't set by default
protected Executor taskExecutor;
private Executor taskExecutor;
private boolean privateExecutor;
protected Deserializer<?> deserializer = new ByteArrayCrLfSerializer();
protected Serializer<?> serializer = new ByteArrayCrLfSerializer();
@@ -91,6 +98,8 @@ public abstract class AbstractConnectionFactory
protected TcpConnectionInterceptorFactoryChain interceptorFactoryChain;
private boolean lookupHost = true;
private List<TcpConnection> connections = new LinkedList<TcpConnection>();
/**
* Sets socket attributes on the socket.
@@ -329,15 +338,22 @@ public abstract class AbstractConnectionFactory
public abstract void close();
/**
* Creates a taskExecutor (if one was not provided) and starts
* the listening process on one of its threads.
* Starts the listening process.
*/
public void start() {
this.active = true;
this.getTaskExecutor().execute(this);
}
/**
* Creates a taskExecutor (if one was not provided).
*/
protected Executor getTaskExecutor() {
if (this.taskExecutor == null) {
this.privateExecutor = true;
this.taskExecutor = Executors.newFixedThreadPool(this.poolSize);
}
this.active = true;
this.taskExecutor.execute(this);
return this.taskExecutor;
}
/**
@@ -346,30 +362,58 @@ public abstract class AbstractConnectionFactory
public void stop() {
this.active = false;
this.close();
synchronized (this.connections) {
Iterator<TcpConnection> iterator = this.connections.iterator();
while (iterator.hasNext()) {
TcpConnection connection = iterator.next();
connection.close();
iterator.remove();
}
}
if (this.privateExecutor) {
ExecutorService executorService = (ExecutorService) this.taskExecutor;
executorService.shutdown();
try {
if (!executorService.awaitTermination(10, TimeUnit.SECONDS)) {
logger.debug("Forcing executor shutdown");
executorService.shutdownNow();
if (!executorService.awaitTermination(10, TimeUnit.SECONDS)) {
logger.debug("Executor failed to shutdown");
}
}
} catch (InterruptedException e) {
executorService.shutdownNow();
Thread.currentThread().interrupt();
}
}
}
protected TcpConnection wrapConnection(TcpConnection connection) throws Exception {
if (this.interceptorFactoryChain == null) {
return connection;
}
TcpConnectionInterceptorFactory[] interceptorFactories =
this.interceptorFactoryChain.getInterceptorFactories();
if (interceptorFactories == null) {
return connection;
}
for (TcpConnectionInterceptorFactory factory : interceptorFactories) {
TcpConnectionInterceptor wrapper = factory.getInterceptor();
wrapper.setTheConnection(connection);
// if no ultimate listener or sender, register each wrapper in turn
if (this.listener == null) {
connection.registerListener(wrapper);
try {
if (this.interceptorFactoryChain == null) {
return connection;
}
if (this.sender == null) {
connection.registerSender(wrapper);
TcpConnectionInterceptorFactory[] interceptorFactories =
this.interceptorFactoryChain.getInterceptorFactories();
if (interceptorFactories == null) {
return connection;
}
connection = wrapper;
for (TcpConnectionInterceptorFactory factory : interceptorFactories) {
TcpConnectionInterceptor wrapper = factory.getInterceptor();
wrapper.setTheConnection(connection);
// if no ultimate listener or sender, register each wrapper in turn
if (this.listener == null) {
connection.registerListener(wrapper);
}
if (this.sender == null) {
connection.registerSender(wrapper);
}
connection = wrapper;
}
return connection;
} finally {
this.addConnection(connection);
}
return connection;
}
/**
@@ -403,6 +447,7 @@ public abstract class AbstractConnectionFactory
}
}
}
this.harvestClosedConnections();
if (logger.isTraceEnabled()) {
if (host == null) {
logger.trace("Port " + this.port + " SelectionCount: " + selectionCount);
@@ -488,4 +533,21 @@ public abstract class AbstractConnectionFactory
callback.run();
}
protected void addConnection(TcpConnection connection) {
synchronized (this.connections) {
this.connections.add(connection);
}
}
protected void harvestClosedConnections() {
synchronized (this.connections) {
Iterator<TcpConnection> iterator = this.connections.iterator();
while (iterator.hasNext()) {
TcpConnection connection = iterator.next();
if (!connection.isOpen()) {
iterator.remove();
}
}
}
}
}

3
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetClientConnectionFactory.java
View File

@@ -53,10 +53,11 @@ public class TcpNetClientConnectionFactory extends
TcpConnection connection = new TcpNetConnection(socket, false, this.isLookupHost());
connection = wrapConnection(connection);
initializeConnection(connection, socket);
this.taskExecutor.execute(connection);
this.getTaskExecutor().execute(connection);
if (!this.singleUse) {
this.theConnection = connection;
}
this.harvestClosedConnections();
return connection;
}

3
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java
View File

@@ -75,7 +75,8 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
TcpConnection connection = new TcpNetConnection(socket, true, this.isLookupHost());
connection = wrapConnection(connection);
this.initializeConnection(connection, socket);
this.taskExecutor.execute(connection);
this.getTaskExecutor().execute(connection);
this.harvestClosedConnections();
}
} catch (Exception e) {
this.listening = false;

10
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java
View File

@@ -24,7 +24,6 @@ import java.nio.channels.SocketChannel;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
@@ -80,11 +79,7 @@ public class TcpNioClientConnectionFactory extends
setSocketAttributes(socketChannel.socket());
TcpNioConnection connection = new TcpNioConnection(socketChannel, false, this.isLookupHost());
connection.setUsingDirectBuffers(this.usingDirectBuffers);
if (this.taskExecutor == null) {
connection.setTaskExecutor(Executors.newSingleThreadExecutor());
} else {
connection.setTaskExecutor(this.taskExecutor);
}
connection.setTaskExecutor(this.getTaskExecutor());
TcpConnection wrappedConnection = wrapConnection(connection);
initializeConnection(wrappedConnection, socketChannel.socket());
socketChannel.configureBlocking(false);
@@ -111,6 +106,9 @@ public class TcpNioClientConnectionFactory extends
}
public void close() {
if (this.selector != null) {
this.selector.wakeup();
}
}
public void run() {

27
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java
View File

@@ -162,8 +162,17 @@ public class TcpNioConnection extends AbstractTcpConnection {
} else {
this.executionControl.decrementAndGet();
}
} catch (IOException e) {
logger.error("Unexpected exception, exiting...", e);
} catch (Exception e) {
if (logger.isTraceEnabled()) {
logger.error("Read exception " +
this.getConnectionId(), e);
} else {
logger.error("Read exception " +
this.getConnectionId() + " " +
e.getClass().getSimpleName() +
":" + e.getCause() + ":" + e.getMessage());
}
this.closeConnection();
return;
}
} finally {
@@ -172,7 +181,7 @@ public class TcpNioConnection extends AbstractTcpConnection {
// timing was such that we were the last assembler and
// a new one wasn't run
try {
if (dataAvailable()) {
if (this.isOpen() && dataAvailable()) {
checkForAssembler();
}
} catch (IOException e) {
@@ -191,7 +200,7 @@ public class TcpNioConnection extends AbstractTcpConnection {
* @return The Message or null if no data is available.
* @throws IOException
*/
private synchronized Message<?> convert() throws IOException {
private synchronized Message<?> convert() throws Exception {
if (!dataAvailable()) {
return null;
}
@@ -206,15 +215,7 @@ public class TcpNioConnection extends AbstractTcpConnection {
}
} else {
if (!(e instanceof SoftEndOfStreamException)) {
if (logger.isTraceEnabled()) {
logger.error("Read exception " +
this.getConnectionId(), e);
} else {
logger.error("Read exception " +
this.getConnectionId() + " " +
e.getClass().getSimpleName() +
":" + e.getCause() + ":" + e.getMessage());
}
throw e;
}
}
return null;

9
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java
View File

@@ -44,6 +44,8 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
protected boolean usingDirectBuffers;
protected Map<SocketChannel, TcpNioConnection> connections = new HashMap<SocketChannel, TcpNioConnection>();
private Selector selector;
/**
* Listens for incoming connections on the port.
@@ -80,6 +82,7 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
final Selector selector = Selector.open();
this.serverChannel.register(selector, SelectionKey.OP_ACCEPT);
this.listening = true;
this.selector = selector;
doSelect(this.serverChannel, selector);
} catch (IOException e) {
@@ -116,7 +119,6 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
/**
* @param selector
* @param connections
* @param server
* @param now
* @throws IOException
@@ -134,7 +136,7 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
if (connection == null) {
return;
}
connection.setTaskExecutor(this.taskExecutor);
connection.setTaskExecutor(this.getTaskExecutor());
connection.setLastRead(now);
connections.put(channel, connection);
channel.register(selector, SelectionKey.OP_READ, connection);
@@ -158,6 +160,9 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
}
public void close() {
if (this.selector != null) {
this.selector.wakeup();
}
if (this.serverChannel == null) {
return;
}

17
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java
View File

@@ -26,7 +26,9 @@ import java.util.Properties;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.Message;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.history.MessageHistory;
@@ -60,6 +62,20 @@ public class ConnectionToConnectionTests {
@Autowired
private QueueChannel serverSideChannel;
// Test jvm shutdown
public static void main(String[] args) {
ConfigurableApplicationContext ctx = new ClassPathXmlApplicationContext(
ConnectionToConnectionTests.class.getPackage().getName()
.replaceAll("\\.", "/")
+ "/common-context.xml");
ctx.close();
ctx = new ClassPathXmlApplicationContext(
ConnectionToConnectionTests.class.getPackage().getName()
.replaceAll("\\.", "/")
+ "/ConnectionToConnectionTests-context.xml");
ctx.close();
}
@Test
public void testConnect() throws Exception {
@@ -115,4 +131,5 @@ public class ConnectionToConnectionTests {
assertFalse(connection.getConnectionId().contains("localhost"));
connection.close();
}
}

2
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java
View File

@@ -231,7 +231,7 @@ public class TcpInboundGatewayTests {
private class FailingService {
@SuppressWarnings("unused")
public String serviceMethod(byte[] bytes) {
throw new RuntimeException("Failed");
throw new RuntimeException("Planned Failure For Tests");
}
}

2
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java
View File

@@ -157,6 +157,7 @@ public class TcpOutboundGatewayTests {
for (int i = 0; i < 10; i++) {
assertTrue(replies.remove("Reply" + i));
}
done.set(true);
}
@Test
@@ -236,6 +237,7 @@ public class TcpOutboundGatewayTests {
for (int i = 0; i < 1; i++) {
assertTrue(replies.remove("Reply" + i));
}
done.set(true);
}

4
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java
View File

@@ -937,11 +937,11 @@ public class TcpSendingMessageHandlerTests {
final AtomicBoolean done = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(new Runnable() {
public void run() {
int i = 0;
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
latch.countDown();
Socket socket = server.accept();
int i = 0;
while (true) {
ObjectInputStream ois = new ObjectInputStream(socket.getInputStream());
Object in = null;
@@ -962,7 +962,7 @@ public class TcpSendingMessageHandlerTests {
oos.writeObject("Reply" + (++i));
}
} catch (Exception e) {
if (!done.get()) {
if (i == 0) {
e.printStackTrace();
}
}

6
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/SOLingerTests.java
View File

@@ -22,7 +22,7 @@ import static org.junit.Assert.fail;
import java.io.IOException;
import java.io.InputStream;
import java.net.Socket;
import java.net.SocketTimeoutException;
import java.net.SocketException;
import javax.net.SocketFactory;
@@ -139,7 +139,7 @@ public class SOLingerTests {
}
try {
Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
socket.setSoTimeout(200);
socket.setSoTimeout(10000);
String test = "Test\r\n";
socket.getOutputStream().write(test.getBytes());
byte[] buff = new byte[test.length() + 5];
@@ -149,7 +149,7 @@ public class SOLingerTests {
n = socket.getInputStream().read();
fail("Expected IOException");
} catch (IOException ioe) {
assertTrue(ioe instanceof SocketTimeoutException);
assertTrue(ioe instanceof SocketException);
}
} catch (Exception e) {
e.printStackTrace();

2
spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketTestUtils.java
View File

@@ -131,7 +131,7 @@ public class SocketTestUtils {
private static void writeByte(OutputStream os, int b, boolean noDelay) throws Exception {
os.write(b);
logger.debug("Wrote 0x" + Integer.toHexString(b));
logger.trace("Wrote 0x" + Integer.toHexString(b));
if (noDelay) {
return;
}