QPID-1549: don't throw not implemented when we get a flush with the expected bit set

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

3 files changed, 35 insertions, 2 deletions

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 8877b7b683..22561dcc33 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
@@ -312,6 +312,19 @@ public class Session extends SessionInvoker
         }
     }
 
+    void flushExpected()
+    {
+        RangeSet rs = new RangeSet();
+        synchronized (processedLock)
+        {
+            if (incomingInit)
+            {
+                rs.add(commandsIn);
+            }
+        }
+        sessionExpected(rs, null);
+    }
+
     public void flushProcessed(Option ... options)
     {
         RangeSet copy;
diff --git a/qpid/java/common/src/main/java/org/apache/qpid/transport/SessionDelegate.java b/qpid/java/common/src/main/java/org/apache/qpid/transport/SessionDelegate.java
index 7cc0ed6617..520291e71c 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/transport/SessionDelegate.java
+++ b/qpid/java/common/src/main/java/org/apache/qpid/transport/SessionDelegate.java
@@ -115,7 +115,7 @@ public class SessionDelegate
         }
         if (flush.getExpected())
         {
-            throw new Error("not implemented");
+            ssn.flushExpected();
         }
     }
 
diff --git a/qpid/java/common/src/test/java/org/apache/qpid/transport/ConnectionTest.java b/qpid/java/common/src/test/java/org/apache/qpid/transport/ConnectionTest.java
index 8d3e363fdd..7ce41db4d0 100644
--- a/qpid/java/common/src/test/java/org/apache/qpid/transport/ConnectionTest.java
+++ b/qpid/java/common/src/test/java/org/apache/qpid/transport/ConnectionTest.java
@@ -36,6 +36,8 @@ import java.util.List;
 import java.util.Collections;
 import java.io.IOException;
 
+import static org.apache.qpid.transport.Option.*;
+
 /**
  * ConnectionTest
  */
@@ -62,7 +64,11 @@ public class ConnectionTest extends TestCase implements SessionListener
 
     protected void tearDown() throws Exception
     {
-        _ioa.close();
+        if (_ioa != null)
+        {
+            _ioa.close();
+        }
+
         super.tearDown();
     }
 
@@ -362,4 +368,18 @@ public class ConnectionTest extends TestCase implements SessionListener
         assertEquals("SINK 3", messages.get(0).getBodyString());
     }
 
+    public void testFlushExpected() throws InterruptedException
+    {
+        startServer();
+
+        Connection conn = new Connection();
+        conn.connect("localhost", port, null, "guest", "guest");
+        Session ssn = conn.createSession();
+        ssn.sessionFlush(EXPECTED);
+        send(ssn, "SINK 0");
+        ssn.sessionFlush(EXPECTED);
+        send(ssn, "SINK 1");
+        ssn.sync();
+    }
+
 }