added close notification

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

6 files changed, 78 insertions, 16 deletions

diff --git a/qpid/java/common/src/main/java/org/apache/qpidity/ToyBroker.java b/qpid/java/common/src/main/java/org/apache/qpidity/ToyBroker.java
index 2658dfcd09..2bd97f3aff 100644
--- a/qpid/java/common/src/main/java/org/apache/qpidity/ToyBroker.java
+++ b/qpid/java/common/src/main/java/org/apache/qpidity/ToyBroker.java
@@ -276,6 +276,7 @@ class ToyBroker extends SessionDelegate
             {
                 t.printStackTrace();
             }
+            public void closed() {}
         };
 
         //hack
diff --git a/qpid/java/common/src/main/java/org/apache/qpidity/ToyClient.java b/qpid/java/common/src/main/java/org/apache/qpidity/ToyClient.java
index 690e729540..977704fc0c 100644
--- a/qpid/java/common/src/main/java/org/apache/qpidity/ToyClient.java
+++ b/qpid/java/common/src/main/java/org/apache/qpidity/ToyClient.java
@@ -73,6 +73,7 @@ class ToyClient extends SessionDelegate
                                                   {
                                                       t.printStackTrace();
                                                   }
+                                                  public void closed() {}
                                               });
         conn.send(new ConnectionEvent(0, new ProtocolHeader(1, 0, 10)));
 
diff --git a/qpid/java/common/src/main/java/org/apache/qpidity/transport/Connection.java b/qpid/java/common/src/main/java/org/apache/qpidity/transport/Connection.java
index fca7437dfd..62aba83e29 100644
--- a/qpid/java/common/src/main/java/org/apache/qpidity/transport/Connection.java
+++ b/qpid/java/common/src/main/java/org/apache/qpidity/transport/Connection.java
@@ -127,6 +127,7 @@ public class Connection
                 it.remove();
             }
         }
+        delegate.closed();
     }
 
     public void close()
diff --git a/qpid/java/common/src/main/java/org/apache/qpidity/transport/ConnectionDelegate.java b/qpid/java/common/src/main/java/org/apache/qpidity/transport/ConnectionDelegate.java
index b497a56f78..001ad7220c 100644
--- a/qpid/java/common/src/main/java/org/apache/qpidity/transport/ConnectionDelegate.java
+++ b/qpid/java/common/src/main/java/org/apache/qpidity/transport/ConnectionDelegate.java
@@ -69,6 +69,8 @@ public abstract class ConnectionDelegate extends MethodDelegate<Channel>
 
     public abstract void exception(Throwable t);
 
+    public abstract void closed();
+
     public void setCondition(Lock negotiationCompleteLock,Condition negotiationComplete)
     {
         _negotiationComplete = negotiationComplete;
diff --git a/qpid/java/common/src/main/java/org/apache/qpidity/transport/Echo.java b/qpid/java/common/src/main/java/org/apache/qpidity/transport/Echo.java
index a3ac24dc48..03d0d3e161 100644
--- a/qpid/java/common/src/main/java/org/apache/qpidity/transport/Echo.java
+++ b/qpid/java/common/src/main/java/org/apache/qpidity/transport/Echo.java
@@ -74,6 +74,7 @@ public class Echo extends SessionDelegate
             {
                 t.printStackTrace();
             }
+            public void closed() {}
         };
 
         //hack
diff --git a/qpid/java/common/src/test/java/org/apache/qpidity/transport/ConnectionTest.java b/qpid/java/common/src/test/java/org/apache/qpidity/transport/ConnectionTest.java
index 15c1b691a7..167b02a556 100644
--- a/qpid/java/common/src/test/java/org/apache/qpidity/transport/ConnectionTest.java
+++ b/qpid/java/common/src/test/java/org/apache/qpidity/transport/ConnectionTest.java
@@ -20,11 +20,15 @@
  */
 package org.apache.qpidity.transport;
 
+import org.apache.mina.util.AvailablePortFinder;
+
 import org.apache.qpidity.transport.network.mina.MinaHandler;
 import org.apache.qpidity.transport.util.Logger;
 
 import junit.framework.TestCase;
 
+import java.util.Random;
+
 /**
  * ConnectionTest
  */
@@ -34,9 +38,14 @@ public class ConnectionTest extends TestCase
 
     private static final Logger log = Logger.get(ConnectionTest.class);
 
-    private static final int PORT = 1234;
+    private int port;
+
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+
+        port = AvailablePortFinder.getNextAvailable(12000);
 
-    public void testWriteToClosed() throws Exception {
         ConnectionDelegate server = new ConnectionDelegate() {
             public void init(Channel ch, ProtocolHeader hdr) {
                 ch.getConnection().close();
@@ -48,24 +57,61 @@ public class ConnectionTest extends TestCase
             public void exception(Throwable t) {
                 log.error(t, "exception caught");
             }
+            public void closed() {}
         };
 
-        MinaHandler.accept("0.0.0.0", PORT, server);
-
-        Connection conn = MinaHandler.connect("0.0.0.0", PORT,
-                                              new ConnectionDelegate()
-                                              {
-                                                  public SessionDelegate getSessionDelegate()
-                                                  {
-                                                      return new SessionDelegate() {};
-                                                  }
-                                                  public void exception(Throwable t)
-                                                  {
-                                                      t.printStackTrace();
-                                                  }
-                                              });
+        MinaHandler.accept("0.0.0.0", port, server);
+    }
+
+    private class Condition
+    {
+        private boolean value = false;
+
+        public synchronized void set()
+        {
+            value = true;
+            notifyAll();
+        }
+
+        public synchronized boolean get(long timeout) throws InterruptedException
+        {
+            if (!value)
+            {
+                wait(timeout);
+            }
+
+            return value;
+        }
+    }
+
+    private Connection connect(final Condition closed)
+    {
+        Connection conn = MinaHandler.connect("0.0.0.0", port, new ConnectionDelegate()
+        {
+            public SessionDelegate getSessionDelegate()
+            {
+                return new SessionDelegate() {};
+            }
+            public void exception(Throwable t)
+            {
+                t.printStackTrace();
+            }
+            public void closed()
+            {
+                if (closed != null)
+                {
+                    closed.set();
+                }
+            }
+        });
 
         conn.send(new ConnectionEvent(0, new ProtocolHeader(1, 0, 10)));
+        return conn;
+    }
+
+    public void testWriteToClosed() throws Exception
+    {
+        Connection conn = connect(null);
         Channel ch = conn.getChannel(0);
         Session ssn = new Session();
         ssn.attach(ch);
@@ -81,4 +127,14 @@ public class ConnectionTest extends TestCase
         }
     }
 
+    public void testClosedNotification() throws Exception
+    {
+        Condition closed = new Condition();
+        Connection conn = connect(closed);
+        if (!closed.get(3000))
+        {
+            fail("never got notified of connection close");
+        }
+    }
+
 }