QPID-1339:

 - Modified QpidTestCase to start/stop multiple brokers for failover
   testing.

 - Modified QpidTestCase to substitute port variables into broker
   start/stop commands.

 - Modified test profiles to use the new port variables.

 - Modified QpidTestCase to permit multiple exclude files.

 - Modified test profiles to make use of a common exclude list:
   ExcludeList

 - Added ConnectionTest.testResumeEmptyReplayBuffer.

 - Made default exception handling for Connection and Session log the
   exception.

 - Added SenderExcetion to specifically signal problems with
   transmitting connection data.

 - Modified Session to catch and deal with connection send failures
   for sessions with positive expiry.

 - Modified FailoverBaseCase to work for non VM brokers.

 - Made FailoverTest fail if failover times out.

 - Modified JMS implementation to make use of the recently added low
   level session resume.

 - Unexcluded failover tests from 0-10 test profiles.

 - Excluded MultipleJCAProviderRegistrationTest due to its testing
   strategy resulting in spurious failure when running as part of the
   larger test suite.


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

1 files changed, 77 insertions, 34 deletions

diff --git a/java/common/src/test/java/org/apache/qpid/transport/ConnectionTest.java b/java/common/src/test/java/org/apache/qpid/transport/ConnectionTest.java
index b1fe08bfb9..1da56654f0 100644
--- a/java/common/src/test/java/org/apache/qpid/transport/ConnectionTest.java
+++ b/java/common/src/test/java/org/apache/qpid/transport/ConnectionTest.java
@@ -48,6 +48,7 @@ public class ConnectionTest extends TestCase implements SessionListener
     private int port;
     private volatile boolean queue = false;
     private List<MessageTransfer> messages = new ArrayList<MessageTransfer>();
+    private List<MessageTransfer> incoming = new ArrayList<MessageTransfer>();
 
     protected void setUp() throws Exception
     {
@@ -71,7 +72,7 @@ public class ConnectionTest extends TestCase implements SessionListener
 
     public void opened(Session ssn) {}
 
-    public void message(Session ssn, MessageTransfer xfr)
+    public void message(final Session ssn, MessageTransfer xfr)
     {
         if (queue)
         {
@@ -86,6 +87,25 @@ public class ConnectionTest extends TestCase implements SessionListener
         {
             ssn.getConnection().close();
         }
+        else if (body.startsWith("DELAYED_CLOSE"))
+        {
+            ssn.processed(xfr);
+            new Thread()
+            {
+                public void run()
+                {
+                    try
+                    {
+                        sleep(3000);
+                    }
+                    catch (InterruptedException e)
+                    {
+                        throw new RuntimeException(e);
+                    }
+                    ssn.getConnection().close();
+                }
+            }.start();
+        }
         else if (body.startsWith("ECHO"))
         {
             int id = xfr.getId();
@@ -139,7 +159,7 @@ public class ConnectionTest extends TestCase implements SessionListener
                 }
             }
         });
-        conn.connect("localhost", port, null, "guest", "guest",false);
+        conn.connect("localhost", port, null, "guest", "guest", false);
         return conn;
     }
 
@@ -148,7 +168,7 @@ public class ConnectionTest extends TestCase implements SessionListener
         Condition closed = new Condition();
         Connection conn = connect(closed);
 
-        Session ssn = conn.createSession();
+        Session ssn = conn.createSession(1);
         send(ssn, "CLOSE");
 
         if (!closed.get(3000))
@@ -167,44 +187,47 @@ public class ConnectionTest extends TestCase implements SessionListener
         }
     }
 
-    public void testResume() throws Exception
+    class FailoverConnectionListener implements ConnectionListener
     {
-        Connection conn = new Connection();
-        conn.connect("localhost", port, null, "guest", "guest",false);
+        public void opened(Connection conn) {}
 
-        conn.setConnectionListener(new ConnectionListener()
+        public void exception(Connection conn, ConnectionException e)
         {
-            public void opened(Connection conn) {}
-            public void exception(Connection conn, ConnectionException e)
-            {
-                throw e;
-            }
-            public void closed(Connection conn)
-            {
-                queue = true;
-                conn.connect("localhost", port, null, "guest", "guest",false);
-                conn.resume();
-            }
-        });
+            throw e;
+        }
 
-        Session ssn = conn.createSession(1);
-        final List<MessageTransfer> incoming = new ArrayList<MessageTransfer>();
-        ssn.setSessionListener(new SessionListener()
+        public void closed(Connection conn)
         {
-            public void opened(Session s) {}
-            public void exception(Session s, SessionException e) {}
-            public void message(Session s, MessageTransfer xfr)
-            {
-                synchronized (incoming)
-                {
-                    incoming.add(xfr);
-                    incoming.notifyAll();
-                }
+            queue = true;
+            conn.connect("localhost", port, null, "guest", "guest");
+            conn.resume();
+        }
+    }
 
-                s.processed(xfr);
+    class TestSessionListener implements SessionListener
+    {
+        public void opened(Session s) {}
+        public void exception(Session s, SessionException e) {}
+        public void message(Session s, MessageTransfer xfr)
+        {
+            synchronized (incoming)
+            {
+                incoming.add(xfr);
+                incoming.notifyAll();
             }
-            public void closed(Session s) {}
-        });
+
+            s.processed(xfr);
+        }
+        public void closed(Session s) {}
+    }
+
+    public void testResumeNonemptyReplayBuffer() throws Exception
+    {
+        Connection conn = new Connection();
+        conn.setConnectionListener(new FailoverConnectionListener());
+        conn.connect("localhost", port, null, "guest", "guest");
+        Session ssn = conn.createSession(1);
+        ssn.setSessionListener(new TestSessionListener());
 
         send(ssn, "SINK 0");
         send(ssn, "ECHO 1");
@@ -251,4 +274,24 @@ public class ConnectionTest extends TestCase implements SessionListener
         }
     }
 
+    public void testResumeEmptyReplayBuffer() throws InterruptedException
+    {
+        Connection conn = new Connection();
+        conn.setConnectionListener(new FailoverConnectionListener());
+        conn.connect("localhost", port, null, "guest", "guest");
+        Session ssn = conn.createSession(1);
+        ssn.setSessionListener(new TestSessionListener());
+
+        send(ssn, "SINK 0");
+        send(ssn, "SINK 1");
+        send(ssn, "DELAYED_CLOSE 2");
+        ssn.sync();
+        Thread.sleep(6000);
+        send(ssn, "SINK 3");
+        ssn.sync();
+        System.out.println(messages);
+        assertEquals(1, messages.size());
+        assertEquals("SINK 3", messages.get(0).getBodyString());
+    }
+
 }