Merged out from trunkQPID-2519

git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/QPID-2519@1187375 13f79535-47bb-0310-9956-ffa450edef68

18 files changed, 1754 insertions, 586 deletions

diff --git a/java/client/src/test/java/org/apache/mina/transport/vmpipe/support/VmPipeIdleStatusChecker.java b/java/client/src/test/java/org/apache/mina/transport/vmpipe/support/VmPipeIdleStatusChecker.java
deleted file mode 100644
index 5323ad28bf..0000000000
--- a/java/client/src/test/java/org/apache/mina/transport/vmpipe/support/VmPipeIdleStatusChecker.java
+++ /dev/null
@@ -1,125 +0,0 @@
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *   http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- *
- */
-package org.apache.mina.transport.vmpipe.support;
-
-import org.apache.mina.common.IdleStatus;
-
-import java.util.HashMap;
-import java.util.IdentityHashMap;
-import java.util.Iterator;
-import java.util.Map;
-
-/**
- * This file is a patch to override MINA, because of the IdentityHashMap bug. Workaround to be supplied in MINA 1.0.7.
- * This patched file will be removed once upgraded onto a newer MINA.
- *
- * Dectects idle sessions and fires <tt>sessionIdle</tt> events to them.
- *
- * @author The Apache Directory Project (mina-dev@directory.apache.org)
- */
-public class VmPipeIdleStatusChecker
-{
-    private static final VmPipeIdleStatusChecker INSTANCE = new VmPipeIdleStatusChecker();
-
-    public static VmPipeIdleStatusChecker getInstance()
-    {
-        return INSTANCE;
-    }
-
-    private final Map sessions = new HashMap(); // will use as a set
-
-    private final Worker worker = new Worker();
-
-    private VmPipeIdleStatusChecker()
-    {
-        worker.start();
-    }
-
-    public void addSession(VmPipeSessionImpl session)
-    {
-        synchronized (sessions)
-        {
-            sessions.put(session, session);
-        }
-    }
-
-    private class Worker extends Thread
-    {
-        private Worker()
-        {
-            super("VmPipeIdleStatusChecker");
-            setDaemon(true);
-        }
-
-        public void run()
-        {
-            for (;;)
-            {
-                try
-                {
-                    Thread.sleep(1000);
-                }
-                catch (InterruptedException e)
-                { }
-
-                long currentTime = System.currentTimeMillis();
-
-                synchronized (sessions)
-                {
-                    Iterator it = sessions.keySet().iterator();
-                    while (it.hasNext())
-                    {
-                        VmPipeSessionImpl session = (VmPipeSessionImpl) it.next();
-                        if (!session.isConnected())
-                        {
-                            it.remove();
-                        }
-                        else
-                        {
-                            notifyIdleSession(session, currentTime);
-                        }
-                    }
-                }
-            }
-        }
-    }
-
-    private void notifyIdleSession(VmPipeSessionImpl session, long currentTime)
-    {
-        notifyIdleSession0(session, currentTime, session.getIdleTimeInMillis(IdleStatus.BOTH_IDLE), IdleStatus.BOTH_IDLE,
-            Math.max(session.getLastIoTime(), session.getLastIdleTime(IdleStatus.BOTH_IDLE)));
-        notifyIdleSession0(session, currentTime, session.getIdleTimeInMillis(IdleStatus.READER_IDLE), IdleStatus.READER_IDLE,
-            Math.max(session.getLastReadTime(), session.getLastIdleTime(IdleStatus.READER_IDLE)));
-        notifyIdleSession0(session, currentTime, session.getIdleTimeInMillis(IdleStatus.WRITER_IDLE), IdleStatus.WRITER_IDLE,
-            Math.max(session.getLastWriteTime(), session.getLastIdleTime(IdleStatus.WRITER_IDLE)));
-    }
-
-    private void notifyIdleSession0(VmPipeSessionImpl session, long currentTime, long idleTime, IdleStatus status,
-        long lastIoTime)
-    {
-        if ((idleTime > 0) && (lastIoTime != 0) && ((currentTime - lastIoTime) >= idleTime))
-        {
-            session.increaseIdleCount(status);
-            session.getFilterChain().fireSessionIdle(session, status);
-        }
-    }
-
-}
diff --git a/java/client/src/test/java/org/apache/qpid/client/AMQSession_0_10Test.java b/java/client/src/test/java/org/apache/qpid/client/AMQSession_0_10Test.java
new file mode 100644
index 0000000000..849827216c
--- /dev/null
+++ b/java/client/src/test/java/org/apache/qpid/client/AMQSession_0_10Test.java
@@ -0,0 +1,765 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.qpid.client;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageListener;
+import javax.jms.MessageProducer;
+
+import junit.framework.TestCase;
+
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.framing.FieldTable;
+import org.apache.qpid.transport.Binary;
+import org.apache.qpid.transport.Connection;
+import org.apache.qpid.transport.Connection.SessionFactory;
+import org.apache.qpid.transport.Connection.State;
+import org.apache.qpid.transport.ExchangeBound;
+import org.apache.qpid.transport.ExchangeBoundResult;
+import org.apache.qpid.transport.ExchangeDeclare;
+import org.apache.qpid.transport.ExchangeDelete;
+import org.apache.qpid.transport.ExchangeQuery;
+import org.apache.qpid.transport.ExchangeQueryResult;
+import org.apache.qpid.transport.ExecutionErrorCode;
+import org.apache.qpid.transport.ExecutionException;
+import org.apache.qpid.transport.ExecutionResult;
+import org.apache.qpid.transport.ExecutionSync;
+import org.apache.qpid.transport.Future;
+import org.apache.qpid.transport.MessageCancel;
+import org.apache.qpid.transport.MessageFlow;
+import org.apache.qpid.transport.MessageRelease;
+import org.apache.qpid.transport.MessageSubscribe;
+import org.apache.qpid.transport.MessageTransfer;
+import org.apache.qpid.transport.Method;
+import org.apache.qpid.transport.Option;
+import org.apache.qpid.transport.ProtocolEvent;
+import org.apache.qpid.transport.QueueDelete;
+import org.apache.qpid.transport.QueueQuery;
+import org.apache.qpid.transport.QueueQueryResult;
+import org.apache.qpid.transport.Sender;
+import org.apache.qpid.transport.Session;
+import org.apache.qpid.transport.SessionAttach;
+import org.apache.qpid.transport.SessionDelegate;
+import org.apache.qpid.transport.SessionDetach;
+import org.apache.qpid.transport.SessionException;
+import org.apache.qpid.transport.SessionRequestTimeout;
+import org.apache.qpid.transport.TxCommit;
+import org.apache.qpid.transport.TxRollback;
+import org.apache.qpid.transport.TxSelect;
+
+/**
+ * Tests AMQSession_0_10 methods.
+ * <p>
+ * The main purpose of the tests in this test suite is to check that
+ * {@link SessionException} is not thrown from methods of
+ * {@link AMQSession_0_10}.
+ */
+public class AMQSession_0_10Test extends TestCase
+{
+
+    public void testExceptionOnCommit()
+    {
+        AMQSession_0_10 session = createThrowingExceptionAMQSession_0_10();
+        try
+        {
+            session.commit();
+            fail("JMSException should be thrown");
+        }
+        catch (Exception e)
+        {
+            assertTrue("JMSException is expected", e instanceof JMSException);
+            assertEquals("541 error code is expected", "541", ((JMSException) e).getErrorCode());
+        }
+    }
+
+    public void testExceptionOnCreateMessageProducer()
+    {
+        AMQSession_0_10 session = createThrowingExceptionAMQSession_0_10();
+        try
+        {
+            session.createMessageProducer(createDestination(), true, true, 1l);
+            fail("JMSException should be thrown");
+        }
+        catch (Exception e)
+        {
+            assertTrue("JMSException is expected but got:" + e, e instanceof JMSException);
+            assertEquals("541 error code is expected", "541", ((JMSException) e).getErrorCode());
+        }
+    }
+
+    public void testExceptionOnRollback()
+    {
+        AMQSession_0_10 session = createThrowingExceptionAMQSession_0_10();
+        try
+        {
+            session.rollback();
+            fail("JMSException should be thrown");
+        }
+        catch (Exception e)
+        {
+            assertTrue("JMSException is expected", e instanceof JMSException);
+        }
+    }
+
+    public void testExceptionOnRecover()
+    {
+        AMQSession_0_10 session = createThrowingExceptionAMQSession_0_10(javax.jms.Session.AUTO_ACKNOWLEDGE);
+        try
+        {
+            session.recover();
+            fail("JMSException should be thrown");
+        }
+        catch (Exception e)
+        {
+            assertTrue("JMSException is expected", e instanceof JMSException);
+        }
+    }
+
+    public void testExceptionOnCreateBrowser()
+    {
+        AMQSession_0_10 session = createThrowingExceptionAMQSession_0_10();
+        AMQQueue destination = createQueue();
+        try
+        {
+            session.createBrowser(destination);
+            fail("JMSException should be thrown");
+        }
+        catch (Exception e)
+        {
+            assertTrue("JMSException is expected", e instanceof JMSException);
+            assertEquals("541 error code is expected", "541", ((JMSException) e).getErrorCode());
+        }
+    }
+
+    public void testExceptionOnCreateConsumer()
+    {
+        AMQSession_0_10 session = createThrowingExceptionAMQSession_0_10();
+        AMQAnyDestination destination = createDestination();
+        try
+        {
+            session.createConsumer(destination);
+            fail("JMSException should be thrown");
+        }
+        catch (Exception e)
+        {
+            assertTrue("JMSException is expected", e instanceof JMSException);
+            assertEquals("541 error code is expected", "541", ((JMSException) e).getErrorCode());
+        }
+    }
+
+    public void testExceptionOnCreateSubscriber()
+    {
+        AMQSession_0_10 session = createThrowingExceptionAMQSession_0_10();
+        AMQAnyDestination destination = createDestination();
+        try
+        {
+            session.createSubscriber(destination);
+            fail("JMSException should be thrown");
+        }
+        catch (Exception e)
+        {
+            assertTrue("JMSException is expected", e instanceof JMSException);
+            assertEquals("541 error code is expected", "541", ((JMSException) e).getErrorCode());
+        }
+    }
+
+    public void testExceptionOnUnsubscribe()
+    {
+        AMQSession_0_10 session = createThrowingExceptionAMQSession_0_10();
+        try
+        {
+            session.unsubscribe("whatever");
+            fail("JMSExceptiuon should be thrown");
+        }
+        catch (Exception e)
+        {
+            assertTrue("JMSException is expected", e instanceof JMSException);
+            assertEquals("541 error code is expected", "541", ((JMSException) e).getErrorCode());
+        }
+    }
+
+    public void testCommit()
+    {
+        AMQSession_0_10 session = createAMQSession_0_10();
+        try
+        {
+            session.commit();
+        }
+        catch (Exception e)
+        {
+            fail("Unexpected exception is cought:" + e.getMessage());
+        }
+        ProtocolEvent event = findSentProtocolEventOfClass(session, TxCommit.class, false);
+        assertNotNull("TxCommit was not sent", event);
+    }
+
+    public void testRollback()
+    {
+        AMQSession_0_10 session = createAMQSession_0_10();
+        try
+        {
+            session.rollback();
+        }
+        catch (Exception e)
+        {
+            fail("Unexpected exception is cought:" + e.getMessage());
+        }
+        ProtocolEvent event = findSentProtocolEventOfClass(session, TxRollback.class, false);
+        assertNotNull("TxRollback was not sent", event);
+    }
+
+    public void testRecover()
+    {
+        AMQSession_0_10 session = createAMQSession_0_10(javax.jms.Session.AUTO_ACKNOWLEDGE);
+        try
+        {
+            session.recover();
+        }
+        catch (Exception e)
+        {
+            fail("Unexpected exception is cought:" + e.getMessage());
+        }
+        ProtocolEvent event = findSentProtocolEventOfClass(session, MessageRelease.class, false);
+        assertNotNull("MessageRelease was not sent", event);
+    }
+
+    public void testCreateProducer()
+    {
+        AMQSession_0_10 session = createAMQSession_0_10();
+        try
+        {
+            session.createProducer(createQueue());
+        }
+        catch (Exception e)
+        {
+            fail("Unexpected exception is cought:" + e.getMessage());
+        }
+        ProtocolEvent event = findSentProtocolEventOfClass(session, ExchangeDeclare.class, false);
+        assertNotNull("ExchangeDeclare was not sent", event);
+    }
+
+    public void testCreateConsumer()
+    {
+        AMQSession_0_10 session = createAMQSession_0_10();
+        try
+        {
+            session.createConsumer(createQueue());
+        }
+        catch (Exception e)
+        {
+            fail("Unexpected exception is cought:" + e.getMessage());
+        }
+        ProtocolEvent event = findSentProtocolEventOfClass(session, MessageSubscribe.class, false);
+        assertNotNull("MessageSubscribe was not sent", event);
+    }
+
+    public void testSync()
+    {
+        AMQSession_0_10 session = createAMQSession_0_10();
+        try
+        {
+            session.sync();
+        }
+        catch (Exception e)
+        {
+            fail("Unexpected exception is cought:" + e.getMessage());
+        }
+        ProtocolEvent event = findSentProtocolEventOfClass(session, ExecutionSync.class, false);
+        assertNotNull("ExecutionSync was not sent", event);
+    }
+
+    public void testRejectMessage()
+    {
+        AMQSession_0_10 session = createAMQSession_0_10();
+        session.rejectMessage(1l, true);
+        ProtocolEvent event = findSentProtocolEventOfClass(session, MessageRelease.class, false);
+        assertNotNull("MessageRelease event was not sent", event);
+    }
+
+    public void testReleaseForRollback()
+    {
+        AMQSession_0_10 session = createAMQSession_0_10();
+        try
+        {
+            session.releaseForRollback();
+        }
+        catch (Exception e)
+        {
+            fail("Unexpected exception is cought:" + e.getMessage());
+        }
+        ProtocolEvent event = findSentProtocolEventOfClass(session, MessageRelease.class, false);
+        assertNotNull("MessageRelease event was not sent", event);
+    }
+
+    public void testSendQueueDelete()
+    {
+        AMQSession_0_10 session = createAMQSession_0_10();
+        try
+        {
+            session.sendQueueDelete(new AMQShortString("test"));
+        }
+        catch (Exception e)
+        {
+            fail("Unexpected exception is cought:" + e.getMessage());
+        }
+        ProtocolEvent event = findSentProtocolEventOfClass(session, QueueDelete.class, false);
+        assertNotNull("QueueDelete event was not sent", event);
+        QueueDelete exchangeDelete = (QueueDelete) event;
+        assertEquals("test", exchangeDelete.getQueue());
+    }
+
+    public void testSendConsume()
+    {
+        AMQSession_0_10 session = createAMQSession_0_10();
+        try
+        {
+            BasicMessageConsumer_0_10 consumer = session.createMessageConsumer(createDestination(), 1, 1, true, false,
+                    null, new FieldTable(), false, true);
+            session.sendConsume(consumer, new AMQShortString("test"), null, true, null, 1);
+        }
+        catch (Exception e)
+        {
+            fail("Unexpected exception is cought:" + e.getMessage());
+        }
+        ProtocolEvent event = findSentProtocolEventOfClass(session, MessageSubscribe.class, false);
+        assertNotNull("MessageSubscribe event was not sent", event);
+    }
+
+    public void testCreateMessageProducer()
+    {
+        AMQSession_0_10 session = createAMQSession_0_10();
+        try
+        {
+            session.createMessageProducer(createDestination(), true, true, 1l);
+        }
+        catch (Exception e)
+        {
+            fail("Unexpected exception is cought:" + e.getMessage());
+        }
+        ProtocolEvent event = findSentProtocolEventOfClass(session, ExchangeDeclare.class, false);
+        assertNotNull("ExchangeDeclare event was not sent", event);
+    }
+
+    public void testSendExchangeDelete()
+    {
+        AMQSession_0_10 session = createAMQSession_0_10();
+        try
+        {
+            session.sendExchangeDelete("test", true);
+        }
+        catch (Exception e)
+        {
+            fail("Unexpected exception is cought:" + e.getMessage());
+        }
+        ProtocolEvent event = findSentProtocolEventOfClass(session, ExchangeDelete.class, false);
+        assertNotNull("ExchangeDelete event was not sent", event);
+        ExchangeDelete exchangeDelete = (ExchangeDelete) event;
+        assertEquals("test", exchangeDelete.getExchange());
+    }
+
+    public void testExceptionOnMessageConsumerReceive()
+    {
+        AMQSession_0_10 session = createThrowingExceptionAMQSession_0_10();
+        try
+        {
+            BasicMessageConsumer_0_10 consumer = session.createMessageConsumer(createDestination(), 1, 1, true, false,
+                    null, new FieldTable(), false, true);
+            session.start();
+            consumer.receive(1);
+            fail("JMSException should be thrown");
+        }
+        catch (Exception e)
+        {
+            assertTrue("JMSException is expected", e instanceof JMSException);
+            assertEquals("541 error code is expected", "541", ((JMSException) e).getErrorCode());
+        }
+    }
+
+    public void testMessageConsumerReceive()
+    {
+        AMQSession_0_10 session = createAMQSession_0_10();
+        try
+        {
+            BasicMessageConsumer_0_10 consumer = session.createMessageConsumer(createDestination(), 1, 1, true, false,
+                    null, new FieldTable(), false, true);
+            session.start();
+            consumer.receive(1);
+        }
+        catch (Exception e)
+        {
+            fail("Unexpected exception is cought:" + e.getMessage());
+        }
+        ProtocolEvent event = findSentProtocolEventOfClass(session, MessageFlow.class, false);
+        assertNotNull("MessageFlow event was not sent", event);
+    }
+
+    public void testExceptionOnMessageConsumerReceiveNoWait()
+    {
+        AMQSession_0_10 session = createThrowingExceptionAMQSession_0_10();
+        try
+        {
+            BasicMessageConsumer_0_10 consumer = session.createMessageConsumer(createDestination(), 1, 1, true, false,
+                    null, new FieldTable(), false, true);
+            session.start();
+            consumer.receiveNoWait();
+            fail("JMSException should be thrown");
+        }
+        catch (Exception e)
+        {
+            assertTrue("JMSException is expected", e instanceof JMSException);
+            assertEquals("541 error code is expected", "541", ((JMSException) e).getErrorCode());
+        }
+    }
+
+    public void testExceptionOnMessageConsumerSetMessageListener()
+    {
+        AMQSession_0_10 session = createThrowingExceptionAMQSession_0_10();
+        try
+        {
+            BasicMessageConsumer_0_10 consumer = session.createMessageConsumer(createDestination(), 1, 1, true, false,
+                    null, new FieldTable(), false, true);
+            consumer.setMessageListener(new MockMessageListener());
+            fail("JMSException should be thrown");
+        }
+        catch (Exception e)
+        {
+            assertTrue("JMSException is expected", e instanceof JMSException);
+            assertEquals("541 error code is expected", "541", ((JMSException) e).getErrorCode());
+        }
+    }
+
+    public void testMessageConsumerSetMessageListener()
+    {
+        AMQSession_0_10 session = createAMQSession_0_10();
+        try
+        {
+            BasicMessageConsumer_0_10 consumer = session.createMessageConsumer(createDestination(), 1, 1, true, false,
+                    null, new FieldTable(), false, true);
+            consumer.setMessageListener(new MockMessageListener());
+        }
+        catch (Exception e)
+        {
+            fail("Unexpected exception is cought:" + e.getMessage());
+        }
+        ProtocolEvent event = findSentProtocolEventOfClass(session, MessageFlow.class, false);
+        assertNotNull("MessageFlow event was not sent", event);
+    }
+
+    public void testMessageConsumerClose()
+    {
+        AMQSession_0_10 session = createAMQSession_0_10();
+        try
+        {
+            BasicMessageConsumer_0_10 consumer = session.createMessageConsumer(createDestination(), 1, 1, true, false,
+                    null, new FieldTable(), false, true);
+            consumer.close();
+        }
+        catch (Exception e)
+        {
+            fail("Unexpected exception is cought:" + e.getMessage());
+        }
+        ProtocolEvent event = findSentProtocolEventOfClass(session, MessageCancel.class, false);
+        assertNotNull("MessageCancel event was not sent", event);
+    }
+
+    public void testExceptionOnMessageConsumerClose()
+    {
+        AMQSession_0_10 session = createThrowingExceptionAMQSession_0_10();
+        try
+        {
+            BasicMessageConsumer_0_10 consumer = session.createMessageConsumer(createDestination(), 1, 1, true, false,
+                    null, new FieldTable(), false, true);
+            consumer.close();
+            fail("JMSException should be thrown");
+        }
+        catch (Exception e)
+        {
+            assertTrue("JMSException is expected", e instanceof JMSException);
+            assertEquals("541 error code is expected", "541", ((JMSException) e).getErrorCode());
+        }
+    }
+
+    public void testMessageProducerSend()
+    {
+        AMQSession_0_10 session = createAMQSession_0_10();
+        try
+        {
+            MessageProducer producer = session.createProducer(createQueue());
+            producer.send(session.createTextMessage("Test"));
+            session.commit();
+        }
+        catch (Exception e)
+        {
+            fail("Unexpected exception is cought:" + e.getMessage());
+        }
+        ProtocolEvent event = findSentProtocolEventOfClass(session, MessageTransfer.class, false);
+        assertNotNull("MessageTransfer event was not sent", event);
+        event = findSentProtocolEventOfClass(session, ExchangeDeclare.class, false);
+        assertNotNull("ExchangeDeclare event was not sent", event);
+    }
+
+    private AMQAnyDestination createDestination()
+    {
+        AMQAnyDestination destination = null;
+        try
+        {
+            destination = new AMQAnyDestination(new AMQShortString("amq.direct"), new AMQShortString("direct"),
+                    new AMQShortString("test"), false, true, new AMQShortString("test"), true, null);
+        }
+        catch (Exception e)
+        {
+            fail("Failued to create destination:" + e.getMessage());
+        }
+        return destination;
+    }
+
+    private AMQQueue createQueue()
+    {
+        AMQQueue destination = null;
+        try
+        {
+            destination = new AMQQueue(new AMQShortString("amq.direct"), new AMQShortString("test"),
+                    new AMQShortString("test"));
+        }
+        catch (Exception e)
+        {
+            fail("Failued to create destination:" + e.getMessage());
+        }
+        return destination;
+    }
+
+    private AMQSession_0_10 createThrowingExceptionAMQSession_0_10()
+    {
+        return createAMQSession_0_10(true, javax.jms.Session.SESSION_TRANSACTED);
+    }
+
+    private AMQSession_0_10 createThrowingExceptionAMQSession_0_10(int akcnowledgeMode)
+    {
+        return createAMQSession_0_10(true, akcnowledgeMode);
+    }
+
+    private ProtocolEvent findSentProtocolEventOfClass(AMQSession_0_10 session, Class<? extends ProtocolEvent> class1,
+            boolean isLast)
+    {
+        ProtocolEvent found = null;
+        List<ProtocolEvent> events = ((MockSession) session.getQpidSession()).getSender().getSendEvents();
+        assertNotNull("Events list should not be null", events);
+        assertFalse("Events list should not be empty", events.isEmpty());
+        if (isLast)
+        {
+            ProtocolEvent event = events.get(events.size() - 1);
+            if (event.getClass().isAssignableFrom(class1))
+            {
+                found = event;
+            }
+        }
+        else
+        {
+            for (ProtocolEvent protocolEvent : events)
+            {
+                if (protocolEvent.getClass().isAssignableFrom(class1))
+                {
+                    found = protocolEvent;
+                    break;
+                }
+            }
+
+        }
+        return found;
+    }
+
+    private AMQSession_0_10 createAMQSession_0_10()
+    {
+        return createAMQSession_0_10(false, javax.jms.Session.SESSION_TRANSACTED);
+    }
+
+    private AMQSession_0_10 createAMQSession_0_10(int acknowledgeMode)
+    {
+        return createAMQSession_0_10(false, acknowledgeMode);
+    }
+
+    private AMQSession_0_10 createAMQSession_0_10(boolean throwException, int acknowledgeMode)
+    {
+        AMQConnection amqConnection = null;
+        try
+        {
+            amqConnection = new MockAMQConnection(
+                    "amqp://guest:guest@client/test?brokerlist='tcp://localhost:1'&maxprefetch='0'");
+        }
+        catch (Exception e)
+        {
+            fail("Failure to create a mock connection:" + e.getMessage());
+        }
+        boolean isTransacted = acknowledgeMode == javax.jms.Session.SESSION_TRANSACTED ? true : false;
+        AMQSession_0_10 session = new AMQSession_0_10(createConnection(throwException), amqConnection, 1, isTransacted, acknowledgeMode,
+                 1, 1, "test");
+        return session;
+    }
+
+    private Connection createConnection(final boolean throwException)
+    {
+        MockTransportConnection connection = new MockTransportConnection();
+        connection.setState(State.OPEN);
+        connection.setSender(new MockSender());
+        connection.setSessionFactory(new SessionFactory()
+        {
+
+            @Override
+            public Session newSession(Connection conn, Binary name, long expiry)
+            {
+                return new MockSession(conn, new SessionDelegate(), name, expiry, throwException);
+            }
+        });
+        return connection;
+    }
+
+    private final class MockMessageListener implements MessageListener
+    {
+        @Override
+        public void onMessage(Message arg0)
+        {
+        }
+    }
+
+    class MockSession extends Session
+    {
+        private final boolean _throwException;
+        private final Connection _connection;
+        private final SessionDelegate _delegate;
+
+        protected MockSession(Connection connection, SessionDelegate delegate, Binary name, long expiry,
+                boolean throwException)
+        {
+            super(connection, delegate, name, expiry);
+            _throwException = throwException;
+            setState(State.OPEN);
+            _connection = connection;
+            _delegate = delegate;
+        }
+
+        public void invoke(Method m, Runnable postIdSettingAction)
+        {
+            if (_throwException)
+            {
+                if (m instanceof SessionAttach || m instanceof SessionRequestTimeout || m instanceof TxSelect)
+                {
+                    // do not throw exception for SessionAttach,
+                    // SessionRequestTimeout and TxSelect
+                    // session needs to be instantiated
+                    return;
+                }
+                ExecutionException e = new ExecutionException();
+                e.setErrorCode(ExecutionErrorCode.INTERNAL_ERROR);
+                throw new SessionException(e);
+            }
+            else
+            {
+                super.invoke(m, postIdSettingAction);
+                if (m instanceof SessionDetach)
+                {
+                    setState(State.CLOSED);
+                }
+            }
+        }
+
+        public void sync()
+        {
+            // to avoid recursive calls
+            setAutoSync(false);
+            // simply send sync command
+            super.executionSync(Option.SYNC);
+        }
+
+        protected <T> Future<T> invoke(Method m, Class<T> klass)
+        {
+            int commandId = getCommandsOut();
+            Future<T> future = super.invoke(m, klass);
+            ExecutionResult result = new ExecutionResult();
+            result.setCommandId(commandId);
+            if (m instanceof ExchangeBound)
+            {
+                ExchangeBoundResult struc = new ExchangeBoundResult();
+                struc.setQueueNotFound(true);
+                result.setValue(struc);
+            }
+            else if (m instanceof ExchangeQuery)
+            {
+                ExchangeQueryResult struc = new ExchangeQueryResult();
+                result.setValue(struc);
+            }
+            else if (m instanceof QueueQuery)
+            {
+                QueueQueryResult struc = new QueueQueryResult();
+                result.setValue(struc);
+            }
+            _delegate.executionResult(this, result);
+            return future;
+        }
+
+        public MockSender getSender()
+        {
+            return (MockSender) _connection.getSender();
+        }
+    }
+
+    class MockTransportConnection extends Connection
+    {
+        public void setState(State state)
+        {
+            super.setState(state);
+        }
+    }
+
+    class MockSender implements Sender<ProtocolEvent>
+    {
+        private List<ProtocolEvent> _sendEvents = new ArrayList<ProtocolEvent>();
+
+        @Override
+        public void setIdleTimeout(int i)
+        {
+        }
+
+        @Override
+        public void send(ProtocolEvent msg)
+        {
+            _sendEvents.add(msg);
+        }
+
+        @Override
+        public void flush()
+        {
+        }
+
+        @Override
+        public void close()
+        {
+        }
+
+        public List<ProtocolEvent> getSendEvents()
+        {
+            return _sendEvents;
+        }
+
+    }
+
+}
diff --git a/java/client/src/test/java/org/apache/qpid/client/MockAMQConnection.java b/java/client/src/test/java/org/apache/qpid/client/MockAMQConnection.java
index da44822ec3..73e67469ae 100644
--- a/java/client/src/test/java/org/apache/qpid/client/MockAMQConnection.java
+++ b/java/client/src/test/java/org/apache/qpid/client/MockAMQConnection.java
@@ -23,7 +23,6 @@ package org.apache.qpid.client;
 import org.apache.qpid.AMQException;
 import org.apache.qpid.client.state.AMQState;
 import org.apache.qpid.framing.ProtocolVersion;
