Candidate fix for QPID-1389. Make sure we don't send commands unless the session is fully opened.

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

4 files changed, 33 insertions, 6 deletions

diff --git a/java/common/src/main/java/org/apache/qpid/transport/ConnectionDelegate.java b/java/common/src/main/java/org/apache/qpid/transport/ConnectionDelegate.java
index e1b4705cc8..a1e3dee024 100644
--- a/java/common/src/main/java/org/apache/qpid/transport/ConnectionDelegate.java
+++ b/java/common/src/main/java/org/apache/qpid/transport/ConnectionDelegate.java
@@ -77,11 +77,6 @@ public abstract class ConnectionDelegate
         conn.getSender().close();
     }
 
-    @Override public void sessionAttached(Connection conn, SessionAttached atc)
-    {
-        
-    }
-
     @Override public void sessionDetach(Connection conn, SessionDetach dtc)
     {
         Session ssn = conn.getSession(dtc.getChannel());
diff --git a/java/common/src/main/java/org/apache/qpid/transport/ServerDelegate.java b/java/common/src/main/java/org/apache/qpid/transport/ServerDelegate.java
index e4b8ade285..529871fbf6 100644
--- a/java/common/src/main/java/org/apache/qpid/transport/ServerDelegate.java
+++ b/java/common/src/main/java/org/apache/qpid/transport/ServerDelegate.java
@@ -148,6 +148,7 @@ public class ServerDelegate extends ConnectionDelegate
         Session ssn = getSession(conn, atc);
         conn.map(ssn, atc.getChannel());
         ssn.sessionAttached(atc.getName());
+        ssn.setState(Session.State.OPEN);
     }
 
 }
diff --git a/java/common/src/main/java/org/apache/qpid/transport/Session.java b/java/common/src/main/java/org/apache/qpid/transport/Session.java
index e96aaf1b99..bab4bb35ee 100644
--- a/java/common/src/main/java/org/apache/qpid/transport/Session.java
+++ b/java/common/src/main/java/org/apache/qpid/transport/Session.java
@@ -157,6 +157,15 @@ public class Session extends SessionInvoker
         }
     }
 
+    void setState(State state)
+    {
+        synchronized (commands)
+        {
+            this.state = state;
+            commands.notifyAll();
+        }
+    }
+
     private void initReceiver()
     {
         synchronized (processedLock)
@@ -390,9 +399,26 @@ public class Session extends SessionInvoker
         {
             synchronized (commands)
             {
-                if (state == CLOSED)
+                if (state != OPEN && state != CLOSED)
                 {
+                    Waiter w = new Waiter(commands, timeout);
+                    while (w.hasTime() && (state != OPEN && state != CLOSED))
+                    {
+                        w.await();
+                    }
+                }
+
+                switch (state)
+                {
+                case OPEN:
+                    break;
+                case CLOSED:
                     throw new SessionClosedException();
+                default:
+                    throw new SessionException
+                        (String.format
+                         ("timed out waiting for session to become open %s",
+                          state));
                 }
 
                 int next = commandsOut++;
diff --git a/java/common/src/main/java/org/apache/qpid/transport/SessionDelegate.java b/java/common/src/main/java/org/apache/qpid/transport/SessionDelegate.java
index e2b6980dd4..7cc0ed6617 100644
--- a/java/common/src/main/java/org/apache/qpid/transport/SessionDelegate.java
+++ b/java/common/src/main/java/org/apache/qpid/transport/SessionDelegate.java
@@ -57,6 +57,11 @@ public class SessionDelegate
         log.warn("UNHANDLED: [%s] %s", ssn, method);
     }
 
+    @Override public void sessionAttached(Session ssn, SessionAttached atc)
+    {
+        ssn.setState(Session.State.OPEN);
+    }
+
     @Override public void sessionTimeout(Session ssn, SessionTimeout t)
     {
         ssn.setExpiry(t.getTimeout());