QPID-4925 : [Java Broker] Timeout connections which do not complete connection handshake

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@1493240 13f79535-47bb-0310-9956-ffa450edef68

4 files changed, 32 insertions, 9 deletions

diff --git a/java/broker/src/main/java/org/apache/qpid/server/logging/messages/Connection_logmessages.properties b/java/broker/src/main/java/org/apache/qpid/server/logging/messages/Connection_logmessages.properties
index 8559862a45..a99bcc7352 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/logging/messages/Connection_logmessages.properties
+++ b/java/broker/src/main/java/org/apache/qpid/server/logging/messages/Connection_logmessages.properties
@@ -22,4 +22,5 @@
 # 1 - Protocol Version
 # 2 - Client Version
 OPEN = CON-1001 : Open[ : Client ID : {0}][ : Protocol Version : {1}][ : Client Version : {2}]
-CLOSE = CON-1002 : Close
\ No newline at end of file
+CLOSE = CON-1002 : Close
+IDLE_CLOSE = CON-1003 : Closed due to inactivity
diff --git a/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java b/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java
index 3274cffc08..bbf90fad86 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java
@@ -790,6 +790,11 @@ public class AMQProtocolEngine implements ServerProtocolEngine, AMQProtocolSessi
             _network.setMaxWriteIdle(delay);
             _network.setMaxReadIdle(BrokerProperties.HEARTBEAT_TIMEOUT_FACTOR * delay);
         }
+        else
+        {
+            _network.setMaxWriteIdle(0);
+            _network.setMaxReadIdle(0);
+        }
     }
 
     /**
diff --git a/java/broker/src/main/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngine.java b/java/broker/src/main/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngine.java
index c0764272a4..3834d22180 100755
--- a/java/broker/src/main/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngine.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngine.java
@@ -32,6 +32,8 @@ import javax.net.ssl.SSLEngine;
 import javax.net.ssl.SSLPeerUnverifiedException;
 import org.apache.log4j.Logger;
 import org.apache.qpid.protocol.ServerProtocolEngine;
+import org.apache.qpid.server.logging.actors.CurrentActor;
+import org.apache.qpid.server.logging.messages.ConnectionMessages;
 import org.apache.qpid.server.model.Broker;
 import org.apache.qpid.server.model.Port;
 import org.apache.qpid.server.model.Transport;
@@ -452,6 +454,7 @@ public class MultiVersionProtocolEngine implements ServerProtocolEngine
     private class SelfDelegateProtocolEngine implements ServerProtocolEngine
     {
         private final ByteBuffer _header = ByteBuffer.allocate(MINIMUM_REQUIRED_HEADER_BYTES);
+        private long _lastReadTime;
 
         public SocketAddress getRemoteAddress()
         {
@@ -475,6 +478,8 @@ public class MultiVersionProtocolEngine implements ServerProtocolEngine
 
         public void received(ByteBuffer msg)
         {
+
+            _lastReadTime = System.currentTimeMillis();
             ByteBuffer msgheader = msg.duplicate();
             if(_header.remaining() > msgheader.limit())
             {
@@ -623,7 +628,8 @@ public class MultiVersionProtocolEngine implements ServerProtocolEngine
 
         public void readerIdle()
         {
-
+            CurrentActor.get().message(ConnectionMessages.IDLE_CLOSE());
+            _network.close();
         }
 
         public void setNetworkConnection(NetworkConnection network, Sender<ByteBuffer> sender)
@@ -634,7 +640,7 @@ public class MultiVersionProtocolEngine implements ServerProtocolEngine
         @Override
         public long getLastReadTime()
         {
-            return 0;
+            return _lastReadTime;
         }
 
         @Override
@@ -650,6 +656,7 @@ public class MultiVersionProtocolEngine implements ServerProtocolEngine
         private final SSLEngine _engine;
         private final SSLReceiver _sslReceiver;
         private final SSLBufferingSender _sslSender;
+        private long _lastReadTime;
 
         private SslDelegateProtocolEngine()
         {
@@ -678,6 +685,7 @@ public class MultiVersionProtocolEngine implements ServerProtocolEngine
         @Override
         public void received(ByteBuffer msg)
         {
+            _lastReadTime = System.currentTimeMillis();
             _sslReceiver.received(msg);
             _sslSender.send();
             _sslSender.flush();
@@ -746,7 +754,7 @@ public class MultiVersionProtocolEngine implements ServerProtocolEngine
         @Override
         public long getLastReadTime()
         {
-            return _decryptEngine.getLastReadTime();
+            return _lastReadTime;
         }
 
         @Override
diff --git a/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java b/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java
index 9e25182b10..4170f36771 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java
@@ -236,15 +236,24 @@ public class ServerConnectionDelegate extends ServerDelegate
             return;
         }
 
+        final NetworkConnection networkConnection = sconn.getNetworkConnection();
+
         if(ok.hasHeartbeat())
         {
-            final int heartbeat = ok.getHeartbeat();
-            if(heartbeat > 0)
+            int heartbeat = ok.getHeartbeat();
+            if(heartbeat < 0)
             {
-                final NetworkConnection networkConnection = sconn.getNetworkConnection();
-                networkConnection.setMaxReadIdle(2 * heartbeat);
-                networkConnection.setMaxWriteIdle(heartbeat);
+                heartbeat = 0;
             }
+
+            networkConnection.setMaxReadIdle(2 * heartbeat);
+            networkConnection.setMaxWriteIdle(heartbeat);
+
+        }
+        else
+        {
+            networkConnection.setMaxReadIdle(0);
+            networkConnection.setMaxWriteIdle(0);
         }
 
         setConnectionTuneOkChannelMax(sconn, okChannelMax);