-import org.apache.qpid.jms.ConnectionURL;
 import org.apache.qpid.jms.BrokerDetails;
 import org.apache.qpid.url.URLSyntaxException;
 
@@ -37,53 +36,18 @@ public class MockAMQConnection extends AMQConnection
         super(broker, username, password, clientName, virtualHost);
     }
 
-    public MockAMQConnection(String broker, String username, String password, String clientName, String virtualHost, SSLConfiguration sslConfig)
-            throws AMQException, URLSyntaxException
-    {
-        super(broker, username, password, clientName, virtualHost, sslConfig);
-    }
-
     public MockAMQConnection(String host, int port, String username, String password, String clientName, String virtualHost)
             throws AMQException, URLSyntaxException
     {
         super(host, port, username, password, clientName, virtualHost);
     }
 
-    public MockAMQConnection(String host, int port, String username, String password, String clientName, String virtualHost, SSLConfiguration sslConfig)
-            throws AMQException, URLSyntaxException
-    {
-        super(host, port, username, password, clientName, virtualHost, sslConfig);
-    }
-
-    public MockAMQConnection(String host, int port, boolean useSSL, String username, String password, String clientName, String virtualHost, SSLConfiguration sslConfig)
-            throws AMQException, URLSyntaxException
-    {
-        super(host, port, useSSL, username, password, clientName, virtualHost, sslConfig);
-    }
-
     public MockAMQConnection(String connection)
             throws AMQException, URLSyntaxException
     {
         super(connection);
     }
 
