improved close handling, made session.sync() return on session close

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

3 files changed, 31 insertions, 1 deletions

diff --git a/java/common/src/main/java/org/apache/qpidity/transport/Channel.java b/java/common/src/main/java/org/apache/qpidity/transport/Channel.java
index a74ff99eb9..f05ec82d65 100644
--- a/java/common/src/main/java/org/apache/qpidity/transport/Channel.java
+++ b/java/common/src/main/java/org/apache/qpidity/transport/Channel.java
@@ -112,6 +112,10 @@ public class Channel extends Invoker
     public void closed()
     {
         System.out.println("channel closed: " + this);
+        if (session != null)
+        {
+            session.closed();
+        }
     }
 
     public void close()
diff --git a/java/common/src/main/java/org/apache/qpidity/transport/Connection.java b/java/common/src/main/java/org/apache/qpidity/transport/Connection.java
index 2c68b7c38b..d29917afd5 100644
--- a/java/common/src/main/java/org/apache/qpidity/transport/Connection.java
+++ b/java/common/src/main/java/org/apache/qpidity/transport/Connection.java
@@ -21,6 +21,7 @@
 package org.apache.qpidity.transport;
 
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.Map;
 
 import java.nio.ByteBuffer;
@@ -94,6 +95,15 @@ public class Connection
     public void closed()
     {
         System.out.println("connection closed: " + this);
+        synchronized (channels)
+        {
+            for (Iterator<Channel> it = channels.values().iterator();
+                 it.hasNext(); )
+            {
+                it.next().closed();
+                it.remove();
+            }
+        }
     }
 
     public void close()
diff --git a/java/common/src/main/java/org/apache/qpidity/transport/Session.java b/java/common/src/main/java/org/apache/qpidity/transport/Session.java
index 59e8daae31..3305d04489 100644
--- a/java/common/src/main/java/org/apache/qpidity/transport/Session.java
+++ b/java/common/src/main/java/org/apache/qpidity/transport/Session.java
@@ -52,6 +52,8 @@ public class Session extends Invoker
     private Map<Long,Method> commands = new HashMap<Long,Method>();
     private long mark = 0;
 
+    private boolean closed = false;
+
 
     public Map<Long,Method> getOutstandingCommands()
     {
@@ -232,7 +234,7 @@ public class Session extends Invoker
                 executionSync();
             }
 
-            while (!commands.isEmpty())
+            while (!closed && !commands.isEmpty())
             {
                 try {
                     System.out.println("\n============sync() waiting for commmands to be completed ==============\n");
@@ -244,6 +246,11 @@ public class Session extends Invoker
                     throw new RuntimeException(e);
                 }
             }
+
+            if (!commands.isEmpty())
+            {
+                throw new RuntimeException("session closed");
+            }
         }
     }
 
@@ -334,4 +341,13 @@ public class Session extends Invoker
         channel.close();
     }
 
+    public void closed()
+    {
+        synchronized (commands)
+        {
+            closed = true;
+            commands.notifyAll();
+        }
+    }
+
 }