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

INT-1797 Remove ERROR Log on NIO Socket Close; Refactor common NIO Client/Server factory code to AbstractConnectionFactory

Browse Source
This commit is contained in:
Gary Russell
2011-02-20 15:05:56 -05:00
parent fe1975f467
commit ecff2ce21e
7 changed files with 172 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -16,14 +16,21 @@
package org.springframework.integration.ip.tcp.connection;
import java.io.IOException;
import java.net.Socket;
import java.net.SocketException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.context.SmartLifecycle;
import org.springframework.core.serializer.Deserializer;
import org.springframework.core.serializer.Serializer;
@@ -346,6 +353,92 @@ public abstract class AbstractConnectionFactory
return connection;
}
/**
*
* Times out any expired connections then, if selectionCount > 0, processes the selected keys.
*
* @param selectionCount
* @param selector
* @param connections
* @throws IOException
*/
protected void processNioSelections(int selectionCount, final Selector selector, ServerSocketChannel server,
Map<SocketChannel, TcpNioConnection> connections) throws IOException {
long now = 0;
if (this.soTimeout > 0) {
Iterator<SocketChannel> it = connections.keySet().iterator();
now = System.currentTimeMillis();
while (it.hasNext()) {
SocketChannel channel = it.next();
if (!channel.isOpen()) {
logger.debug("Removing closed channel");
it.remove();
} else {
TcpNioConnection connection = connections.get(channel);
if (now - connection.getLastRead() > this.soTimeout) {
logger.warn("Timing out TcpNioConnection " +
this.port + " : " +
connection.getConnectionId());
connection.timeout();
}
}
}
}
if (logger.isTraceEnabled())
logger.trace("Host" + this.host + " port " + this.port + " SelectionCount: " + selectionCount);
if (selectionCount > 0) {
Set<SelectionKey> keys = selector.selectedKeys();
Iterator<SelectionKey> iterator = keys.iterator();
while (iterator.hasNext()) {
final SelectionKey key = iterator.next();
iterator.remove();
if (!key.isValid()) {
logger.debug("Selection key no longer valid");
}
else if (key.isReadable()) {
key.interestOps(key.interestOps() - key.readyOps());
final TcpNioConnection connection;
connection = (TcpNioConnection) key.attachment();
connection.setLastRead(System.currentTimeMillis());
this.taskExecutor.execute(new Runnable() {
public void run() {
try {
connection.readPacket();
} catch (Exception e) {
if (connection.isOpen()) {
logger.error("Exception on read " +
connection.getConnectionId() + " " +
e.getMessage());
connection.close();
} else {
logger.debug("Connection closed");
}
}
if (key.channel().isOpen()) {
key.interestOps(SelectionKey.OP_READ);
selector.wakeup();
}
}});
}
else if (key.isAcceptable()) {
doAccept(selector, server, now);
}
else {
logger.error("Unexpected key: " + key);
}
}
}
}
/**
* @param selector
* @param now
* @throws IOException
*/
protected void doAccept(final Selector selector, ServerSocketChannel server, long now) throws IOException {
throw new UnsupportedOperationException("Nio server factory must override this method");
}
public int getPhase() {
return 0;
}

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