-    public MockAMQConnection(String connection, SSLConfiguration sslConfig)
-            throws AMQException, URLSyntaxException
-    {
-        super(connection, sslConfig);
-    }
-
-    public MockAMQConnection(ConnectionURL connectionURL, SSLConfiguration sslConfig)
-            throws AMQException
-    {
-        super(connectionURL, sslConfig);
-    }
-
-    protected MockAMQConnection(String username, String password, String clientName, String virtualHost)
-    {
-        super(username, password, clientName, virtualHost);
-    }
-
     @Override
     public ProtocolVersion makeBrokerConnection(BrokerDetails brokerDetail) throws IOException
     {
diff --git a/java/client/src/test/java/org/apache/qpid/client/message/TestMessageHelper.java b/java/client/src/test/java/org/apache/qpid/client/message/TestMessageHelper.java
index 7ee991b63c..b5c31e7c5e 100644
--- a/java/client/src/test/java/org/apache/qpid/client/message/TestMessageHelper.java
+++ b/java/client/src/test/java/org/apache/qpid/client/message/TestMessageHelper.java
@@ -43,4 +43,9 @@ public class TestMessageHelper
     {
         return new JMSStreamMessage(AMQMessageDelegateFactory.FACTORY_0_8);
     }
+
+    public static JMSObjectMessage newJMSObjectMessage()
+    {
+        return new JMSObjectMessage(AMQMessageDelegateFactory.FACTORY_0_8);
+    }
 }
