QPID-2855 : Broker Transport should not block on awaiting session close confirmation

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

3 files changed, 47 insertions, 16 deletions

diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java
index 42ccee3b3f..7393b17243 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java
@@ -336,6 +336,12 @@ public class ServerSession extends Session implements PrincipalHolder, SessionCo
 
     }
 
+    @Override
+    protected void awaitClose()
+    {
+        // Broker shouldn't block awaiting close - thus do override this method to do nothing
+    }
+
     public void acknowledge(final Subscription_0_10 sub, final QueueEntry entry)
     {
         _transaction.dequeue(entry.getQueue(), entry.getMessage(),
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSessionDelegate.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSessionDelegate.java
index 73eebec7bc..b4bb6eb0b4 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSessionDelegate.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSessionDelegate.java
@@ -101,11 +101,14 @@ public class ServerSessionDelegate extends SessionDelegate
     public void command(Session session, Method method)
     {
         SecurityManager.setThreadPrincipal(session.getConnection().getAuthorizationID());
-        
-        super.command(session, method);
-        if (method.isSync())
+
+        if(!session.isClosing())
         {
-            session.flushProcessed();
+            super.command(session, method);
+            if (method.isSync())
+            {
+                session.flushProcessed();
+            }
         }
     }
 
@@ -1189,6 +1192,12 @@ public class ServerSessionDelegate extends SessionDelegate
         ((ServerSession)session).onClose();
     }
 
+    @Override
+    public void detached(Session session)
+    {
+        closed(session);
+    }
+
     public Collection<Subscription_0_10> getSubscriptions(Session session)
     {
         return ((ServerSession)session).getSubscriptions();
diff --git a/qpid/java/common/src/main/java/org/apache/qpid/transport/Session.java b/qpid/java/common/src/main/java/org/apache/qpid/transport/Session.java
index 9b84ff422b..e989849477 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/transport/Session.java
+++ b/qpid/java/common/src/main/java/org/apache/qpid/transport/Session.java
@@ -543,7 +543,7 @@ public class Session extends SessionInvoker
                     }
                 }
 
-                if (state != OPEN && state != CLOSED)
+                if (state != OPEN && state != CLOSED && state != CLOSING)
                 {
                     Thread current = Thread.currentThread();
                     if (!current.equals(resumer))
@@ -568,6 +568,7 @@ public class Session extends SessionInvoker
                             ("timed out waiting for resume to finish");
                     }
                     break;
+                case CLOSING:
                 case CLOSED:
                     ExecutionException exc = getException();
                     if (exc != null)
@@ -906,18 +907,23 @@ public class Session extends SessionInvoker
             sessionRequestTimeout(0);
             sessionDetach(name.getBytes());
 
-            Waiter w = new Waiter(commands, timeout);
-            while (w.hasTime() && state != CLOSED)
-            {
-                w.await();
-            }
-
-            if (state != CLOSED)
-            {
-                throw new SessionException("close() timed out");
-            }
+            awaitClose();
  
-            connection.removeSession(this);
+
+        }
+    }
+
+    protected void awaitClose() 
+    {
+        Waiter w = new Waiter(commands, timeout);
+        while (w.hasTime() && state != CLOSED)
+        {
+            w.await();
+        }
+
+        if (state != CLOSED)
+        {
+            throw new SessionException("close() timed out");
         }
     }
 
@@ -960,6 +966,16 @@ public class Session extends SessionInvoker
                 delegate.detached(this);
             }
         }
+
+        if(state == CLOSED)
+        {
+            connection.removeSession(this);            
+        }
+    }
+
+    public boolean isClosing()
+    {
+        return state == CLOSED || state == CLOSING;
     }
 
     public String toString()