diff options
Diffstat (limited to 'java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkTransport.java')
| -rw-r--r-- | java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkTransport.java | 51 |
1 files changed, 31 insertions, 20 deletions
diff --git a/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkTransport.java b/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkTransport.java index 9b6f0a0b1b..c8027e143e 100644 --- a/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkTransport.java +++ b/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkTransport.java @@ -41,9 +41,8 @@ import org.apache.qpid.transport.ConnectionSettings; import org.apache.qpid.transport.NetworkTransportConfiguration; import org.apache.qpid.transport.Receiver; import org.apache.qpid.transport.TransportException; -import org.apache.qpid.transport.network.IncomingNetworkTransport; -import org.apache.qpid.transport.network.NetworkConnection; -import org.apache.qpid.transport.network.OutgoingNetworkTransport; +import org.apache.qpid.transport.network.*; + import org.slf4j.LoggerFactory; public class IoNetworkTransport implements OutgoingNetworkTransport, IncomingNetworkTransport @@ -56,7 +55,9 @@ public class IoNetworkTransport implements OutgoingNetworkTransport, IncomingNet private IoNetworkConnection _connection; private AcceptingThread _acceptor; - public NetworkConnection connect(ConnectionSettings settings, Receiver<ByteBuffer> delegate, SSLContext sslContext) + public NetworkConnection connect(ConnectionSettings settings, + Receiver<ByteBuffer> delegate, + TransportActivity transportActivity) { int sendBufferSize = settings.getWriteBufferSize(); int receiveBufferSize = settings.getReadBufferSize(); @@ -91,7 +92,9 @@ public class IoNetworkTransport implements OutgoingNetworkTransport, IncomingNet try { - _connection = new IoNetworkConnection(_socket, delegate, sendBufferSize, receiveBufferSize, TIMEOUT); + IdleTimeoutTicker ticker = new IdleTimeoutTicker(transportActivity, TIMEOUT); + _connection = new IoNetworkConnection(_socket, delegate, sendBufferSize, receiveBufferSize, TIMEOUT, ticker); + ticker.setConnection(_connection); _connection.start(); } catch(Exception e) @@ -128,9 +131,10 @@ public class IoNetworkTransport implements OutgoingNetworkTransport, IncomingNet return _connection; } - public void accept(NetworkTransportConfiguration config, ProtocolEngineFactory factory, SSLContext sslContext) + public void accept(NetworkTransportConfiguration config, + ProtocolEngineFactory factory, + SSLContext sslContext) { - try { _acceptor = new AcceptingThread(config, factory, sslContext); @@ -141,8 +145,6 @@ public class IoNetworkTransport implements OutgoingNetworkTransport, IncomingNet { throw new TransportException("Unable to start server socket", e); } - - } private class AcceptingThread extends Thread @@ -152,15 +154,16 @@ public class IoNetworkTransport implements OutgoingNetworkTransport, IncomingNet private ProtocolEngineFactory _factory; private SSLContext _sslContext; private ServerSocket _serverSocket; + private int _timeout; private AcceptingThread(NetworkTransportConfiguration config, ProtocolEngineFactory factory, - SSLContext sslContext) - throws IOException + SSLContext sslContext) throws IOException { _config = config; _factory = factory; _sslContext = sslContext; + _timeout = TIMEOUT; InetSocketAddress address = config.getAddress(); @@ -172,15 +175,19 @@ public class IoNetworkTransport implements OutgoingNetworkTransport, IncomingNet { SSLServerSocketFactory socketFactory = _sslContext.getServerSocketFactory(); _serverSocket = socketFactory.createServerSocket(); - ((SSLServerSocket)_serverSocket).setNeedClientAuth(config.needClientAuth()); - ((SSLServerSocket)_serverSocket).setWantClientAuth(config.wantClientAuth()); + if(config.needClientAuth()) + { + ((SSLServerSocket)_serverSocket).setNeedClientAuth(true); + } + else if(config.wantClientAuth()) + { + ((SSLServerSocket)_serverSocket).setWantClientAuth(true); + } } _serverSocket.setReuseAddress(true); _serverSocket.bind(address); - - } @@ -217,6 +224,7 @@ public class IoNetworkTransport implements OutgoingNetworkTransport, IncomingNet { socket = _serverSocket.accept(); socket.setTcpNoDelay(_config.getTcpNoDelay()); + socket.setSoTimeout(_timeout); final Integer sendBufferSize = _config.getSendBufferSize(); final Integer receiveBufferSize = _config.getReceiveBufferSize(); @@ -224,10 +232,12 @@ public class IoNetworkTransport implements OutgoingNetworkTransport, IncomingNet socket.setSendBufferSize(sendBufferSize); socket.setReceiveBufferSize(receiveBufferSize); - ProtocolEngine engine = _factory.newProtocolEngine(); - NetworkConnection connection = new IoNetworkConnection(socket, engine, sendBufferSize, receiveBufferSize, TIMEOUT); + final IdleTimeoutTicker ticker = new IdleTimeoutTicker(engine, TIMEOUT); + NetworkConnection connection = new IoNetworkConnection(socket, engine, sendBufferSize, receiveBufferSize, _timeout, + ticker); + ticker.setConnection(connection); if(_sslContext != null) { @@ -248,14 +258,14 @@ public class IoNetworkTransport implements OutgoingNetworkTransport, IncomingNet } catch(RuntimeException e) { - LOGGER.error("Error in Acceptor thread on port " + _config.getPort(), e); + LOGGER.error("Error in Acceptor thread on address " + _config.getAddress(), e); closeSocketIfNecessary(socket); } catch(IOException e) { if(!_closed) { - LOGGER.error("Error in Acceptor thread on port " + _config.getPort(), e); + LOGGER.error("Error in Acceptor thread on address " + _config.getAddress(), e); closeSocketIfNecessary(socket); try { @@ -275,7 +285,7 @@ public class IoNetworkTransport implements OutgoingNetworkTransport, IncomingNet { if(LOGGER.isDebugEnabled()) { - LOGGER.debug("Acceptor exiting, no new connections will be accepted on port " + _config.getPort()); + LOGGER.debug("Acceptor exiting, no new connections will be accepted on address " + _config.getAddress()); } } } @@ -294,6 +304,7 @@ public class IoNetworkTransport implements OutgoingNetworkTransport, IncomingNet } } } + } } |