diff --git a/java/client/src/test/java/org/apache/qpid/client/protocol/AMQProtocolHandlerTest.java b/java/client/src/test/java/org/apache/qpid/client/protocol/AMQProtocolHandlerTest.java
index f520a21ba0..e159ceb148 100644
--- a/java/client/src/test/java/org/apache/qpid/client/protocol/AMQProtocolHandlerTest.java
+++ b/java/client/src/test/java/org/apache/qpid/client/protocol/AMQProtocolHandlerTest.java
@@ -20,23 +20,24 @@
  */
 package org.apache.qpid.client.protocol;
 
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
 import junit.framework.TestCase;
-import org.apache.qpid.framing.AMQFrame;
+
+import org.apache.qpid.AMQException;
+import org.apache.qpid.client.AMQAuthenticationException;
+import org.apache.qpid.client.MockAMQConnection;
+import org.apache.qpid.client.state.AMQState;
 import org.apache.qpid.framing.AMQBody;
+import org.apache.qpid.framing.AMQFrame;
 import org.apache.qpid.framing.AMQMethodBody;
 import org.apache.qpid.framing.amqp_8_0.BasicRecoverOkBodyImpl;
-import org.apache.qpid.AMQException;
 import org.apache.qpid.protocol.AMQConstant;
-import org.apache.qpid.transport.TestNetworkDriver;
-import org.apache.qpid.client.MockAMQConnection;
-import org.apache.qpid.client.AMQAuthenticationException;
-import org.apache.qpid.client.state.AMQState;
+import org.apache.qpid.transport.TestNetworkConnection;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-
 /**
  * This is a test address QPID-1431 where frame listeners would fail to be notified of an incomming exception.
  *
@@ -72,8 +73,8 @@ public class AMQProtocolHandlerTest extends TestCase
     public void setUp() throws Exception
     {
         //Create a new ProtocolHandler with a fake connection.
-        _handler = new AMQProtocolHandler(new MockAMQConnection("amqp://guest:guest@client/test?brokerlist='vm://:1'"));
-        _handler.setNetworkDriver(new TestNetworkDriver());
+        _handler = new AMQProtocolHandler(new MockAMQConnection("amqp://guest:guest@client/test?brokerlist='tcp://localhost:1'"));
+        _handler.setNetworkConnection(new TestNetworkConnection());
          AMQBody body = BasicRecoverOkBodyImpl.getFactory().newInstance(null, 1);
         _blockFrame = new AMQFrame(0, body);
 
diff --git a/java/client/src/test/java/org/apache/qpid/client/protocol/MockIoSession.java b/java/client/src/test/java/org/apache/qpid/client/protocol/MockIoSession.java
deleted file mode 100644
index f0938a4bc0..0000000000
--- a/java/client/src/test/java/org/apache/qpid/client/protocol/MockIoSession.java
+++ /dev/null
@@ -1,312 +0,0 @@
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *   http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- *
- */
-package org.apache.qpid.client.protocol;
-
-import org.apache.mina.common.*;
-import org.apache.mina.common.support.DefaultCloseFuture;
-import org.apache.mina.common.support.DefaultWriteFuture;
-import org.apache.mina.common.support.AbstractIoFilterChain;
-import org.apache.qpid.client.protocol.AMQProtocolSession;
-
-import java.net.SocketAddress;
-import java.net.InetSocketAddress;
-import java.util.Set;
-
-public class MockIoSession implements IoSession
-{
-    private AMQProtocolSession _protocolSession;
-
-    /**
-     * Stores the last response written
-     */
-    private Object _lastWrittenObject;
-
-    private boolean _closing;
-    private IoFilterChain _filterChain;
-
-    public MockIoSession()
-    {
-        _filterChain = new AbstractIoFilterChain(this)
-        {
-            protected void doWrite(IoSession ioSession, IoFilter.WriteRequest writeRequest) throws Exception
-            {
-
-            }
-
-            protected void doClose(IoSession ioSession) throws Exception
-            {
-
-            }
-        };
-    }
-
-    public Object getLastWrittenObject()
-    {
-        return _lastWrittenObject;
-    }
-
-    public IoService getService()
-    {
-        return null;  //To change body of implemented methods use File | Settings | File Templates.
-    }
-
-    public IoServiceConfig getServiceConfig()
-    {
-        return null;
-    }
-
-    public IoHandler getHandler()
-    {
-        return null;  //To change body of implemented methods use File | Settings | File Templates.
-    }
-
-    public IoSessionConfig getConfig()
-    {
-        return null;  //To change body of implemented methods use File | Settings | File Templates.
-    }
-
-    public IoFilterChain getFilterChain()
-    {
-        return _filterChain;
-    }
-
-    public WriteFuture write(Object message)
-    {
-        WriteFuture wf = new DefaultWriteFuture(null);
-        _lastWrittenObject = message;
-        return wf;
-    }
-
-    public CloseFuture close()
-    {
-        _closing = true;
-        CloseFuture cf = new DefaultCloseFuture(null);
-        cf.setClosed();
-        return cf;
-    }
-
-    public Object getAttachment()
-    {
-        return _protocolSession;
-    }
-
-    public Object setAttachment(Object attachment)
-    {
-        Object current = _protocolSession;
-        _protocolSession = (AMQProtocolSession) attachment;
-        return current;
-    }
-
-    public Object getAttribute(String key)
-    {
-        return null;  //To change body of implemented methods use File | Settings | File Templates.
-    }
-
-    public Object setAttribute(String key, Object value)
-    {
-        return null;  //To change body of implemented methods use File | Settings | File Templates.
-    }
-
-    public Object setAttribute(String key)
-    {
-        return null;  //To change body of implemented methods use File | Settings | File Templates.
-    }
-
-    public Object removeAttribute(String key)
-    {
-        return null;  //To change body of implemented methods use File | Settings | File Templates.
-    }
-
-    public boolean containsAttribute(String key)
-    {
-        return false;  //To change body of implemented methods use File | Settings | File Templates.
-    }
-
-    public Set getAttributeKeys()
-    {
-        return null;  //To change body of implemented methods use File | Settings | File Templates.
-    }
-
-    public TransportType getTransportType()
-    {
-        return null;  //To change body of implemented methods use File | Settings | File Templates.
-    }
-
-    public boolean isConnected()
-    {
-        return false;  //To change body of implemented methods use File | Settings | File Templates.
-    }
-
-    public boolean isClosing()
-    {
-        return _closing;
-    }
-
-    public CloseFuture getCloseFuture()
-    {
-        return null;  //To change body of implemented methods use File | Settings | File Templates.
-    }
-
-    public SocketAddress getRemoteAddress()
-    {
-        return new InetSocketAddress("127.0.0.1", 1234);  //To change body of implemented methods use File | Settings | File Templates.
-    }
-
-    public SocketAddress getLocalAddress()
-    {
-        return null;  //To change body of implemented methods use File | Settings | File Templates.
-    }
-
-    public SocketAddress getServiceAddress()
-    {
-        return null;  //To change body of implemented methods use File | Settings | File Templates.
-    }
-
-    public int getIdleTime(IdleStatus status)
-    {
-        return 0;  //To change body of implemented methods use File | Settings | File Templates.
-    }
-
-    public long getIdleTimeInMillis(IdleStatus status)
-    {
-        return 0;  //To change body of implemented methods use File | Settings | File Templates.
-    }
-
-    public void setIdleTime(IdleStatus status, int idleTime)
-    {
-        //To change body of implemented methods use File | Settings | File Templates.
-    }
-
-    public int getWriteTimeout()
-    {
-        return 0;  //To change body of implemented methods use File | Settings | File Templates.
-    }
-
-    public long getWriteTimeoutInMillis()
-    {
-        return 0;  //To change body of implemented methods use File | Settings | File Templates.
-    }
-
-    public void setWriteTimeout(int writeTimeout)
-    {
-        //To change body of implemented methods use File | Settings | File Templates.
-    }
-
-    public TrafficMask getTrafficMask()
-    {
-        return null;  //To change body of implemented methods use File | Settings | File Templates.
-    }
-
-    public void setTrafficMask(TrafficMask trafficMask)
-    {
-        //To change body of implemented methods use File | Settings | File Templates.
-    }
-
-    public void suspendRead()
-    {
-        //To change body of implemented methods use File | Settings | File Templates.
-    }
-
-    public void suspendWrite()
-    {
-        //To change body of implemented methods use File | Settings | File Templates.
-    }
-
-    public void resumeRead()
-    {
-        //To change body of implemented methods use File | Settings | File Templates.
-    }
-
-    public void resumeWrite()
-    {
-        //To change body of implemented methods use File | Settings | File Templates.
-    }
-
-    public long getReadBytes()
-    {
-        return 0;  //To change body of implemented methods use File | Settings | File Templates.
-    }
-
-    public long getWrittenBytes()
-    {
-        return 0;  //To change body of implemented methods use File | Settings | File Templates.
-    }
-
-    public long getReadMessages()
-    {
-        return 0L;
-    }
-
-    public long getWrittenMessages()
-    {
-        return 0L;
-    }
-
-    public long getWrittenWriteRequests()
-    {
-        return 0;  //To change body of implemented methods use File | Settings | File Templates.
-    }
-
-    public int getScheduledWriteRequests()
-    {
-        return 0;  //To change body of implemented methods use File | Settings | File Templates.
-    }
-
-    public int getScheduledWriteBytes()
-    {
-        return 0;  //TODO
-    }
-
-    public long getCreationTime()
-    {
-        return 0;  //To change body of implemented methods use File | Settings | File Templates.
-    }
-
-    public long getLastIoTime()
-    {
-        return 0;  //To change body of implemented methods use File | Settings | File Templates.
-    }
-
-    public long getLastReadTime()
-    {
-        return 0;  //To change body of implemented methods use File | Settings | File Templates.
-    }
-
-    public long getLastWriteTime()
-    {
-        return 0;  //To change body of implemented methods use File | Settings | File Templates.
-    }
-
-    public boolean isIdle(IdleStatus status)
-    {
-        return false;  //To change body of implemented methods use File | Settings | File Templates.
-    }
-
-    public int getIdleCount(IdleStatus status)
-    {
-        return 0;  //To change body of implemented methods use File | Settings | File Templates.
-    }
-
-    public long getLastIdleTime(IdleStatus status)
-    {
-        return 0;  //To change body of implemented methods use File | Settings | File Templates.
-    }
-}
diff --git a/java/client/src/test/java/org/apache/qpid/client/security/CallbackHandlerRegistryTest.java b/java/client/src/test/java/org/apache/qpid/client/security/CallbackHandlerRegistryTest.java
new file mode 100644
index 0000000000..cc5d48fbef
--- /dev/null
+++ b/java/client/src/test/java/org/apache/qpid/client/security/CallbackHandlerRegistryTest.java
@@ -0,0 +1,185 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.client.security;
+
+import java.io.IOException;
+import java.util.Properties;
+
+import javax.security.auth.callback.Callback;
+import javax.security.auth.callback.CallbackHandler;
+import javax.security.auth.callback.UnsupportedCallbackException;
+
+import org.apache.qpid.jms.ConnectionURL;
+import org.apache.qpid.test.utils.QpidTestCase;
+
+
+/**
+ * Tests the ability of {@link CallbackHandlerRegistry} to correctly parse
+ * the properties describing the available callback handlers.   Ensures also
+ * that it is able to select the mechanism and create an implementation
+ * given a variety of starting conditions.
+ *
+ */
+public class CallbackHandlerRegistryTest extends QpidTestCase
+{
+    private CallbackHandlerRegistry _registry;  // Object under test
+
+    public void testCreateHandlerSuccess()
+    {
+        final Properties props = new Properties();
+        props.put("TESTA.1", TestACallbackHandler.class.getName());
+
+        _registry = new CallbackHandlerRegistry(props);
+        assertEquals(1,_registry.getMechanisms().size());
+
+        final CallbackHandler handler = _registry.createCallbackHandler("TESTA");
+        assertTrue(handler instanceof TestACallbackHandler);
+    }
+
+    public void testCreateHandlerForUnknownMechanismName()
+    {
+        final Properties props = new Properties();
+        props.put("TEST1.1", TestACallbackHandler.class.getName());
+
+        _registry = new CallbackHandlerRegistry(props);
+
+        try
+        {
+            _registry.createCallbackHandler("NOTFOUND");
+            fail("Exception not thrown");
+        }
+        catch (IllegalArgumentException iae)
+        {
+            // PASS
+        }
+    }
+
+    public void testSelectMechanism()
+    {
+        final Properties props = new Properties();
+        props.put("TESTA.1", TestACallbackHandler.class.getName());
+        props.put("TESTB.2", TestBCallbackHandler.class.getName());
+
+        _registry = new CallbackHandlerRegistry(props);
+        assertEquals(2,_registry.getMechanisms().size());
+
+        final String selectedMechanism = _registry.selectMechanism("TESTA");
+        assertEquals("TESTA", selectedMechanism);
+    }
+
+    public void testSelectReturnsFirstMutallyAvailableMechanism()
+    {
+        final Properties props = new Properties();
+        props.put("TESTA.1", TestACallbackHandler.class.getName());
+        props.put("TESTB.2", TestBCallbackHandler.class.getName());
+
+        _registry = new CallbackHandlerRegistry(props);
+
+        final String selectedMechanism = _registry.selectMechanism("TESTD TESTB TESTA");
+        // TESTA should be returned as it is higher than TESTB in the properties file.
+        assertEquals("Selected mechanism should respect the ordinal", "TESTA", selectedMechanism);
+    }
+
+    public void testRestrictedSelectReturnsMechanismFromRestrictedList()
+    {
+        final Properties props = new Properties();
+        props.put("TESTA.1", TestACallbackHandler.class.getName());
+        props.put("TESTB.2", TestBCallbackHandler.class.getName());
+        props.put("TESTC.3", TestCCallbackHandler.class.getName());
+
+        _registry = new CallbackHandlerRegistry(props);
+
+        final String selectedMechanism = _registry.selectMechanism("TESTC TESTB TESTA", "TESTB TESTC");
+        // TESTB should be returned as client has restricted the mechanism list to TESTB and TESTC
+        assertEquals("Selected mechanism should respect the ordinal and be limitted by restricted list","TESTB", selectedMechanism);
+    }
+
+    public void testOldPropertyFormatRejected()
+    {
+        final Properties props = new Properties();
+        props.put("CallbackHandler.TESTA", TestACallbackHandler.class.getName());
+
+        try
+        {
+            new CallbackHandlerRegistry(props);
+            fail("exception not thrown");
+        }
+        catch(IllegalArgumentException iae)
+        {
+            // PASS
+        }
+    }
+
+    public void testPropertyWithNonnumericalOrdinal()
+    {
+        final Properties props = new Properties();
+        props.put("TESTA.z", TestACallbackHandler.class.getName());
+        try
+        {
+            new CallbackHandlerRegistry(props);
+            fail("exception not thrown");
+        }
+        catch(IllegalArgumentException iae)
+        {
+            // PASS
+        }
+    }
+
+    public void testUnexpectedCallbackImplementationsIgnored()
+    {
+        final Properties props = new Properties();
+        props.put("TESTA.1", TestACallbackHandler.class.getName());
+        props.put("TESTB.2", "NotFound");
+        props.put("TESTC.3", "java.lang.String");
+
+        _registry = new CallbackHandlerRegistry(props);
+
+        assertEquals(1,_registry.getMechanisms().size());
+    }
+
+    static class TestACallbackHandler extends TestCallbackHandler
+    {
+    }
+
+    static class TestBCallbackHandler extends TestCallbackHandler
+    {
+    }
+
+    static class TestCCallbackHandler extends TestCallbackHandler
+    {
+    }
+
+    static abstract class TestCallbackHandler implements AMQCallbackHandler
+    {
+        @Override
+        public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException
+        {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public void initialise(ConnectionURL connectionURL)
+        {
+            throw new UnsupportedOperationException();
+        }
+    }
+
+}
diff --git a/java/client/src/test/java/org/apache/qpid/client/security/UsernameHashedPasswordCallbackHandlerTest.java b/java/client/src/test/java/org/apache/qpid/client/security/UsernameHashedPasswordCallbackHandlerTest.java
new file mode 100644
index 0000000000..9e23f722eb
--- /dev/null
+++ b/java/client/src/test/java/org/apache/qpid/client/security/UsernameHashedPasswordCallbackHandlerTest.java
@@ -0,0 +1,99 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.client.security;
+
+import java.security.MessageDigest;
+import java.util.Arrays;
+
+import javax.security.auth.callback.Callback;
+import javax.security.auth.callback.NameCallback;
+import javax.security.auth.callback.PasswordCallback;
+
+import junit.framework.TestCase;
+
+import org.apache.qpid.client.AMQConnection;
+import org.apache.qpid.client.AMQConnectionURL;
+import org.apache.qpid.client.MockAMQConnection;
+import org.apache.qpid.client.protocol.AMQProtocolHandler;
+import org.apache.qpid.client.protocol.AMQProtocolSession;
+
+/**
+ * Unit tests for the UsernameHashPasswordCallbackHandler.  This callback handler is
+ * used by the CRAM-MD5-HASHED SASL mechanism.
+ *
+ */
+public class UsernameHashedPasswordCallbackHandlerTest extends TestCase
+{
+    private AMQCallbackHandler _callbackHandler = new UsernameHashedPasswordCallbackHandler(); // Class under test
+    private static final String PROMPT_UNUSED = "unused";
+
+    @Override
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+
+        final String url = "amqp://username:password@client/test?brokerlist='tcp://localhost:1'";
+        _callbackHandler.initialise(new AMQConnectionURL(url));
+    }
+
+    /**
+     *  Tests that the callback handler can correctly retrieve the username from the connection url.
+     */
+    public void testNameCallback() throws Exception
+    {
+        final String expectedName = "username";
+        NameCallback nameCallback = new NameCallback(PROMPT_UNUSED);
+
+        assertNull("Unexpected name before test", nameCallback.getName());
+        _callbackHandler.handle(new Callback[] {nameCallback});
+        assertEquals("Unexpected name", expectedName, nameCallback.getName());
+    }
+
+    /**
+     *  Tests that the callback handler can correctly retrieve the password from the connection url
+     *  and calculate a MD5.
+     */
+    public void testDigestedPasswordCallback() throws Exception
+    {
+        final char[] expectedPasswordDigested = getHashPassword("password");
+        
+        PasswordCallback passwordCallback = new PasswordCallback(PROMPT_UNUSED, false);
+        assertNull("Unexpected password before test", passwordCallback.getPassword());
+        _callbackHandler.handle(new Callback[] {passwordCallback});
+        assertTrue("Unexpected password", Arrays.equals(expectedPasswordDigested, passwordCallback.getPassword()));
+    }
+
+    private char[] getHashPassword(final String password) throws Exception
+    {
+        MessageDigest md5Digester = MessageDigest.getInstance("MD5");
+        final byte[] digest = md5Digester.digest(password.getBytes("UTF-8"));
+
+        char[] hash = new char[digest.length];
+
+        int index = 0;
+        for (byte b : digest)
+        {
+            hash[index++] = (char) b;
+        }
+
+        return hash;
+    }
+}
diff --git a/java/client/src/test/java/org/apache/qpid/client/security/UsernamePasswordCallbackHandlerTest.java b/java/client/src/test/java/org/apache/qpid/client/security/UsernamePasswordCallbackHandlerTest.java
new file mode 100644
index 0000000000..83ddfd72fa
--- /dev/null
+++ b/java/client/src/test/java/org/apache/qpid/client/security/UsernamePasswordCallbackHandlerTest.java
@@ -0,0 +1,78 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.client.security;
+
+import javax.security.auth.callback.Callback;
+import javax.security.auth.callback.NameCallback;
+import javax.security.auth.callback.PasswordCallback;
+
+import junit.framework.TestCase;
+
+import org.apache.qpid.client.AMQConnection;
+import org.apache.qpid.client.AMQConnectionURL;
+import org.apache.qpid.client.MockAMQConnection;
+import org.apache.qpid.client.protocol.AMQProtocolHandler;
+import org.apache.qpid.client.protocol.AMQProtocolSession;
+
+/**
+ * Unit tests for the UsernamePasswordCallbackHandler.
+ *
+ */
+public class UsernamePasswordCallbackHandlerTest extends TestCase
+{
+    private AMQCallbackHandler _callbackHandler = new UsernamePasswordCallbackHandler(); // Class under test
+    private static final String PROMPT_UNUSED = "unused";
+
+    @Override
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+
+        final String url = "amqp://username:password@client/test?brokerlist='tcp://localhost:1'";
+
+        _callbackHandler.initialise(new AMQConnectionURL(url));
+    }
+
+    /**
+     *  Tests that the callback handler can correctly retrieve the username from the connection url.
+     */
+    public void testNameCallback() throws Exception
+    {
+        final String expectedName = "username";
+        NameCallback nameCallback = new NameCallback(PROMPT_UNUSED);
+
+        assertNull("Unexpected name before test", nameCallback.getName());
+        _callbackHandler.handle(new Callback[] {nameCallback});
+        assertEquals("Unexpected name", expectedName, nameCallback.getName());
+    }
+
+    /**
+     *  Tests that the callback handler can correctly retrieve the password from the connection url.
+     */
+    public void testPasswordCallback() throws Exception
+    {
+        final String expectedPassword = "password";
+        PasswordCallback passwordCallback = new PasswordCallback(PROMPT_UNUSED, false);
+        assertNull("Unexpected password before test", passwordCallback.getPassword());
+        _callbackHandler.handle(new Callback[] {passwordCallback});
+        assertEquals("Unexpected password", expectedPassword, new String(passwordCallback.getPassword()));
+    }    
+}
diff --git a/java/client/src/test/java/org/apache/qpid/client/util/ClassLoadingAwareObjectInputStreamTest.java b/java/client/src/test/java/org/apache/qpid/client/util/ClassLoadingAwareObjectInputStreamTest.java
new file mode 100644
index 0000000000..a12e4ce977
--- /dev/null
+++ b/java/client/src/test/java/org/apache/qpid/client/util/ClassLoadingAwareObjectInputStreamTest.java
@@ -0,0 +1,86 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.client.util;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.InputStream;
+import java.io.ObjectOutputStream;
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.qpid.test.utils.QpidTestCase;
+
+public class ClassLoadingAwareObjectInputStreamTest extends QpidTestCase
+{
+    InputStream _in;
+    ClassLoadingAwareObjectInputStream _claOIS;
+
+    protected void setUp() throws Exception
+    {
+        //Create a viable input stream for instantiating the CLA OIS
+        ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+        ObjectOutputStream out = new ObjectOutputStream(baos);
+        out.writeObject("testString");
+        out.flush();
+        out.close();
+
+
+        _in = new ByteArrayInputStream(baos.toByteArray());
+
+        _claOIS = new ClassLoadingAwareObjectInputStream(_in);
+    }
+
+    /**
+     * Test that the resolveProxyClass method returns a proxy class implementing the desired interface
+     */
+    public void testResolveProxyClass() throws Exception
+    {
+        //try to proxy an interface
+        Class<?> clazz = _claOIS.resolveProxyClass(new String[]{"java.lang.CharSequence"});
+
+        //verify the proxy supports the expected interface (only)
+        List<Class<?>> interfaces = Arrays.asList(clazz.getInterfaces());
+        assertTrue("Unexpected interfaces supported by proxy", interfaces.contains(CharSequence.class));
+        assertEquals("Unexpected interfaces supported by proxy", 1, interfaces.size());
+    }
+
+    /**
+     * Test that the resolveProxyClass method throws a ClassNotFoundException wrapping an
+     * IllegalArgumentException if it is provided arguments which violate the restrictions allowed
+     * by Proxy.getProxyClass (as required by the ObjectInputStream.resolveProxyClass javadoc).
+     */
+    public void testResolveProxyClassThrowsCNFEWrappingIAE() throws Exception
+    {
+        try
+        {
+            //try to proxy a *class* rather than an interface, which is illegal
+            _claOIS.resolveProxyClass(new String[]{"java.lang.String"});
+            fail("should have thrown an exception");
+        }
+        catch(ClassNotFoundException cnfe)
+        {
+            //expected, but must verify it is wrapping an IllegalArgumentException
+            assertTrue(cnfe.getCause() instanceof IllegalArgumentException);
+        }
+    }
+}
diff --git a/java/client/src/test/java/org/apache/qpid/jms/FailoverPolicyTest.java b/java/client/src/test/java/org/apache/qpid/jms/FailoverPolicyTest.java
new file mode 100644
index 0000000000..438995aedc
--- /dev/null
+++ b/java/client/src/test/java/org/apache/qpid/jms/FailoverPolicyTest.java
@@ -0,0 +1,338 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.jms;
+
+import javax.jms.ConnectionConsumer;
+import javax.jms.ConnectionMetaData;
+import javax.jms.Destination;
+import javax.jms.ExceptionListener;
+import javax.jms.JMSException;
+import javax.jms.ServerSessionPool;
+import javax.jms.Topic;
+
+import org.apache.qpid.client.AMQConnectionURL;
+import org.apache.qpid.jms.failover.FailoverExchangeMethod;
+import org.apache.qpid.jms.failover.FailoverMethod;
+import org.apache.qpid.jms.failover.FailoverRoundRobinServers;
+import org.apache.qpid.jms.failover.FailoverSingleServer;
+import org.apache.qpid.jms.failover.NoFailover;
+
+import junit.framework.TestCase;
+
+/**
+ * Tests the ability of FailoverPolicy to instantiate the correct FailoverMethod.
+ *
+ * This test presently does <i>not</i> test {@link FailoverPolicy#FailoverPolicy(FailoverMethod) or
+ * {@link FailoverPolicy#addMethod(FailoverMethod)} as it appears that this functionality
+ * is no longer in use.
+ *
+ */
+public class FailoverPolicyTest extends TestCase
+{
+    private FailoverPolicy _failoverPolicy = null; // class under test
+    private String _url;
+    private Connection _connection = null;
+    private ConnectionURL _connectionUrl = null;
+
+    /**
+     * Tests single server method is selected for a brokerlist with one broker when
+     * the failover option is not specified.
+     */
+    public void testBrokerListWithOneBrokerDefaultsToSingleServerPolicy() throws Exception
+    {
+        _url = "amqp://user:pass@clientid/test?brokerlist='tcp://localhost:5672'";
+        _connectionUrl = new AMQConnectionURL(_url);
+        _connection = createStubConnection();
+
+        _failoverPolicy = new FailoverPolicy(_connectionUrl, _connection);
+
+        assertTrue("Unexpected failover method", _failoverPolicy.getCurrentMethod() instanceof FailoverSingleServer);
+    }
+
+    /**
+     * Tests round robin method is selected for a brokerlist with two brokers when
+     * the failover option is not specified.
+     */
+    public void testBrokerListWithTwoBrokersDefaultsToRoundRobinPolicy() throws Exception
+    {
+        _url = "amqp://user:pass@clientid/test?brokerlist='tcp://localhost:5672;tcp://localhost:5673'";
+        _connectionUrl = new AMQConnectionURL(_url);
+        _connection = createStubConnection();
+
+        _failoverPolicy = new FailoverPolicy(_connectionUrl, _connection);
+
+        assertTrue("Unexpected failover method", _failoverPolicy.getCurrentMethod() instanceof FailoverRoundRobinServers);
+    }
+
+    /**
+     * Tests single server method is selected for a brokerlist with one broker when
+     * the failover option passed as 'singlebroker'.
+     */
+    public void testExplictFailoverOptionSingleBroker() throws Exception
+    {
+        _url = "amqp://user:pass@clientid/test?brokerlist='tcp://localhost:5672'&failover='singlebroker'";
+        _connectionUrl = new AMQConnectionURL(_url);
+        _connection = createStubConnection();
+
+        _failoverPolicy = new FailoverPolicy(_connectionUrl, _connection);
+
+        assertTrue("Unexpected failover method", _failoverPolicy.getCurrentMethod() instanceof FailoverSingleServer);
+    }
+
+    /**
+     * Tests round robin method is selected for a brokerlist with two brokers when
+     * the failover option passed as 'roundrobin'.
+     */
+    public void testExplictFailoverOptionRoundrobin() throws Exception
+    {
+        _url = "amqp://user:pass@clientid/test?brokerlist='tcp://localhost:5672;tcp://localhost:5673'&failover='roundrobin'";
+        _connectionUrl = new AMQConnectionURL(_url);
+        _connection = createStubConnection();
+
+        _failoverPolicy = new FailoverPolicy(_connectionUrl, _connection);
+
+        assertTrue("Unexpected failover method", _failoverPolicy.getCurrentMethod() instanceof FailoverRoundRobinServers);
+    }
+
+    /**
+     * Tests no failover method is selected for a brokerlist with one broker when
+     * the failover option passed as 'nofailover'.
+     */
+    public void testExplictFailoverOptionNofailover() throws Exception
+    {
+        _url = "amqp://user:pass@clientid/test?brokerlist='tcp://localhost:5672'&failover='nofailover'";
+        _connectionUrl = new AMQConnectionURL(_url);
+        _connection = createStubConnection();
+
+        _failoverPolicy = new FailoverPolicy(_connectionUrl, _connection);
+
+        assertTrue("Unexpected failover method", _failoverPolicy.getCurrentMethod() instanceof NoFailover);
+    }
+
+    /**
+     * Tests failover exchange method is selected for a brokerlist with one broker when
+     * the failover option passed as 'failover_exchange'.
+     */
+    public void testExplictFailoverOptionFailoverExchange() throws Exception
+    {
+        _url = "amqp://user:pass@clientid/test?brokerlist='tcp://localhost:5672'&failover='failover_exchange'";
+        _connectionUrl = new AMQConnectionURL(_url);
+        _connection = createStubConnection();
+
+        _failoverPolicy = new FailoverPolicy(_connectionUrl, _connection);
+
+        assertTrue("Unexpected failover method", _failoverPolicy.getCurrentMethod() instanceof FailoverExchangeMethod);
+    }
+
+    /**
+     * Tests that a custom method can be selected for a brokerlist with one brokers when
+     * the failover option passed as a qualified class-name.
+     */
+    public void testExplictFailoverOptionDynamicallyLoadedFailoverMethod() throws Exception
+    {
+        _url = "amqp://user:pass@clientid/test?brokerlist='tcp://localhost:5672'&failover='org.apache.qpid.jms.FailoverPolicyTest$MyFailoverMethod'";
+        _connectionUrl = new AMQConnectionURL(_url);
+        _connection = createStubConnection();
+
+        _failoverPolicy = new FailoverPolicy(_connectionUrl, _connection);
+
+        assertTrue("Unexpected failover method", _failoverPolicy.getCurrentMethod() instanceof MyFailoverMethod);
+    }
+
+    /**
+     * Tests that an unknown method caused an exception.
+     */
+    public void testUnknownFailoverMethod() throws Exception
+    {
+        _url = "amqp://user:pass@clientid/test?brokerlist='tcp://localhost:5672'&failover='unknown'";
+        _connectionUrl = new AMQConnectionURL(_url);
+        _connection = createStubConnection();
+
+        try
+        {
+            new FailoverPolicy(_connectionUrl, _connection);
+            fail("Exception not thrown");
+        }
+        catch(IllegalArgumentException iae)
+        {
+            // PASS
+        }
+    }
+
+    private Connection createStubConnection()
+    {
+        return new Connection()
+        {
+
+            @Override
+            public Session createSession(boolean transacted,
+                    int acknowledgeMode, int prefetch) throws JMSException
+            {
+                return null;
+            }
+
+            @Override
+            public Session createSession(boolean transacted,
+                    int acknowledgeMode, int prefetchHigh, int prefetchLow)
+                    throws JMSException
+            {
+                return null;
+            }
+
+            @Override
+            public ConnectionListener getConnectionListener()
+            {
+                return null;
+            }
+
+            @Override
+            public long getMaximumChannelCount() throws JMSException
+            {
+                return 0;
+            }
+
+            @Override
+            public void setConnectionListener(ConnectionListener listener)
+            {
+            }
+
+            @Override
+            public void close() throws JMSException
+            {
+            }
+
+            @Override
+            public ConnectionConsumer createConnectionConsumer(
+                    Destination arg0, String arg1, ServerSessionPool arg2,
+                    int arg3) throws JMSException
+            {
+                return null;
+            }
+
+            @Override
+            public ConnectionConsumer createDurableConnectionConsumer(
+                    Topic arg0, String arg1, String arg2,
+                    ServerSessionPool arg3, int arg4) throws JMSException
+            {
+                return null;
+            }
+
+            @Override
+            public javax.jms.Session createSession(boolean arg0, int arg1)
+                    throws JMSException
+            {
+                return null;
+            }
+
+            @Override
+            public String getClientID() throws JMSException
+            {
+                return null;
+            }
+
+            @Override
+            public ExceptionListener getExceptionListener() throws JMSException
+            {
+                return null;
+            }
+
+            @Override
+            public ConnectionMetaData getMetaData() throws JMSException
+            {
+                return null;
+            }
+
+            @Override
+            public void setClientID(String arg0) throws JMSException
+            {
+            }
+
+            @Override
+            public void setExceptionListener(ExceptionListener arg0)
+                    throws JMSException
+            {
+            }
+
+            @Override
+            public void start() throws JMSException
+            {
+            }
+
+            @Override
+            public void stop() throws JMSException
+            {
+            }
+        };
+    }
+
+    // Class used to test the ability of FailoverPolicy to load an implementation.
+    static class MyFailoverMethod implements FailoverMethod
+    {
+        public MyFailoverMethod(ConnectionURL connectionDetails)
+        {
+        }
+
+        @Override
+        public void attainedConnection()
+        {
+        }
+
+        @Override
+        public boolean failoverAllowed()
+        {
+            return false;
+        }
+
+        @Override
+        public BrokerDetails getCurrentBrokerDetails()
+        {
+            return null;
+        }
+
+        @Override
+        public BrokerDetails getNextBrokerDetails()
+        {
+            return null;
+        }
+
+        @Override
+        public String methodName()
+        {
+            return null;
+        }
+
+        @Override
+        public void reset()
+        {
+        }
+
+        @Override
+        public void setBroker(BrokerDetails broker)
+        {
+        }
+
+        @Override
+        public void setRetries(int maxRetries)
+        {
+        }
+    }
+
+}
diff --git a/java/client/src/test/java/org/apache/qpid/test/unit/client/BrokerDetails/BrokerDetailsTest.java b/java/client/src/test/java/org/apache/qpid/test/unit/client/BrokerDetails/BrokerDetailsTest.java
index 1b27ff6300..9095f94960 100644
--- a/java/client/src/test/java/org/apache/qpid/test/unit/client/BrokerDetails/BrokerDetailsTest.java
+++ b/java/client/src/test/java/org/apache/qpid/test/unit/client/BrokerDetails/BrokerDetailsTest.java
@@ -43,15 +43,6 @@ public class BrokerDetailsTest extends TestCase
         assertTrue(broker.getProperty("immediatedelivery").equals("true"));
     }
 