@@ -21,9 +21,7 @@ import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
@@ -120,70 +118,12 @@ public class TcpNioClientConnectionFactory extends
try {
this.selector = Selector.open();
while (this.active) {
int selectionCount = selector.select(this.soTimeout);
SocketChannel newChannel;
int selectionCount = selector.select(this.soTimeout);
while ((newChannel = newChannels.poll()) != null) {
newChannel.register(this.selector, SelectionKey.OP_READ, connections.get(newChannel));
}
if (logger.isTraceEnabled())
logger.trace("Connection " + host + ":" + port + " SelectionCount: " + selectionCount);
long now = 0;
if (this.soTimeout > 0) {
Iterator<SocketChannel> it = connections.keySet().iterator();
now = System.currentTimeMillis();
while (it.hasNext()) {
SocketChannel channel = it.next();
if (!channel.isOpen()) {
logger.debug("Removing closed channel");
it.remove();
} else {
TcpNioConnection connection = this.connections.get(channel);
if (now - connection.getLastRead() > this.soTimeout) {
logger.warn("Timing out TcpNioConnection " + connection.getConnectionId());
connection.timeout();
}
}
}
}
if (selectionCount > 0) {
Set<SelectionKey> keys = selector.selectedKeys();
Iterator<SelectionKey> iterator = keys.iterator();
while (iterator.hasNext()) {
final SelectionKey key = iterator.next();
iterator.remove();
if (!key.isValid()) {
logger.debug("Selection key no longer valid");
}
else if (key.isReadable()) {
key.interestOps(key.interestOps() - key.readyOps());
final TcpNioConnection connection;
connection = (TcpNioConnection) key.attachment();
connection.setLastRead(System.currentTimeMillis());
this.taskExecutor.execute(new Runnable() {
public void run() {
try {
connection.readPacket();
} catch (Exception e) {
if (connection.isOpen()) {
logger.error("Exception on read " +
connection.getConnectionId() + " " +
e.getMessage());
connection.close();
} else {
logger.debug("Connection closed");
}
}
if (key.channel().isOpen()) {
key.interestOps(SelectionKey.OP_READ);
selector.wakeup();
}
}});
}
else {
logger.error("Unexpected key: " + key);
}
}
}
this.processNioSelections(selectionCount, selector, null, this.connections);
}
} catch (Exception e) {
logger.error("Exception in reader thread", e);

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

@@ -22,6 +22,7 @@ import java.io.PipedInputStream;
import java.io.PipedOutputStream;
import java.net.SocketTimeoutException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
@@ -277,7 +278,6 @@ public class TcpNioConnection extends AbstractTcpConnection {
int len = socketChannel.read(rawBuffer);
if (len < 0) {
this.closeConnection();
throw new IOException("Channel closed");
}
rawBuffer.flip();
if (logger.isDebugEnabled()) {
@@ -305,8 +305,11 @@ public class TcpNioConnection extends AbstractTcpConnection {
* Invoked by the factory when there is data to be read.
*/
public void readPacket() {
logger.debug("Reading...");
try {
doRead();
} catch (ClosedChannelException cce) {
this.closeConnection();
} catch (Exception e) {
logger.error("Exception on Read " +
this.getConnectionId() + " " +

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

@@ -27,9 +27,7 @@ import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
/**
/**
@@ -112,84 +110,36 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
throws IOException, ClosedChannelException, SocketException {
while (this.active) {
int selectionCount = selector.select(this.soTimeout);
if (logger.isTraceEnabled())
logger.trace("Port " + port + " SelectionCount: " + selectionCount);
long now = 0;
if (this.soTimeout > 0) {
Iterator<SocketChannel> it = connections.keySet().iterator();
now = System.currentTimeMillis();
while (it.hasNext()) {
SocketChannel channel = it.next();
if (!channel.isOpen()) {
logger.debug("Removing closed channel");
it.remove();
} else {
TcpNioConnection connection = this.connections.get(channel);
if (now - connection.getLastRead() > this.soTimeout) {
logger.warn("Timing out TcpNioConnection " +
this.port + " : " +
connection.getConnectionId());
connection.timeout();
}
}
}
}
if (selectionCount > 0) {
Set<SelectionKey> keys = selector.selectedKeys();
Iterator<SelectionKey> iterator = keys.iterator();
SocketChannel channel = null;
while (iterator.hasNext()) {
final SelectionKey key = iterator.next();
iterator.remove();
if (!key.isValid()) {
logger.debug("Selection key no longer valid");
}
else if (key.isAcceptable()) {
logger.debug("New accept");
channel = server.accept();
channel.configureBlocking(false);
Socket socket = channel.socket();
setSocketAttributes(socket);
TcpNioConnection connection = createTcpNioConnection(channel);
if (connection == null) {
continue;
}
connection.setTaskExecutor(this.taskExecutor);
connection.setLastRead(now);
connections.put(channel, connection);
channel.register(selector, SelectionKey.OP_READ, connection);
}
else if (key.isReadable()) {
key.interestOps(key.interestOps() - key.readyOps());
final TcpNioConnection connection;
connection = (TcpNioConnection) key.attachment();
connection.setLastRead(System.currentTimeMillis());
this.taskExecutor.execute(new Runnable() {
public void run() {
try {
connection.readPacket();
} catch (Exception e) {
if (connection.isOpen()) {
logger.error("Exception on read " + e.getMessage());
connection.close();
} else {
logger.debug("Connection closed");
}
}
if (key.channel().isOpen()) {
key.interestOps(SelectionKey.OP_READ);
selector.wakeup();
}
}});
}
else {
logger.error("Unexpected key: " + key);
}
}
}
this.processNioSelections(selectionCount, selector, server, this.connections);
}
}
/**
* @param selector
* @param connections
* @param server
* @param now
* @throws IOException
* @throws SocketException
* @throws ClosedChannelException
*/
@Override
protected void doAccept(final Selector selector, ServerSocketChannel server, long now) throws IOException {
logger.debug("New accept");
SocketChannel channel = server.accept();
channel.configureBlocking(false);
Socket socket = channel.socket();
setSocketAttributes(socket);
TcpNioConnection connection = createTcpNioConnection(channel);
if (connection == null) {
return;
}
connection.setTaskExecutor(this.taskExecutor);
connection.setLastRead(now);
connections.put(channel, connection);
channel.register(selector, SelectionKey.OP_READ, connection);
}
private TcpNioConnection createTcpNioConnection(SocketChannel socketChannel) {
try {
TcpNioConnection connection = new TcpNioConnection(socketChannel, true);

4
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/AbstractByteArraySerializer.java
View File

@@ -59,8 +59,8 @@ public abstract class AbstractByteArraySerializer implements
protected void checkClosure(int bite) throws IOException {
if (bite < 0) {
logger.debug("Socket closed");
throw new IOException("Socket closed");
logger.debug("Socket closed during message assembly");
throw new IOException("Socket closed during message assembly");
}
}

2
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests-context.xml
View File

@@ -13,6 +13,8 @@
<int-ip:tcp-connection-factory id="server"
type="server"
using-nio="true"
single-use="true"
port="#{tcpIpUtils.findAvailableServerSocket(10000)}"
/>

40
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java
View File

@@ -357,6 +357,46 @@ public class TcpNioConnectionReadTests {
scf.close();
}
/**
* Tests socket closure when no data received.
*
* @throws Exception
*/
@Test
public void testCloseCleanupPartialData() throws Exception {
int port = SocketTestUtils.findAvailableServerSocket();
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
serializer.setMaxMessageSize(1024);
final List<Message<?>> responses = new ArrayList<Message<?>>();
final Semaphore semaphore = new Semaphore(0);
final List<TcpConnection> added = new ArrayList<TcpConnection>();
final List<TcpConnection> removed = new ArrayList<TcpConnection>();
AbstractServerConnectionFactory scf = getConnectionFactory(port, serializer,new TcpListener() {
public boolean onMessage(Message<?> message) {
responses.add(message);
semaphore.release();
return false;
}
}, new TcpSender() {
public void addNewConnection(TcpConnection connection) {
added.add(connection);
semaphore.release();
}
public void removeDeadConnection(TcpConnection connection) {
removed.add(connection);
semaphore.release();
}
});
Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
socket.getOutputStream().write("partial".getBytes());
socket.close();
whileOpen(semaphore, added);
assertEquals(1, added.size());
assertTrue(semaphore.tryAcquire(10000, TimeUnit.MILLISECONDS));
assertTrue(removed.size() > 0);
scf.close();
}
/**
* Tests socket closure when mid-message
*