Corrected VMBroker Creation and clean up per test

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

18 files changed, 462 insertions, 110 deletions

diff --git a/java/client/test/src/org/apache/qpid/ack/DisconnectAndRedeliverTest.java b/java/client/test/src/org/apache/qpid/ack/DisconnectAndRedeliverTest.java
index 6034247e48..fca04c8330 100644
--- a/java/client/test/src/org/apache/qpid/ack/DisconnectAndRedeliverTest.java
+++ b/java/client/test/src/org/apache/qpid/ack/DisconnectAndRedeliverTest.java
@@ -23,6 +23,7 @@ import org.apache.log4j.xml.DOMConfigurator;
 import org.apache.qpid.client.AMQConnection;
 import org.apache.qpid.client.AMQQueue;
 import org.apache.qpid.client.AMQSession;
+import org.apache.qpid.client.vmbroker.AMQVMBrokerCreationException;
 import org.apache.qpid.client.transport.TransportConnection;
 import org.apache.qpid.server.registry.ApplicationRegistry;
 import org.apache.qpid.server.store.TestableMemoryMessageStore;
@@ -38,6 +39,7 @@ public class DisconnectAndRedeliverTest
 {
     private static final Logger _logger = Logger.getLogger(DisconnectAndRedeliverTest.class);
 
+
     static
     {
         String workdir = System.getProperty("QPID_WORK");
@@ -54,7 +56,21 @@ public class DisconnectAndRedeliverTest
     @Before
     public void resetAppliactionRegistry() throws Exception
     {
-        ApplicationRegistry.initialise(new TestApplicationRegistry());
+        createVMBroker();
+        ApplicationRegistry.initialise(new TestApplicationRegistry(), 1);
+    }
+
+
+    public void createVMBroker()
+    {
+        try
+        {
+            TransportConnection.createVMBroker(1);
+        }
+        catch (AMQVMBrokerCreationException e)
+        {
+            Assert.fail("Unable to create broker: " + e);
+        }
     }
 
     @After
@@ -83,6 +99,8 @@ public class DisconnectAndRedeliverTest
         ((AMQSession) consumerSession).declareExchangeSynch("amq.direct", "direct");
 
         Connection con2 = new AMQConnection("vm://:1", "guest", "guest", "producer1", "/test");
+                 
+
         Session producerSession = con2.createSession(false, Session.CLIENT_ACKNOWLEDGE);
         MessageProducer producer = producerSession.createProducer(queue);
 
diff --git a/java/client/test/src/org/apache/qpid/ack/RecoverTest.java b/java/client/test/src/org/apache/qpid/ack/RecoverTest.java
index e12ca45de3..78be978e2e 100644
--- a/java/client/test/src/org/apache/qpid/ack/RecoverTest.java
+++ b/java/client/test/src/org/apache/qpid/ack/RecoverTest.java
@@ -49,6 +49,19 @@ public class RecoverTest
         DOMConfigurator.configure("broker/etc/log4j.xml");
     }
 
+    @Before
+    public void createVMBroker()
+    {
+        try
+        {
+            TransportConnection.createVMBroker(1);
+        }
+        catch (AMQVMBrokerCreationException e)
+        {
+            Assert.fail("Unable to create broker: " + e);
+        }
+    }
+
     @After
     public void stopVmBroker()
     {
diff --git a/java/client/test/src/org/apache/qpid/basic/BytesMessageTest.java b/java/client/test/src/org/apache/qpid/basic/BytesMessageTest.java
index e8a9debe1d..a832213066 100644
--- a/java/client/test/src/org/apache/qpid/basic/BytesMessageTest.java
+++ b/java/client/test/src/org/apache/qpid/basic/BytesMessageTest.java
@@ -22,9 +22,14 @@ import org.apache.qpid.client.AMQConnection;
 import org.apache.qpid.client.AMQDestination;
 import org.apache.qpid.client.AMQQueue;
 import org.apache.qpid.client.AMQSession;
+import org.apache.qpid.client.vmbroker.AMQVMBrokerCreationException;
+import org.apache.qpid.client.transport.TransportConnection;
 import org.apache.qpid.client.message.JMSBytesMessage;
 import org.apache.mina.common.ByteBuffer;
 import org.junit.Test;
+import org.junit.Before;
+import org.junit.Assert;
+import org.junit.After;
 
 import javax.jms.*;
 import java.util.ArrayList;
@@ -41,9 +46,29 @@ public class BytesMessageTest implements MessageListener
     private int _count = 100;
     public String _connectionString = "vm://:1";
 
-    void init() throws Exception
+    @Before
+    public void init() throws Exception
     {
         init(new AMQConnection(_connectionString, "guest", "guest", randomize("Client"), "/test_path"));
+        createVMBroker();
+    }
+
+    public void createVMBroker()
+    {
+        try
+        {
+            TransportConnection.createVMBroker(1);
+        }
+        catch (AMQVMBrokerCreationException e)
+        {
+            Assert.fail("Unable to create broker: " + e);
+        }
+    }
+
+    @After
+    public void stopVmBroker()
+    {
+        TransportConnection.killVMBroker(1);
     }
 
     void init(AMQConnection connection) throws Exception
@@ -65,7 +90,6 @@ public class BytesMessageTest implements MessageListener
     @Test
     public void test() throws Exception
     {
-        init();
 
         try
         {
@@ -96,7 +120,7 @@ public class BytesMessageTest implements MessageListener
 
     void waitFor(int count) throws InterruptedException
     {
-        synchronized (received)
+        synchronized(received)
         {
             while (received.size() < count)
             {
@@ -164,7 +188,7 @@ public class BytesMessageTest implements MessageListener
 
     public void onMessage(Message message)
     {
-        synchronized (received)
+        synchronized(received)
         {
             received.add((JMSBytesMessage) message);
             received.notify();
diff --git a/java/client/test/src/org/apache/qpid/basic/FieldTableMessageTest.java b/java/client/test/src/org/apache/qpid/basic/FieldTableMessageTest.java
index 82e43773a4..545f7b1094 100644
--- a/java/client/test/src/org/apache/qpid/basic/FieldTableMessageTest.java
+++ b/java/client/test/src/org/apache/qpid/basic/FieldTableMessageTest.java
@@ -22,6 +22,8 @@ import org.apache.qpid.client.AMQConnection;
 import org.apache.qpid.client.AMQDestination;
 import org.apache.qpid.client.AMQQueue;
 import org.apache.qpid.client.AMQSession;
+import org.apache.qpid.client.vmbroker.AMQVMBrokerCreationException;
+import org.apache.qpid.client.transport.TransportConnection;
 import org.apache.qpid.client.message.JMSBytesMessage;
 import org.apache.qpid.framing.AMQFrameDecodingException;
 import org.apache.qpid.framing.FieldTable;
@@ -29,6 +31,8 @@ import org.apache.qpid.framing.FieldTableTest;
 import org.apache.mina.common.ByteBuffer;
 import org.junit.Before;
 import org.junit.Test;
+import org.junit.Assert;
+import org.junit.After;
 
 import javax.jms.*;
 import java.io.IOException;
@@ -49,8 +53,28 @@ public class FieldTableMessageTest implements MessageListener
     public void init() throws Exception
     {
         init(new AMQConnection(_connectionString, "guest", "guest", randomize("Client"), "/test_path"));
+        createVMBroker();
     }
 
+    public void createVMBroker()
+    {
+        try
+        {
+            TransportConnection.createVMBroker(1);
+        }
+        catch (AMQVMBrokerCreationException e)
+        {
+            Assert.fail("Unable to create broker: " + e);
+        }
+    }
+
+    @After
+    public void stopVmBroker()
+    {
+        TransportConnection.killVMBroker(1);
+    }
+
+
     private void init(AMQConnection connection) throws Exception
     {
         init(connection, new AMQQueue(randomize("FieldTableMessageTest"), true));
@@ -109,7 +133,7 @@ public class FieldTableMessageTest implements MessageListener
     {
         synchronized(received)
         {
-            while(received.size() < count)
+            while (received.size() < count)
             {
                 received.wait();
             }
diff --git a/java/client/test/src/org/apache/qpid/basic/ObjectMessageTest.java b/java/client/test/src/org/apache/qpid/basic/ObjectMessageTest.java
index a78cd6f72b..ceb2fa907e 100644
--- a/java/client/test/src/org/apache/qpid/basic/ObjectMessageTest.java
+++ b/java/client/test/src/org/apache/qpid/basic/ObjectMessageTest.java
@@ -22,9 +22,13 @@ import org.apache.qpid.client.AMQConnection;
 import org.apache.qpid.client.AMQDestination;
 import org.apache.qpid.client.AMQQueue;
 import org.apache.qpid.client.AMQSession;
+import org.apache.qpid.client.vmbroker.AMQVMBrokerCreationException;
+import org.apache.qpid.client.transport.TransportConnection;
 import org.apache.qpid.client.message.JMSObjectMessage;
 import org.junit.Before;
 import org.junit.Test;
+import org.junit.Assert;
+import org.junit.After;
 
 import javax.jms.JMSException;
 import javax.jms.Message;
@@ -50,6 +54,25 @@ public class ObjectMessageTest implements MessageListener
     {
         String broker = _connectionString;
         init(new AMQConnection(broker, "guest", "guest", randomize("Client"), "/test_path"));
+        createVMBroker();
+    }
+
+    public void createVMBroker()
+    {
+        try
+        {
+            TransportConnection.createVMBroker(1);
+        }
+        catch (AMQVMBrokerCreationException e)
+        {
+            Assert.fail("Unable to create broker: " + e);
+        }
+    }
+
+    @After
+    public void stopVmBroker()
+    {
+        TransportConnection.killVMBroker(1);
     }
 
     private void init(AMQConnection connection) throws Exception
@@ -95,7 +118,7 @@ public class ObjectMessageTest implements MessageListener
     {
         synchronized(received)
         {
-            while(received.size() < count)
+            while (received.size() < count)
             {
                 received.wait();
             }
@@ -116,25 +139,26 @@ public class ObjectMessageTest implements MessageListener
     private static void assertEqual(Iterator expected, Iterator actual)
     {
         List<String> errors = new ArrayList<String>();
-        while(expected.hasNext() && actual.hasNext())
+        while (expected.hasNext() && actual.hasNext())
         {
-            try{
+            try
+            {
                 assertEqual(expected.next(), actual.next());
             }
-            catch(Exception e)
+            catch (Exception e)
             {
                 errors.add(e.getMessage());
             }
         }
-        while(expected.hasNext())
+        while (expected.hasNext())
         {
             errors.add("Expected " + expected.next() + " but no more actual values.");
         }
-        while(actual.hasNext())
+        while (actual.hasNext())
         {
             errors.add("Found " + actual.next() + " but no more expected values.");
         }
-        if(!errors.isEmpty())
+        if (!errors.isEmpty())
         {
             throw new RuntimeException(errors.toString());
         }
@@ -142,7 +166,7 @@ public class ObjectMessageTest implements MessageListener
 
     private static void assertEqual(Object expected, Object actual)
     {
-        if(!expected.equals(actual))
+        if (!expected.equals(actual))
         {
             throw new RuntimeException("Expected '" + expected + "' found '" + actual + "'");
         }
@@ -183,7 +207,7 @@ public class ObjectMessageTest implements MessageListener
 
         public String toString()
         {
-            return "Payload[" + data +"]";
+            return "Payload[" + data + "]";
         }
     }
 
diff --git a/java/client/test/src/org/apache/qpid/basic/SessionStartTest.java b/java/client/test/src/org/apache/qpid/basic/SessionStartTest.java
index f201f23681..edd7beae14 100644
--- a/java/client/test/src/org/apache/qpid/basic/SessionStartTest.java
+++ b/java/client/test/src/org/apache/qpid/basic/SessionStartTest.java
@@ -40,12 +40,31 @@ public class SessionStartTest implements MessageListener
     private AMQDestination _destination;
     private AMQSession _session;
     private int count;
-    public String _connectionString = "vm://:1";  
+    public String _connectionString = "vm://:1";
 
     @Before
     public void init() throws Exception
     {
         init(new AMQConnection(_connectionString, "guest", "guest", randomize("Client"), "/test_path"));
+        createVMBroker();
+    }
+
+    public void createVMBroker()
+    {
+        try
+        {
+            TransportConnection.createVMBroker(1);
+        }
+        catch (AMQVMBrokerCreationException e)
+        {
+            Assert.fail("Unable to create broker: " + e);
+        }
+    }
+
+    @After
+    public void stopVmBroker()
+    {
+        TransportConnection.killVMBroker(1);
     }
 
     private void init(AMQConnection connection) throws Exception
diff --git a/java/client/test/src/org/apache/qpid/basic/TextMessageTest.java b/java/client/test/src/org/apache/qpid/basic/TextMessageTest.java
index 2b76f64659..5bf5da2fb7 100644
--- a/java/client/test/src/org/apache/qpid/basic/TextMessageTest.java
+++ b/java/client/test/src/org/apache/qpid/basic/TextMessageTest.java
@@ -22,9 +22,13 @@ import org.apache.qpid.client.AMQConnection;
 import org.apache.qpid.client.AMQDestination;
 import org.apache.qpid.client.AMQQueue;
 import org.apache.qpid.client.AMQSession;
+import org.apache.qpid.client.vmbroker.AMQVMBrokerCreationException;
+import org.apache.qpid.client.transport.TransportConnection;
 import org.apache.qpid.client.message.JMSTextMessage;
 import org.junit.Before;
 import org.junit.Test;
+import org.junit.Assert;
+import org.junit.After;
 
 import javax.jms.*;
 import java.util.ArrayList;
@@ -45,6 +49,25 @@ public class TextMessageTest implements MessageListener
     public void init() throws Exception
     {
         init(new AMQConnection(_connectionString, "guest", "guest", randomize("Client"), "/test_path"));
+        createVMBroker();
+    }
+
+    public void createVMBroker()
+    {
+        try
+        {
+            TransportConnection.createVMBroker(1);
+        }
+        catch (AMQVMBrokerCreationException e)
+        {
+            Assert.fail("Unable to create broker: " + e);
+        }
+    }
+
+    @After
+    public void stopVmBroker()
+    {
+        TransportConnection.killVMBroker(1);
     }
 
     private void init(AMQConnection connection) throws Exception
@@ -91,7 +114,7 @@ public class TextMessageTest implements MessageListener
     {
         synchronized(received)
         {
-            while(received.size() < count)
+            while (received.size() < count)
             {
                 received.wait();
             }
@@ -112,25 +135,26 @@ public class TextMessageTest implements MessageListener
     private static void assertEqual(Iterator expected, Iterator actual)
     {
         List<String> errors = new ArrayList<String>();
-        while(expected.hasNext() && actual.hasNext())
+        while (expected.hasNext() && actual.hasNext())
         {
-            try{
+            try
+            {
                 assertEqual(expected.next(), actual.next());
             }
-            catch(Exception e)
+            catch (Exception e)
             {
                 errors.add(e.getMessage());
             }
         }
-        while(expected.hasNext())
+        while (expected.hasNext())
         {
             errors.add("Expected " + expected.next() + " but no more actual values.");
         }
-        while(actual.hasNext())
+        while (actual.hasNext())
         {
             errors.add("Found " + actual.next() + " but no more expected values.");
         }
-        if(!errors.isEmpty())
+        if (!errors.isEmpty())
         {
             throw new RuntimeException(errors.toString());
         }
@@ -138,7 +162,7 @@ public class TextMessageTest implements MessageListener
 
     private static void assertEqual(Object expected, Object actual)
     {
-        if(!expected.equals(actual))
+        if (!expected.equals(actual))
         {
             throw new RuntimeException("Expected '" + expected + "' found '" + actual + "'");
         }
@@ -163,7 +187,10 @@ public class TextMessageTest implements MessageListener
         TextMessageTest test = new TextMessageTest();
         test._connectionString = argv.length == 0 ? "vm://:1" : argv[0];
         test.init();
-        if (argv.length > 1) test._count = Integer.parseInt(argv[1]);
+        if (argv.length > 1)
+        {
+            test._count = Integer.parseInt(argv[1]);
+        }
         test.test();
     }
 
diff --git a/java/client/test/src/org/apache/qpid/client/TestAMQConnection.java b/java/client/test/src/org/apache/qpid/client/TestAMQConnection.java
index e656bf5fff..1858abf536 100644
--- a/java/client/test/src/org/apache/qpid/client/TestAMQConnection.java
+++ b/java/client/test/src/org/apache/qpid/client/TestAMQConnection.java
@@ -26,7 +26,8 @@ import junit.framework.JUnit4TestAdapter;
 

 import javax.jms.*;

 

-public class TestAMQConnection {

+public class TestAMQConnection

+{

 

     private static AMQConnection _connection;

     private static AMQTopic _topic;

@@ -34,18 +35,47 @@ public class TestAMQConnection {
     private static QueueSession _queueSession;

     private static TopicSession _topicSession;

 

-    @BeforeClass

-    public static void setUp() throws AMQException, URLSyntaxException, JMSException {

+

+    @Before

+    public void setUp() throws AMQException, URLSyntaxException, JMSException

+    {

+        createVMBroker();

         //initialise the variables we need for testing

         _connection = new AMQConnection("vm://:1", "guest", "guest", "fred", "/test");

         _topic = new AMQTopic("mytopic");

         _queue = new AMQQueue("myqueue");

     }

 

+    public void createVMBroker()

+    {

+        try

+        {

+            TransportConnection.createVMBroker(1);

+        }

+        catch (AMQVMBrokerCreationException e)

+        {

+            Assert.fail("Unable to create broker: " + e);

+        }

+    }

+

+    @After

+    public void stopVmBroker()

+    {

+        try

+        {

+            _connection.close();

+        }

+        catch (JMSException e)

+        {

+            //ignore 

+        }

+        TransportConnection.killVMBroker(1);

+    }

+

     /**

-    * Simple tests to check we can create TopicSession and QueueSession ok

-    * And that they throw exceptions where appropriate as per JMS spec

-    */

+     * Simple tests to check we can create TopicSession and QueueSession ok

+     * And that they throw exceptions where appropriate as per JMS spec

+     */

 

     @Test

     public void testCreateQueueSession() throws JMSException

@@ -59,38 +89,45 @@ public class TestAMQConnection {
         _topicSession = _connection.createTopicSession(false, AMQSession.NO_ACKNOWLEDGE);

     }

 

-    @Test(expected=javax.jms.IllegalStateException.class)

-    public void testTopicSessionCreateBrowser() throws JMSException {

+    @Test(expected = javax.jms.IllegalStateException.class)

+    public void testTopicSessionCreateBrowser() throws JMSException

+    {

         _topicSession.createBrowser(_queue);

     }

 

-    @Test(expected=javax.jms.IllegalStateException.class)

-    public void testTopicSessionCreateQueue() throws JMSException {

+    @Test(expected = javax.jms.IllegalStateException.class)

+    public void testTopicSessionCreateQueue() throws JMSException

+    {

         _topicSession.createQueue("abc");

     }

 

-    @Test(expected=javax.jms.IllegalStateException.class)

-    public void testTopicSessionCreateTemporaryQueue() throws JMSException {

+    @Test(expected = javax.jms.IllegalStateException.class)

+    public void testTopicSessionCreateTemporaryQueue() throws JMSException

+    {

         _topicSession.createTemporaryQueue();

     }

 

-    @Test(expected=javax.jms.IllegalStateException.class)

-    public void testQueueSessionCreateTemporaryTopic() throws JMSException {

+    @Test(expected = javax.jms.IllegalStateException.class)

+    public void testQueueSessionCreateTemporaryTopic() throws JMSException

+    {

         _queueSession.createTemporaryTopic();

     }

 

-    @Test(expected=javax.jms.IllegalStateException.class)

-    public void testQueueSessionCreateTopic() throws JMSException {

+    @Test(expected = javax.jms.IllegalStateException.class)

+    public void testQueueSessionCreateTopic() throws JMSException

+    {

         _queueSession.createTopic("abc");

     }

 

-    @Test(expected=javax.jms.IllegalStateException.class)

-    public void testQueueSessionDurableSubscriber() throws JMSException {

-        _queueSession.createDurableSubscriber(_topic,"abc");

+    @Test(expected = javax.jms.IllegalStateException.class)

+    public void testQueueSessionDurableSubscriber() throws JMSException

+    {

+        _queueSession.createDurableSubscriber(_topic, "abc");

     }

 

-    @Test(expected=javax.jms.IllegalStateException.class)

-    public void testQueueSessionUnsubscribe() throws JMSException {

+    @Test(expected = javax.jms.IllegalStateException.class)

+    public void testQueueSessionUnsubscribe() throws JMSException

+    {

         _queueSession.unsubscribe("abc");

     }

 

diff --git a/java/client/test/src/org/apache/qpid/client/TestAMQSession.java b/java/client/test/src/org/apache/qpid/client/TestAMQSession.java
index 48ccc843f1..99da8c5b10 100644
--- a/java/client/test/src/org/apache/qpid/client/TestAMQSession.java
+++ b/java/client/test/src/org/apache/qpid/client/TestAMQSession.java
@@ -38,15 +38,44 @@ public class TestAMQSession
     private static AMQSession _session;

     private static AMQTopic _topic;

     private static AMQQueue _queue;

+    private static AMQConnection _connection;

 

-    @BeforeClass

-    public static void setUp() throws AMQException, URLSyntaxException, JMSException

+    @Before

+    public  void setUp() throws AMQException, URLSyntaxException, JMSException

     {

+        createVMBroker();

         //initialise the variables we need for testing

-        AMQConnection connection = new AMQConnection("vm://:1", "guest", "guest", "fred", "/test");

+        _connection = new AMQConnection("vm://:1", "guest", "guest", "fred", "/test");

         _topic = new AMQTopic("mytopic");

         _queue = new AMQQueue("myqueue");

-        _session = (AMQSession) connection.createSession(false, AMQSession.NO_ACKNOWLEDGE);

+        _session = (AMQSession) _connection.createSession(false, AMQSession.NO_ACKNOWLEDGE);

+    }

+

+

+    public void createVMBroker()

+    {

+        try

+        {

+            TransportConnection.createVMBroker(1);

+        }

+        catch (AMQVMBrokerCreationException e)

+        {

+            Assert.fail("Unable to create broker: " + e);

+        }

+    }

+

+    @After

+    public void stopVmBroker()

+    {

+        try

+        {

+            _connection.close();

+        }

+        catch (JMSException e)

+        {

+            //just close

+        }

+        TransportConnection.killVMBroker(1);

     }

 

     @Test

diff --git a/java/client/test/src/org/apache/qpid/client/channelclose/ChannelCloseOkTest.java b/java/client/test/src/org/apache/qpid/client/channelclose/ChannelCloseOkTest.java
index e12d6c0fc5..7b714a01b7 100644
--- a/java/client/test/src/org/apache/qpid/client/channelclose/ChannelCloseOkTest.java
+++ b/java/client/test/src/org/apache/qpid/client/channelclose/ChannelCloseOkTest.java
@@ -20,11 +20,14 @@ package org.apache.qpid.client.channelclose;
 import junit.framework.JUnit4TestAdapter;
 import org.apache.qpid.client.AMQConnection;
 import org.apache.qpid.client.AMQQueue;
+import org.apache.qpid.client.vmbroker.AMQVMBrokerCreationException;
+import org.apache.qpid.client.transport.TransportConnection;
 import org.apache.log4j.Logger;
 import org.junit.After;
 import static org.junit.Assert.assertEquals;
 import org.junit.Before;
 import org.junit.Test;
+import org.junit.Assert;
 
 import javax.jms.*;
 import java.util.ArrayList;
@@ -33,16 +36,16 @@ import java.util.List;
 /**
  * Due to bizarre exception handling all sessions are closed if you get
  * a channel close request and no exception listener is registered.
- *
+ * <p/>
  * JIRA issue IBTBLZ-10.
- *
+ * <p/>
  * Simulate by:
- *
+ * <p/>
  * 0. Create two sessions with no exception listener.
  * 1. Publish message to queue/topic that does not exist (wrong routing key).
  * 2. This will cause a channel close.
  * 3. Since client does not have an exception listener, currently all sessions are
- *    closed.
+ * closed.
  */
 public class ChannelCloseOkTest
 {
@@ -57,18 +60,40 @@ public class ChannelCloseOkTest
     private final static Logger _log = Logger.getLogger(ChannelCloseOkTest.class);
     public String _connectionString = "vm://:1";
 
+
+    public void createVMBroker()
+    {
+        try
+        {
+            TransportConnection.createVMBroker(1);
+        }
+        catch (AMQVMBrokerCreationException e)
+        {
+            Assert.fail("Unable to create broker: " + e);
+        }
+    }
+
+    public void stopVmBroker()
+    {
+        TransportConnection.killVMBroker(1);
+    }
+
     @Before
     public void init() throws Exception
     {
+        createVMBroker();
+
         _connection = new AMQConnection(_connectionString, "guest", "guest", randomize("Client"), "/test_path");
+
         _destination1 = new AMQQueue("q1", true);
         _destination2 = new AMQQueue("q2", true);
         _session1 = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
-        _session1.createConsumer(_destination1).setMessageListener(new MessageListener() {
+        _session1.createConsumer(_destination1).setMessageListener(new MessageListener()
+        {
             public void onMessage(Message message)
             {
                 _log.debug("consumer 1 got message [" + getTextMessage(message) + "]");
-                synchronized (_received1)
+                synchronized(_received1)
                 {
                     _received1.add(message);
                     _received1.notify();
@@ -76,11 +101,12 @@ public class ChannelCloseOkTest
             }
         });
         _session2 = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
-        _session2.createConsumer(_destination2).setMessageListener(new MessageListener() {
+        _session2.createConsumer(_destination2).setMessageListener(new MessageListener()
+        {
             public void onMessage(Message message)
             {
-                _log.debug("consumer 2 got message [" +  getTextMessage(message) + "]");
-                synchronized (_received2)
+                _log.debug("consumer 2 got message [" + getTextMessage(message) + "]");
+                synchronized(_received2)
                 {
                     _received2.add(message);
                     _received2.notify();
@@ -93,7 +119,7 @@ public class ChannelCloseOkTest
 
     private String getTextMessage(Message message)
     {
-        TextMessage tm = (TextMessage)message;
+        TextMessage tm = (TextMessage) message;
         try
         {
             return tm.getText();
@@ -112,6 +138,8 @@ public class ChannelCloseOkTest
             System.out.println(">>>>>>>>>>>>>>.. closing");
             _connection.close();
         }
+
+        stopVmBroker();
     }
 
     @Test
@@ -123,7 +151,8 @@ public class ChannelCloseOkTest
     @Test
     public void testWithExceptionListener() throws Exception
     {
-        _connection.setExceptionListener(new ExceptionListener() {
+        _connection.setExceptionListener(new ExceptionListener()
+        {
             public void onException(JMSException jmsException)
             {
                 _log.error("onException - ", jmsException);
@@ -177,7 +206,7 @@ public class ChannelCloseOkTest
 
     private void waitFor(List<Message> received, int count) throws InterruptedException
     {
-        synchronized (received)
+        synchronized(received)
         {
             while (received.size() < count)
             {
diff --git a/java/client/test/src/org/apache/qpid/forwardall/Combined.java b/java/client/test/src/org/apache/qpid/forwardall/Combined.java
index ea5216dd2d..f767e8b1e8 100644
--- a/java/client/test/src/org/apache/qpid/forwardall/Combined.java
+++ b/java/client/test/src/org/apache/qpid/forwardall/Combined.java
@@ -31,18 +31,36 @@ import org.apache.qpid.client.vmbroker.AMQVMBrokerCreationException;
  */
 public class Combined
 {
+
+    @Before
+    public void createVMBroker()
+    {
+        try
+        {
+            TransportConnection.createVMBroker(1);
+        }
+        catch (AMQVMBrokerCreationException e)
+        {
+            Assert.fail("Unable to create broker: " + e);
+        }
+    }
+
+    @After
+    public void stopVmBroker()
+    {
+        ServiceCreator.closeAll();
+        TransportConnection.killVMBroker(1);
+    }
+
     @Test
     public void forwardAll() throws Exception
     {
         int services = 2;
         ServiceCreator.start("vm://:1", services);
 
-        //give them time to get registered etc.
-        System.out.println("Services started, waiting for them to initialise...");
-        Thread.sleep(5 * 1000);
         System.out.println("Starting client...");
 
-        new Client("vm://:1", services).waitUntilComplete();
+        new Client("vm://:1", services).shutdownWhenComplete();
 
         System.out.println("Completed successfully!");
     }
diff --git a/java/client/test/src/org/apache/qpid/forwardall/Service.java b/java/client/test/src/org/apache/qpid/forwardall/Service.java
index bc3cc4c1f4..f0a62456f5 100644
--- a/java/client/test/src/org/apache/qpid/forwardall/Service.java
+++ b/java/client/test/src/org/apache/qpid/forwardall/Service.java
@@ -24,6 +24,7 @@ import org.apache.qpid.client.AMQSession;
 import javax.jms.Destination;
 import javax.jms.Message;
 import javax.jms.MessageListener;
+import javax.jms.JMSException;
 
 /**
  * Declare a queue and bind it to amq.direct with a 'well known' routing key,
@@ -56,12 +57,17 @@ public class Service implements MessageListener
             Destination replyTo = request.getJMSReplyTo();
             _session.createProducer(replyTo).send(response);
         }
-        catch(Exception e)
+        catch (Exception e)
         {
             e.printStackTrace(System.out);
         }
     }
 
+    public void close() throws JMSException
+    {
+        _connection.close();
+    }
+
     static AMQConnection connect(String broker) throws Exception
     {
         return new AMQConnection(broker, "guest", "guest", "Client" + System.currentTimeMillis(), "/test_path");
diff --git a/java/client/test/src/org/apache/qpid/forwardall/ServiceCreator.java b/java/client/test/src/org/apache/qpid/forwardall/ServiceCreator.java
index 6d43f85cf2..e58cc924d4 100644
--- a/java/client/test/src/org/apache/qpid/forwardall/ServiceCreator.java
+++ b/java/client/test/src/org/apache/qpid/forwardall/ServiceCreator.java
@@ -17,9 +17,15 @@
  */
 package org.apache.qpid.forwardall;
 
+import javax.jms.JMSException;
+
 public class ServiceCreator implements Runnable
 {
+    private static Thread[] threads;
+    private static ServiceCreator[] _services;
+
     private final String broker;
+    private Service service;
 
     ServiceCreator(String broker)
     {
@@ -30,7 +36,7 @@ public class ServiceCreator implements Runnable
     {
         try
         {
-            new Service(broker);
+            service = new Service(broker);
         }
         catch (Exception e)
         {
@@ -38,15 +44,37 @@ public class ServiceCreator implements Runnable
         }
     }
 
+    public void closeSC() throws JMSException
+    {
+        service.close();
+    }
+
+    static void closeAll()
+    {
+        for (int i = 0; i < _services.length; i++)
+        {
+            try
+            {
+                _services[i].closeSC();
+            }
+            catch (JMSException e)
+            {
+                //ignore
+            }
+        }
+    }
+
     static void start(String broker, int services) throws InterruptedException
     {
-        Thread[] threads = new Thread[services];
+        threads = new Thread[services];
+        _services = new ServiceCreator[services];
         ServiceCreator runner = new ServiceCreator(broker);
         //start services
         System.out.println("Starting " + services + " services...");
         for (int i = 0; i < services; i++)
         {
             threads[i] = new Thread(runner);
+            _services[i] = runner;
             threads[i].start();
         }
 
@@ -60,7 +88,8 @@ public class ServiceCreator implements Runnable
     {
         final String connectionString;
         final int services;
-        if (argv.length == 0) {
+        if (argv.length == 0)
+        {
             connectionString = "localhost:5672";
             services = 100;
         }
diff --git a/java/client/test/src/org/apache/qpid/jndi/referenceabletest/Bind.java b/java/client/test/src/org/apache/qpid/jndi/referenceabletest/Bind.java
index 221237a37c..48523183fb 100644
--- a/java/client/test/src/org/apache/qpid/jndi/referenceabletest/Bind.java
+++ b/java/client/test/src/org/apache/qpid/jndi/referenceabletest/Bind.java
@@ -91,7 +91,7 @@ class Bind
             }
         }
 
-
+        Connection connection = null;
         try
         {
             // Create the initial context
@@ -100,7 +100,7 @@ class Bind
             // Create the connection factory to be bound
             ConnectionFactory connectionFactory = null;
             // Create the Connection to be bound
-            Connection connection = null;
+
 
             try
             {
@@ -166,6 +166,20 @@ class Bind
                 throw(NoInitialContextException) e;
             }
         }
+        finally
+        {
+            try
+            {
+                if (connection != null)
+                {
+                    connection.close();
+                }
+            }
+            catch (JMSException e)
+            {
+                //ignore just want it closed
+            }
+        }
     }
 
     public String connectionFactoryValue()
diff --git a/java/client/test/src/org/apache/qpid/jndi/referenceabletest/JNDIReferenceableTest.java b/java/client/test/src/org/apache/qpid/jndi/referenceabletest/JNDIReferenceableTest.java
index 41af18b2d5..1b3caf8d47 100644
--- a/java/client/test/src/org/apache/qpid/jndi/referenceabletest/JNDIReferenceableTest.java
+++ b/java/client/test/src/org/apache/qpid/jndi/referenceabletest/JNDIReferenceableTest.java
@@ -19,6 +19,10 @@ package org.apache.qpid.jndi.referenceabletest;
 
 import org.junit.Test;
 import org.junit.Assert;
+import org.junit.After;
+import org.junit.Before;
+import org.apache.qpid.client.transport.TransportConnection;
+import org.apache.qpid.client.vmbroker.AMQVMBrokerCreationException;
 import junit.framework.JUnit4TestAdapter;
 
 import javax.naming.NameAlreadyBoundException;
@@ -37,6 +41,24 @@ import javax.naming.NoInitialContextException;
  */
 public class JNDIReferenceableTest
 {
+    @Before
+    public void createVMBroker()
+    {
+        try
+        {
+            TransportConnection.createVMBroker(1);
+        }
+        catch (AMQVMBrokerCreationException e)
+        {
+            Assert.fail("Unable to create broker: " + e);
+        }
+    }
+
+    @After
+    public void stopVmBroker()
+    {
+        TransportConnection.killVMBroker(1);
+    }
 
     @Test
     public void referenceable()
@@ -70,19 +92,19 @@ public class JNDIReferenceableTest
         catch (NoInitialContextException e)
         {
             Assert.fail("You don't have the File System SPI on you class path.\n" +
-                    "This can be downloaded from sun here:\n" +
-                    "http://java.sun.com/products/jndi/downloads/index.html\n" +
-                    "Click : Download JNDI 1.2.1 & More button\n" +
-                    "Download: File System Service Provider, 1.2 Beta 3\n" +
-                    "and add the two jars in the lib dir to your class path.");
+                        "This can be downloaded from sun here:\n" +
+                        "http://java.sun.com/products/jndi/downloads/index.html\n" +
+                        "Click : Download JNDI 1.2.1 & More button\n" +
+                        "Download: File System Service Provider, 1.2 Beta 3\n" +
+                        "and add the two jars in the lib dir to your class path.");
         }
 
         Assert.assertTrue(b.bound());
-
+        
         Lookup l = new Lookup();
 
         Assert.assertTrue(l.connectionFactoryValue().equals(b.connectionFactoryValue()));
-        
+
         Assert.assertTrue(l.connectionValue().equals(b.connectionValue()));
 
         Assert.assertTrue(l.topicValue().equals(b.topicValue()));
diff --git a/java/client/test/src/org/apache/qpid/jndi/referenceabletest/Lookup.java b/java/client/test/src/org/apache/qpid/jndi/referenceabletest/Lookup.java
index 481500b8f0..f2530b03b6 100644
--- a/java/client/test/src/org/apache/qpid/jndi/referenceabletest/Lookup.java
+++ b/java/client/test/src/org/apache/qpid/jndi/referenceabletest/Lookup.java
@@ -24,6 +24,7 @@ import org.apache.qpid.client.AMQTopic;
 import javax.naming.Context;
 import javax.naming.InitialContext;
 import javax.naming.NamingException;
+import javax.jms.JMSException;
 import java.io.File;
 import java.util.Hashtable;
 
@@ -46,6 +47,7 @@ class Lookup
     AMQTopic _topic = null;
     AMQConnection _connection = null;
     AMQConnectionFactory _connectionFactory = null;
+    private String _connectionURL;
 
     public Lookup()
     {
@@ -83,6 +85,8 @@ class Lookup
 
             _connection = (AMQConnection) ctx.lookup("Connection");
 
+            _connectionURL = _connection.toURL();
+
             _connectionFactory = (AMQConnectionFactory) ctx.lookup("ConnectionFactory");
             //System.out.println(topic);
 
@@ -93,6 +97,20 @@ class Lookup
         {
             System.out.println("Operation failed: " + e);
         }
+        finally
+        {
+            try
+            {
+                if (_connection != null)
+                {
+                    _connection.close();
+                }
+            }
+            catch (JMSException e)
+            {
+                //ignore just need to close
+            }
+        }
     }
 
     public String connectionFactoryValue()
@@ -102,7 +120,7 @@ class Lookup
 
     public String connectionValue()
     {
-        return _connection.toURL();
+        return _connectionURL;
     }
 
     public String topicValue()
diff --git a/java/client/test/src/org/apache/qpid/jndi/referenceabletest/Unbind.java b/java/client/test/src/org/apache/qpid/jndi/referenceabletest/Unbind.java
index 767f7c5737..3440ffd6ce 100644
--- a/java/client/test/src/org/apache/qpid/jndi/referenceabletest/Unbind.java
+++ b/java/client/test/src/org/apache/qpid/jndi/referenceabletest/Unbind.java
@@ -21,6 +21,8 @@ import javax.naming.Context;
 import javax.naming.InitialContext;
 import javax.naming.NameNotFoundException;
 import javax.naming.NamingException;
+import javax.jms.Connection;
+import javax.jms.JMSException;
 import java.io.File;
 import java.util.Hashtable;
 
@@ -48,7 +50,6 @@ class Unbind
 
     public Unbind(boolean output)
     {
-
         // Set up the environment for creating the initial context
         Hashtable env = new Hashtable(11);
         env.put(Context.INITIAL_CONTEXT_FACTORY,
@@ -99,6 +100,14 @@ class Unbind
                 try
                 {
                     obj = ctx.lookup("Connection");
+                    try
+                    {
+                        ((Connection) obj).close();
+                    }
+                    catch (JMSException e)
+                    {
+                        //ignore just need to close
+                    }
                 }
                 catch (NameNotFoundException ne2)
                 {
@@ -125,7 +134,9 @@ class Unbind
             //System.out.println("unbind failed; object still there: " + obj);
 
             // Close the context when we're done
+
             ctx.close();
+
         }
         catch (NamingException e)
         {
diff --git a/java/client/test/src/org/apache/qpid/transacted/TransactedTest.java b/java/client/test/src/org/apache/qpid/transacted/TransactedTest.java
index 3a574ea158..765d3ab168 100644
--- a/java/client/test/src/org/apache/qpid/transacted/TransactedTest.java
+++ b/java/client/test/src/org/apache/qpid/transacted/TransactedTest.java
@@ -54,9 +54,28 @@ public class TransactedTest
     private MessageConsumer testConsumer2;
 
 
+    public void createVMBroker()
+    {
+        try
+        {
+            TransportConnection.createVMBroker(1);
+        }
+        catch (AMQVMBrokerCreationException e)
+        {
+            Assert.fail("Unable to create broker: " + e);
+        }
+    }
+
+    @After
+    public void stopVmBroker()
+    {
+        TransportConnection.killVMBroker(1);
+    }
+
     @Before
     public void setup() throws Exception
     {
+        createVMBroker();
         queue1 = new AMQQueue("Q1", false);
         queue2 = new AMQQueue("Q2", false);
 
@@ -71,20 +90,6 @@ public class TransactedTest
         prepProducer1 = prepSession.createProducer(queue1);
         prepCon.start();
 
-        // Sleep to ensure all queues have been created in the Broker.
-        try
-        {
-            System.out.println("Allowing Server to create queues");
-            Thread.sleep(2000);
-        }
-        catch (InterruptedException e)
-        {
-            //do nothing
-        }
-        finally
-        {
-            System.out.println("Setup Complete");
-        }
 
         //add some messages
         prepProducer1.send(prepSession.createTextMessage("A"));
@@ -96,21 +101,6 @@ public class TransactedTest
         testConsumer1 = testSession.createConsumer(queue1);
         testConsumer2 = testSession.createConsumer(queue2);
         testCon.start();
-
-         // Sleep to ensure all queues have been created in the Broker.
-        try
-        {
-            System.out.println("Allowing Server to create queues");
-            Thread.sleep(2000);
-        }
-        catch (InterruptedException e)
-        {
-            //do nothing
-        }
-        finally
-        {
-            System.out.println("Setup Complete");
-        }
     }
 
     @After