-    public void testVMBroker() throws URLSyntaxException
-    {
-        String url = "vm://:2";
-
-        AMQBrokerDetails broker = new AMQBrokerDetails(url);
-        assertTrue(broker.getTransport().equals("vm"));
-        assertEquals(broker.getPort(), 2);
-    }
-
     public void testTransportsDefaultToTCP() throws URLSyntaxException
     {
         String url = "localhost:5672";
diff --git a/java/client/src/test/java/org/apache/qpid/test/unit/client/channelclose/ChannelCloseMethodHandlerNoCloseOk.java b/java/client/src/test/java/org/apache/qpid/test/unit/client/channelclose/ChannelCloseMethodHandlerNoCloseOk.java
index 66f220643c..d560c413e6 100644
--- a/java/client/src/test/java/org/apache/qpid/test/unit/client/channelclose/ChannelCloseMethodHandlerNoCloseOk.java
+++ b/java/client/src/test/java/org/apache/qpid/test/unit/client/channelclose/ChannelCloseMethodHandlerNoCloseOk.java
@@ -73,7 +73,7 @@ public class ChannelCloseMethodHandlerNoCloseOk implements StateAwareMethodListe
             {
                 throw new AMQNoRouteException("Error: " + reason, null, null);
             }
-            else if (errorCode == AMQConstant.INVALID_ARGUMENT)
+            else if (errorCode == AMQConstant.ARGUMENT_INVALID)
             {
                 _logger.debug("Broker responded with Invalid Argument.");
 
diff --git a/java/client/src/test/java/org/apache/qpid/test/unit/client/connectionurl/ConnectionURLTest.java b/java/client/src/test/java/org/apache/qpid/test/unit/client/connectionurl/ConnectionURLTest.java
index 2be3720c20..4624b36fea 100644
--- a/java/client/src/test/java/org/apache/qpid/test/unit/client/connectionurl/ConnectionURLTest.java
+++ b/java/client/src/test/java/org/apache/qpid/test/unit/client/connectionurl/ConnectionURLTest.java
@@ -300,53 +300,6 @@ public class ConnectionURLTest extends TestCase
         assertTrue(connectionurl.getOption("immediatedelivery").equals("true"));
     }
 
-    public void testSinglevmURL() throws URLSyntaxException
-    {
-        String url = "amqp://guest:guest@/test?brokerlist='vm://:2'";
-
-        ConnectionURL connectionurl = new AMQConnectionURL(url);
-
-        assertTrue(connectionurl.getFailoverMethod() == null);
-        assertTrue(connectionurl.getUsername().equals("guest"));
-        assertTrue(connectionurl.getPassword().equals("guest"));
-        assertTrue(connectionurl.getVirtualHost().equals("/test"));
-
-        assertTrue(connectionurl.getBrokerCount() == 1);
-
-        BrokerDetails service = connectionurl.getBrokerDetails(0);
-
-        assertTrue(service.getTransport().equals("vm"));
-        assertTrue(service.getHost().equals(""));
-        assertTrue(service.getPort() == 2);
-
-    }
-
-    public void testFailoverVMURL() throws URLSyntaxException
-    {
-        String url = "amqp://ritchiem:bob@/test?brokerlist='vm://:2;vm://:3',failover='roundrobin'";
-
-        ConnectionURL connectionurl = new AMQConnectionURL(url);
-
-        assertTrue(connectionurl.getFailoverMethod().equals("roundrobin"));
-        assertTrue(connectionurl.getUsername().equals("ritchiem"));
-        assertTrue(connectionurl.getPassword().equals("bob"));
-        assertTrue(connectionurl.getVirtualHost().equals("/test"));
-
-        assertTrue(connectionurl.getBrokerCount() == 2);
-
-        BrokerDetails service = connectionurl.getBrokerDetails(0);
-
-        assertTrue(service.getTransport().equals("vm"));
-        assertTrue(service.getHost().equals(""));
-        assertTrue(service.getPort() == 2);
-
-        service = connectionurl.getBrokerDetails(1);
-        assertTrue(service.getTransport().equals("vm"));
-        assertTrue(service.getHost().equals(""));
-        assertTrue(service.getPort() == 3);
-    }
-
-
     public void testNoVirtualHostURL()
     {
         String url = "amqp://user@?brokerlist='tcp://localhost:5672'";
@@ -487,27 +440,6 @@ public class ConnectionURLTest extends TestCase
 
     }
 
-    public void testSocketProtocol() throws URLSyntaxException
-    {
-        String url = "amqp://guest:guest@id/test" + "?brokerlist='socket://VM-Unique-socketID'";
-
-        try
-        {
-            AMQConnectionURL curl = new AMQConnectionURL(url);
-            assertNotNull(curl);
-            assertEquals(1, curl.getBrokerCount());
-            assertNotNull(curl.getBrokerDetails(0));
-            assertEquals(BrokerDetails.SOCKET, curl.getBrokerDetails(0).getTransport());
-            assertEquals("VM-Unique-socketID", curl.getBrokerDetails(0).getHost());
-            assertEquals("URL does not toString as expected",
-                         url.replace(":guest", ":********"), curl.toString());
-        }
-        catch (URLSyntaxException e)
-        {
-            fail(e.getMessage());
-        }
-    }
-
     public void testSingleTransportMultiOptionOnBrokerURL() throws URLSyntaxException
     {
         String url = "amqp://guest:guest@/test?brokerlist='tcp://localhost:5672?foo='jim'&bar='bob'&fred='jimmy'',routingkey='jim',timeout='200',immediatedelivery='true'";
@@ -549,6 +481,37 @@ public class ConnectionURLTest extends TestCase
         assertTrue("String representation should contain options and values", url.toString().contains("maxprefetch='12345'"));
     }
 
+    public void testHostNamesWithUnderScore() throws URLSyntaxException
+    {
+        String url = "amqp://guest:guest@clientid/test?brokerlist='tcp://under_score:6672'";
+
+        ConnectionURL connectionurl = new AMQConnectionURL(url);
+
+        assertTrue(connectionurl.getUsername().equals("guest"));
+        assertTrue(connectionurl.getPassword().equals("guest"));
+        assertTrue(connectionurl.getVirtualHost().equals("/test"));
+
+        assertTrue(connectionurl.getBrokerCount() == 1);
+        BrokerDetails service = connectionurl.getBrokerDetails(0);
+        assertTrue(service.getTransport().equals("tcp"));        
+        assertTrue(service.getHost().equals("under_score"));
+        assertTrue(service.getPort() == 6672);
+        
+        url = "amqp://guest:guest@clientid/test?brokerlist='tcp://under_score'";
+
+        connectionurl = new AMQConnectionURL(url);
+
+        assertTrue(connectionurl.getUsername().equals("guest"));
+        assertTrue(connectionurl.getPassword().equals("guest"));
+        assertTrue(connectionurl.getVirtualHost().equals("/test"));
+
+        assertTrue(connectionurl.getBrokerCount() == 1);
+        service = connectionurl.getBrokerDetails(0);
+        assertTrue(service.getTransport().equals("tcp"));        
+        assertTrue(service.getHost().equals("under_score"));
+        assertTrue(service.getPort() == 5672);
+    }
+    
     public static junit.framework.Test suite()
     {
         return new junit.framework.TestSuite(ConnectionURLTest.class);
diff --git a/java/client/src/test/java/org/apache/qpid/test/unit/client/message/ObjectMessageUnitTest.java b/java/client/src/test/java/org/apache/qpid/test/unit/client/message/ObjectMessageUnitTest.java
new file mode 100644
index 0000000000..e37970e9a2
--- /dev/null
+++ b/java/client/src/test/java/org/apache/qpid/test/unit/client/message/ObjectMessageUnitTest.java
@@ -0,0 +1,105 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.test.unit.client.message;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+
+import org.apache.qpid.client.message.JMSObjectMessage;
+import org.apache.qpid.client.message.TestMessageHelper;
+import org.apache.qpid.test.utils.QpidTestCase;
+
+public class ObjectMessageUnitTest extends QpidTestCase
+{
+    private JMSObjectMessage _om;
+
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+        _om = TestMessageHelper.newJMSObjectMessage();
+    }
+
+    /**
+     * Test that setObject with a primitive works
+     */
+    public void testSetObjectWithBooleanPrimitive() throws Exception
+    {
+        _om.setObject(true);
+
+        //make the message readable
+        Object object = _om.getObject();
+
+        assertTrue("Unexpected type returned", object instanceof Boolean);
+        assertEquals("Unexpected value returned", true, object);
+    }
+
+    /**
+     * Test that setObject with a serializable Object works
+     */
+    public void testSetObjectWithString() throws Exception
+    {
+        _om.setObject("test string");
+
+        //make the message readable
+        Object object = _om.getObject();
+
+        assertTrue("Unexpected type returned", object instanceof String);
+        assertEquals("Unexpected value returned", "test string", object);
+    }
+
+    /**
+     * Test that setObject with a Collection of serializable's works, returning
+     * the items in the list when deserialized and ignoring any values
+     * added to the collection after setObject() is called on the message.
+     */
+    public void testSetObjectWithArrayListOfInteger() throws Exception
+    {
+        ArrayList<Integer> list = new ArrayList<Integer>();
+        list.add(1234);
+        list.add(Integer.MIN_VALUE);
+        list.add(Integer.MAX_VALUE);
+
+        _om.setObject(list);
+
+        //add something extra to the list now, and check it isn't in the value read back
+        list.add(0);
+
+        //make the message readable
+
+        //retrieve the Object
+        Object object = _om.getObject();
+
+        ArrayList<?> returnedList = null;
+        if(object instanceof ArrayList<?>)
+        {
+            returnedList = (ArrayList<?>) object;
+        }
+        else
+        {
+            fail("returned object was not an ArrayList");
+        }
+
+        //verify the extra added Integer was not present, then remove it from original list again and compare contents with the returned list
+        assertFalse("returned list should not have had the value added after setObject() was used", returnedList.contains(0));
+        list.remove(Integer.valueOf(0));
+        assertTrue("list contents were not equal", Arrays.equals(list.toArray(), returnedList.toArray()));
+    }
+}
diff --git a/java/client/src/test/java/org/apache/qpid/test/unit/jndi/ConnectionFactoryTest.java b/java/client/src/test/java/org/apache/qpid/test/unit/jndi/ConnectionFactoryTest.java
index 9e76b0d468..20496026ce 100644
--- a/java/client/src/test/java/org/apache/qpid/test/unit/jndi/ConnectionFactoryTest.java
+++ b/java/client/src/test/java/org/apache/qpid/test/unit/jndi/ConnectionFactoryTest.java
@@ -21,10 +21,10 @@
 package org.apache.qpid.test.unit.jndi;
 
 import junit.framework.TestCase;
+
 import org.apache.qpid.client.AMQConnectionFactory;
 import org.apache.qpid.jms.BrokerDetails;
 import org.apache.qpid.jms.ConnectionURL;
-import org.apache.qpid.url.URLSyntaxException;
 
 public class ConnectionFactoryTest extends TestCase
 {
@@ -34,21 +34,9 @@ public class ConnectionFactoryTest extends TestCase
     public static final String URL = "amqp://guest:guest@clientID/test?brokerlist='tcp://localhost:5672'";
     public static final String URL_STAR_PWD = "amqp://guest:********@clientID/test?brokerlist='tcp://localhost:5672'";
 
-    public void testConnectionURLString()
+    public void testConnectionURLStringMasksPassword() throws Exception
     {
-        AMQConnectionFactory factory = new AMQConnectionFactory();
-
-        assertNull("ConnectionURL should have no value at start",
-                   factory.getConnectionURL());
-
-        try
-        {
-            factory.setConnectionURLString(URL);
-        }
-        catch (URLSyntaxException e)
-        {
-            fail(e.getMessage());
-        }
+        AMQConnectionFactory factory = new AMQConnectionFactory(URL);
 
         //URL will be returned with the password field swapped for '********'
         assertEquals("Connection URL not correctly set", URL_STAR_PWD, factory.getConnectionURLString());
diff --git a/java/client/src/test/java/org/apache/qpid/test/unit/jndi/JNDIPropertyFileTest.java b/java/client/src/test/java/org/apache/qpid/test/unit/jndi/JNDIPropertyFileTest.java
index a1b14d5723..2052312f54 100644
--- a/java/client/src/test/java/org/apache/qpid/test/unit/jndi/JNDIPropertyFileTest.java
+++ b/java/client/src/test/java/org/apache/qpid/test/unit/jndi/JNDIPropertyFileTest.java
@@ -24,6 +24,7 @@ import java.util.Properties;
 
 import javax.jms.Queue;
 import javax.jms.Topic;
+import javax.naming.ConfigurationException;
 import javax.naming.Context;
 import javax.naming.InitialContext;
 
@@ -67,4 +68,22 @@ public class JNDIPropertyFileTest extends TestCase
             assertEquals("Topic" + i + "WithSpace",bindingKey.asString());            
         }
     }
+    
+    public void testConfigurationErrors() throws Exception
+    {
+        Properties properties = new Properties();
+        properties.put("java.naming.factory.initial", "org.apache.qpid.jndi.PropertiesFileInitialContextFactory");
+        properties.put("destination.my-queue","amq.topic/test;create:always}");
+        
+        try
+        {
+            ctx = new InitialContext(properties);
+            fail("A configuration exception should be thrown with details about the address syntax error");
+        }
+        catch(ConfigurationException e)
+        {
+            assertTrue("Incorrect exception", e.getMessage().contains("Failed to parse entry: amq.topic/test;create:always}"));
+        }
+        
+    }
 }
diff --git a/java/client/src/test/java/org/apache/qpid/test/unit/message/TestAMQSession.java b/java/client/src/test/java/org/apache/qpid/test/unit/message/TestAMQSession.java
index 47c0359b94..6759b43387 100644
--- a/java/client/src/test/java/org/apache/qpid/test/unit/message/TestAMQSession.java
+++ b/java/client/src/test/java/org/apache/qpid/test/unit/message/TestAMQSession.java
@@ -20,17 +20,24 @@
  */
 package org.apache.qpid.test.unit.message;
 
-import org.apache.qpid.client.*;
+import java.util.Map;
+
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.TemporaryQueue;
+import javax.jms.Topic;
+import javax.jms.TopicSubscriber;
+
+import org.apache.qpid.AMQException;
+import org.apache.qpid.client.AMQDestination;
+import org.apache.qpid.client.AMQSession;
+import org.apache.qpid.client.BasicMessageConsumer_0_8;
+import org.apache.qpid.client.BasicMessageProducer_0_8;
+import org.apache.qpid.client.failover.FailoverException;
 import org.apache.qpid.client.message.AMQMessageDelegateFactory;
 import org.apache.qpid.client.protocol.AMQProtocolHandler;
-import org.apache.qpid.client.failover.FailoverException;
 import org.apache.qpid.framing.AMQShortString;
 import org.apache.qpid.framing.FieldTable;
-import org.apache.qpid.AMQException;
-
-import javax.jms.*;
-
-import java.util.Map;
 
 public class TestAMQSession extends AMQSession<BasicMessageConsumer_0_8, BasicMessageProducer_0_8>
 {
@@ -57,7 +64,12 @@ public class TestAMQSession extends AMQSession<BasicMessageConsumer_0_8, BasicMe
 
     }
 
-    public void sendCommit() throws AMQException, FailoverException
+    public void commitImpl() throws AMQException, FailoverException
+    {
+
+    }
+
+    public void acknowledgeImpl()
     {
 
     }
@@ -117,7 +129,7 @@ public class TestAMQSession extends AMQSession<BasicMessageConsumer_0_8, BasicMe
 
     }
 
-    public BasicMessageProducer_0_8 createMessageProducer(Destination destination, boolean mandatory, boolean immediate, boolean waitUntilSent, long producerId)
+    public BasicMessageProducer_0_8 createMessageProducer(Destination destination, boolean mandatory, boolean immediate, long producerId)
     {
         return null;
     }
@@ -195,4 +207,10 @@ public class TestAMQSession extends AMQSession<BasicMessageConsumer_0_8, BasicMe
     {
         return false;
     }
+
+    @Override
+    public AMQException getLastException()
+    {
+        return null;
+    }
 }