synchronized with trunk except for ruby dir

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

13 files changed, 2160 insertions, 144 deletions

diff --git a/java/common/src/test/java/org/apache/qpid/AMQExceptionTest.java b/java/common/src/test/java/org/apache/qpid/AMQExceptionTest.java
new file mode 100644
index 0000000000..ef6cd41492
--- /dev/null
+++ b/java/common/src/test/java/org/apache/qpid/AMQExceptionTest.java
@@ -0,0 +1,106 @@
+/*
+ *
+ * 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;
+
+import junit.framework.TestCase;
+import org.apache.qpid.protocol.AMQConstant;
+import org.apache.qpid.framing.AMQFrameDecodingException;
+
+/**
+ * This test is to ensure that when an AMQException is rethrown that the specified exception is correctly wrapped up.
+ *
+ * There are three cases:
+ * Re-throwing an AMQException
+ * Re-throwing a Subclass of AMQException
+ * Re-throwing a Subclass of AMQException that does not have the default AMQException constructor which will force the
+ * creation of an AMQException.
+ */
+public class AMQExceptionTest extends TestCase
+{
+    /**
+     * Test that an AMQException will be correctly created and rethrown.
+     */
+    public void testRethrowGeneric()
+    {
+        AMQException test = new AMQException(AMQConstant.ACCESS_REFUSED, "refused", new RuntimeException());
+
+        AMQException e = reThrowException(test);
+
+        assertEquals("Exception not of correct class", AMQException.class, e.getClass());
+
+    }
+
+    /**
+     * Test that a subclass of AMQException that has the default constructor will be correctly created and rethrown.
+     */
+    public void testRethrowAMQESubclass()
+    {
+        AMQFrameDecodingException test = new AMQFrameDecodingException(AMQConstant.INTERNAL_ERROR,
+                                                                       "Error",
+                                                                       new Exception());
+        AMQException e = reThrowException(test);
+
+        assertEquals("Exception not of correct class", AMQFrameDecodingException.class, e.getClass());
+    }
+
+    /**
+     * Test that a subclass of AMQException that doesnot have the  default constructor will be correctly rethrown as an
+     * AMQException
+     */
+    public void testRethrowAMQESubclassNoConstructor()
+    {
+        AMQExceptionSubclass test = new AMQExceptionSubclass("Invalid Argument Exception");
+
+        AMQException e = reThrowException(test);
+
+        assertEquals("Exception not of correct class", AMQException.class, e.getClass());
+    }
+
+    /**
+     * Private method to rethrown and validate the basic values of the rethrown
+     * @param test Exception to rethrow
+     * @throws AMQException the rethrown exception
+     */
+    private AMQException reThrowException(AMQException test)
+    {
+        AMQException amqe = test.cloneForCurrentThread();
+
+        assertEquals("Error code does not match.", test.getErrorCode(), amqe.getErrorCode());
+        assertTrue("Exception message does not start as expected.", amqe.getMessage().startsWith(test.getMessage()));
+        assertEquals("Test Exception is not set as the cause", test, amqe.getCause());
+        assertEquals("Cause is not correct", test.getCause(), amqe.getCause().getCause());
+
+        return amqe;
+    }
+
+    /**
+     * Private class that extends AMQException but does not have a default exception.
+     */
+    private class AMQExceptionSubclass extends AMQException
+    {
+
+        public AMQExceptionSubclass(String msg)
+        {
+            super(null, msg, null);
+        }
+    }
+}
+
diff --git a/java/common/src/test/java/org/apache/qpid/codec/AMQDecoderTest.java b/java/common/src/test/java/org/apache/qpid/codec/AMQDecoderTest.java
new file mode 100644
index 0000000000..46c812e265
--- /dev/null
+++ b/java/common/src/test/java/org/apache/qpid/codec/AMQDecoderTest.java
@@ -0,0 +1,130 @@
+package org.apache.qpid.codec;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+
+import junit.framework.TestCase;
+
+import org.apache.qpid.framing.AMQDataBlock;
+import org.apache.qpid.framing.AMQFrame;
+import org.apache.qpid.framing.AMQFrameDecodingException;
+import org.apache.qpid.framing.AMQProtocolVersionException;
+import org.apache.qpid.framing.HeartbeatBody;
+
+public class AMQDecoderTest extends TestCase
+{
+
+    private AMQCodecFactory _factory;
+    private AMQDecoder _decoder;
+
+
+    public void setUp()
+    {
+        _factory = new AMQCodecFactory(false, null);
+        _decoder = _factory.getDecoder();
+    }
+   
+    
+    public void testSingleFrameDecode() throws AMQProtocolVersionException, AMQFrameDecodingException
+    {
+        ByteBuffer msg = HeartbeatBody.FRAME.toNioByteBuffer();
+        ArrayList<AMQDataBlock> frames = _decoder.decodeBuffer(msg);
+        if (frames.get(0) instanceof AMQFrame)
+        {
+            assertEquals(HeartbeatBody.FRAME.getBodyFrame().getFrameType(), ((AMQFrame) frames.get(0)).getBodyFrame().getFrameType());
+        }
+        else
+        {
+            fail("decode was not a frame");
+        }
+    }
+    
+    public void testPartialFrameDecode() throws AMQProtocolVersionException, AMQFrameDecodingException
+    {
+        ByteBuffer msg = HeartbeatBody.FRAME.toNioByteBuffer();
+        ByteBuffer msgA = msg.slice();
+        int msgbPos = msg.remaining() / 2;
+        int msgaLimit = msg.remaining() - msgbPos;
+        msgA.limit(msgaLimit);
+        msg.position(msgbPos);
+        ByteBuffer msgB = msg.slice();
+        ArrayList<AMQDataBlock> frames = _decoder.decodeBuffer(msgA);
+        assertEquals(0, frames.size());
+        frames = _decoder.decodeBuffer(msgB);
+        assertEquals(1, frames.size());
+        if (frames.get(0) instanceof AMQFrame)
+        {
+            assertEquals(HeartbeatBody.FRAME.getBodyFrame().getFrameType(), ((AMQFrame) frames.get(0)).getBodyFrame().getFrameType());
+        }
+        else
+        {
+            fail("decode was not a frame");
+        }
+    }
+    
+    public void testMultipleFrameDecode() throws AMQProtocolVersionException, AMQFrameDecodingException
+    {
+        ByteBuffer msgA = HeartbeatBody.FRAME.toNioByteBuffer();
+        ByteBuffer msgB = HeartbeatBody.FRAME.toNioByteBuffer();
+        ByteBuffer msg = ByteBuffer.allocate(msgA.remaining() + msgB.remaining());
+        msg.put(msgA);
+        msg.put(msgB);
+        msg.flip();
+        ArrayList<AMQDataBlock> frames = _decoder.decodeBuffer(msg);
+        assertEquals(2, frames.size());
+        for (AMQDataBlock frame : frames)
+        {
+            if (frame instanceof AMQFrame)
+            {
+                assertEquals(HeartbeatBody.FRAME.getBodyFrame().getFrameType(), ((AMQFrame) frame).getBodyFrame().getFrameType());
+            }
+            else
+            {
+                fail("decode was not a frame");
+            }
+        }
+    }
+    
+    public void testMultiplePartialFrameDecode() throws AMQProtocolVersionException, AMQFrameDecodingException
+    {
+        ByteBuffer msgA = HeartbeatBody.FRAME.toNioByteBuffer();
+        ByteBuffer msgB = HeartbeatBody.FRAME.toNioByteBuffer();
+        ByteBuffer msgC = HeartbeatBody.FRAME.toNioByteBuffer();
+        
+        ByteBuffer sliceA = ByteBuffer.allocate(msgA.remaining() + msgB.remaining() / 2);
+        sliceA.put(msgA);
+        int limit = msgB.limit();
+        int pos = msgB.remaining() / 2;
+        msgB.limit(pos);
+        sliceA.put(msgB);
+        sliceA.flip();
+        msgB.limit(limit);
+        msgB.position(pos);
+        
+        ByteBuffer sliceB = ByteBuffer.allocate(msgB.remaining() + pos);
+        sliceB.put(msgB);
+        msgC.limit(pos);
+        sliceB.put(msgC);
+        sliceB.flip();
+        msgC.limit(limit);
+        
+        ArrayList<AMQDataBlock> frames = _decoder.decodeBuffer(sliceA);
+        assertEquals(1, frames.size());
+        frames = _decoder.decodeBuffer(sliceB);
+        assertEquals(1, frames.size());
+        frames = _decoder.decodeBuffer(msgC);
+        assertEquals(1, frames.size());
+        for (AMQDataBlock frame : frames)
+        {
+            if (frame instanceof AMQFrame)
+            {
+                assertEquals(HeartbeatBody.FRAME.getBodyFrame().getFrameType(), ((AMQFrame) frame).getBodyFrame().getFrameType());
+            }
+            else
+            {
+                fail("decode was not a frame");
+            }
+        }
+    }
+    
+}
diff --git a/java/common/src/test/java/org/apache/qpid/codec/MockAMQVersionAwareProtocolSession.java b/java/common/src/test/java/org/apache/qpid/codec/MockAMQVersionAwareProtocolSession.java
new file mode 100644
index 0000000000..c3b91d5d18
--- /dev/null
+++ b/java/common/src/test/java/org/apache/qpid/codec/MockAMQVersionAwareProtocolSession.java
@@ -0,0 +1,84 @@
+package org.apache.qpid.codec;
+
+import java.nio.ByteBuffer;
+
+import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.AMQDataBlock;
+import org.apache.qpid.framing.AMQMethodBody;
+import org.apache.qpid.framing.ContentBody;
+import org.apache.qpid.framing.ContentHeaderBody;
+import org.apache.qpid.framing.HeartbeatBody;
+import org.apache.qpid.framing.MethodRegistry;
+import org.apache.qpid.framing.ProtocolVersion;
+import org.apache.qpid.protocol.AMQVersionAwareProtocolSession;
+import org.apache.qpid.transport.Sender;
+
+public class MockAMQVersionAwareProtocolSession implements AMQVersionAwareProtocolSession
+{
+
+    public void contentBodyReceived(int channelId, ContentBody body) throws AMQException
+    {
+        // TODO Auto-generated method stub
+
+    }
+
+    public void contentHeaderReceived(int channelId, ContentHeaderBody body) throws AMQException
+    {
+        // TODO Auto-generated method stub
+
+    }
+
+    public MethodRegistry getMethodRegistry()
+    {
+        return MethodRegistry.getMethodRegistry(ProtocolVersion.v0_9);
+    }
+
+    public void heartbeatBodyReceived(int channelId, HeartbeatBody body) throws AMQException
+    {
+        // TODO Auto-generated method stub
+
+    }
+
+    public void init()
+    {
+        // TODO Auto-generated method stub
+
+    }
+
+    public void methodFrameReceived(int channelId, AMQMethodBody body) throws AMQException
+    {
+        // TODO Auto-generated method stub
+
+    }
+
+    public void setSender(Sender<ByteBuffer> sender)
+    {
+        // TODO Auto-generated method stub
+
+    }
+
+    public void writeFrame(AMQDataBlock frame)
+    {
+        // TODO Auto-generated method stub
+
+    }
+
+    public byte getProtocolMajorVersion()
+    {
+        // TODO Auto-generated method stub
+        return 0;
+    }
+
+    public byte getProtocolMinorVersion()
+    {
+        // TODO Auto-generated method stub
+        return 0;
+    }
+
+    public ProtocolVersion getProtocolVersion()
+    {
+        // TODO Auto-generated method stub
+        return null;
+    }
+
+}
diff --git a/java/common/src/test/java/org/apache/qpid/framing/abstraction/MessagePublishInfoImplTest.java b/java/common/src/test/java/org/apache/qpid/framing/abstraction/MessagePublishInfoImplTest.java
new file mode 100644
index 0000000000..3243136287
--- /dev/null
+++ b/java/common/src/test/java/org/apache/qpid/framing/abstraction/MessagePublishInfoImplTest.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.framing.abstraction;
+
+import junit.framework.TestCase;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.framing.abstraction.MessagePublishInfoImpl;
+
+public class MessagePublishInfoImplTest extends TestCase
+{
+    MessagePublishInfoImpl _mpi;
+    final AMQShortString _exchange = new AMQShortString("exchange");
+    final AMQShortString _routingKey = new AMQShortString("routingKey");
+
+    public void setUp()
+    {
+        _mpi = new MessagePublishInfoImpl(_exchange, true, true, _routingKey);
+    }
+
+    /** Test that we can update the exchange value. */
+    public void testExchange()
+    {
+        assertEquals(_exchange, _mpi.getExchange());
+        AMQShortString newExchange = new AMQShortString("newExchange");
+        //Check we can update the exchange
+        _mpi.setExchange(newExchange);
+        assertEquals(newExchange, _mpi.getExchange());
+        //Ensure that the new exchange doesn't equal the old one
+        assertFalse(_exchange.equals(_mpi.getExchange()));
+    }
+
+    /**
+     * Check that the immedate value is set correctly and defaulted correctly
+     */
+    public void testIsImmediate()
+    {
+        //Check that the set value is correct
+        assertTrue("Set value for immediate not as expected", _mpi.isImmediate());
+
+        MessagePublishInfoImpl mpi = new MessagePublishInfoImpl();
+
+        assertFalse("Default value for immediate should be false", mpi.isImmediate());
+
+        mpi.setImmediate(true);
+        
+        assertTrue("Updated value for immediate not as expected", mpi.isImmediate());
+
+    }
+
+    /**
+     * Check that the mandatory value is set correctly and defaulted correctly
+     */
+    public void testIsMandatory()
+    {
+        assertTrue("Set value for mandatory not as expected", _mpi.isMandatory());
+
+        MessagePublishInfoImpl mpi = new MessagePublishInfoImpl();
+
+        assertFalse("Default value for mandatory should be false", mpi.isMandatory());
+
+        mpi.setMandatory(true);
+
+        assertTrue("Updated value for mandatory not as expected", mpi.isMandatory());
+    }
+
+    /**
+     * Check that the routingKey value is perserved
+     */
+    public void testRoutingKey()
+    {
+        assertEquals(_routingKey, _mpi.getRoutingKey());
+        AMQShortString newRoutingKey = new AMQShortString("newRoutingKey");
+
+        //Check we can update the routingKey
+        _mpi.setRoutingKey(newRoutingKey);
+        assertEquals(newRoutingKey, _mpi.getRoutingKey());
+        //Ensure that the new routingKey doesn't equal the old one
+        assertFalse(_routingKey.equals(_mpi.getRoutingKey()));
+
+    }
+}
diff --git a/java/common/src/test/java/org/apache/qpid/pool/PoolingFilterTest.java b/java/common/src/test/java/org/apache/qpid/pool/PoolingFilterTest.java
deleted file mode 100644
index 6383d52298..0000000000
--- a/java/common/src/test/java/org/apache/qpid/pool/PoolingFilterTest.java
+++ /dev/null
@@ -1,111 +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.pool;
-
-import junit.framework.TestCase;
-import junit.framework.Assert;
-import org.apache.qpid.session.TestSession;
-import org.apache.mina.common.IoFilter;
-import org.apache.mina.common.IoSession;
-import org.apache.mina.common.IdleStatus;
-
-import java.util.concurrent.RejectedExecutionException;
-
-public class PoolingFilterTest extends TestCase
-{
-    private PoolingFilter _pool;
-    ReferenceCountingExecutorService _executorService;
-
-    public void setUp()
-    {
-
-        //Create Pool
-        _executorService = ReferenceCountingExecutorService.getInstance();
-        _executorService.acquireExecutorService();
-        _pool = PoolingFilter.createAynschWritePoolingFilter(_executorService,
-                                  "AsynchronousWriteFilter");
-
-    }
-
-    public void testRejectedExecution() throws Exception
-    {
-
-        TestSession testSession = new TestSession();
-        _pool.createNewJobForSession(testSession);
-        _pool.filterWrite(new NoOpFilter(), testSession, new IoFilter.WriteRequest("Message"));
-
-        //Shutdown the pool
-        _executorService.getPool().shutdownNow();
-
-        try
-        {
-
-            testSession = new TestSession();
-            _pool.createNewJobForSession(testSession);
-            //prior to fix for QPID-172 this would throw RejectedExecutionException
-            _pool.filterWrite(null, testSession, null);
-        }
-        catch (RejectedExecutionException rje)
-        {
-            Assert.fail("RejectedExecutionException should not occur after pool has shutdown:" + rje);
-        }
-    }
-
-    private static class NoOpFilter implements IoFilter.NextFilter
-    {
-
-        public void sessionOpened(IoSession session)
-        {
-        }
-
-        public void sessionClosed(IoSession session)
-        {
-        }
-
-        public void sessionIdle(IoSession session, IdleStatus status)
-        {
-        }
-
-        public void exceptionCaught(IoSession session, Throwable cause)
-        {
-        }
-
-        public void messageReceived(IoSession session, Object message)
-        {
-        }
-
-        public void messageSent(IoSession session, Object message)
-        {
-        }
-
-        public void filterWrite(IoSession session, IoFilter.WriteRequest writeRequest)
-        {
-        }
-
-        public void filterClose(IoSession session)
-        {
-        }
-
-        public void sessionCreated(IoSession session)
-        {
-        }
-    }
-}
diff --git a/java/common/src/test/java/org/apache/qpid/thread/ThreadFactoryTest.java b/java/common/src/test/java/org/apache/qpid/thread/ThreadFactoryTest.java
new file mode 100644
index 0000000000..9932633cb9
--- /dev/null
+++ b/java/common/src/test/java/org/apache/qpid/thread/ThreadFactoryTest.java
@@ -0,0 +1,46 @@
+package org.apache.qpid.thread;
+
+import junit.framework.TestCase;
+
+public class ThreadFactoryTest extends TestCase
+{
+    public void testThreadFactory()
+    {
+        Class threadFactoryClass = null;
+        try
+        {
+            threadFactoryClass = Class.forName(System.getProperty("qpid.thread_factory",
+                    "org.apache.qpid.thread.DefaultThreadFactory"));            
+        }
+        // If the thread factory class was wrong it will flagged way before it gets here.
+        catch(Exception e)
+        {            
+            fail("Invalid thread factory class");
+        }
+        
+        assertEquals(threadFactoryClass, Threading.getThreadFactory().getClass());
+    }
+    
+    public void testThreadCreate()
+    {
+        Runnable r = new Runnable(){
+          
+            public void run(){
+                
+            }            
+        };
+        
+        Thread t = null;
+        try
+        {
+            t = Threading.getThreadFactory().createThread(r,5);
+        }
+        catch(Exception e)
+        {
+            fail("Error creating thread using Qpid thread factory");
+        }
+        
+        assertNotNull(t);
+        assertEquals(5,t.getPriority());
+    }
+}
diff --git a/java/common/src/test/java/org/apache/qpid/transport/ConnectionTest.java b/java/common/src/test/java/org/apache/qpid/transport/ConnectionTest.java
index b9ca210483..8aa8c5f647 100644
--- a/java/common/src/test/java/org/apache/qpid/transport/ConnectionTest.java
+++ b/java/common/src/test/java/org/apache/qpid/transport/ConnectionTest.java
@@ -26,62 +26,148 @@ import org.apache.qpid.util.concurrent.Condition;
 
 import org.apache.qpid.transport.network.ConnectionBinding;
 import org.apache.qpid.transport.network.io.IoAcceptor;
-import org.apache.qpid.transport.network.io.IoTransport;
 import org.apache.qpid.transport.util.Logger;
+import org.apache.qpid.transport.util.Waiter;
 
 import junit.framework.TestCase;
 
-import java.util.Random;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Collections;
+import java.io.IOException;
+
+import static org.apache.qpid.transport.Option.*;
 
 /**
  * ConnectionTest
  */
 
-public class ConnectionTest extends TestCase
+public class ConnectionTest extends TestCase implements SessionListener
 {
 
     private static final Logger log = Logger.get(ConnectionTest.class);
 
     private int port;
+    private volatile boolean queue = false;
+    private List<MessageTransfer> messages = new ArrayList<MessageTransfer>();
+    private List<MessageTransfer> incoming = new ArrayList<MessageTransfer>();
+
+    private IoAcceptor _ioa = null;
+
 
     protected void setUp() throws Exception
     {
         super.setUp();
 
         port = AvailablePortFinder.getNextAvailable(12000);
+    }
 
-        ConnectionDelegate server = new ConnectionDelegate() {
-            public void init(Channel ch, ProtocolHeader hdr) {
-                ch.getConnection().close();
-            }
+    protected void tearDown() throws Exception
+    {
+        if (_ioa != null)
+        {
+            _ioa.close();
+        }
 
-            public SessionDelegate getSessionDelegate() {
-                return new SessionDelegate() {};
-            }
-            public void exception(Throwable t) {
-                log.error(t, "exception caught");
-            }
-            public void closed() {}
-        };
+        super.tearDown();
+    }
+
+    public void opened(Session ssn) {}
+
+    public void resumed(Session ssn) {}
+
+    public void message(final Session ssn, MessageTransfer xfr)
+    {
+        if (queue)
+        {
+            messages.add(xfr);
+            ssn.processed(xfr);
+            return;
+        }
 
-        IoAcceptor ioa = new IoAcceptor
-            ("localhost", port, new ConnectionBinding(server));
-        ioa.start();
+        String body = xfr.getBodyString();
+
+        if (body.startsWith("CLOSE"))
+        {
+            ssn.getConnection().close();
+        }
+        else if (body.startsWith("DELAYED_CLOSE"))
+        {
+            ssn.processed(xfr);
+            new Thread()
+            {
+                public void run()
+                {
+                    try
+                    {
+                        sleep(3000);
+                    }
+                    catch (InterruptedException e)
+                    {
+                        throw new RuntimeException(e);
+                    }
+                    ssn.getConnection().close();
+                }
+            }.start();
+        }
+        else if (body.startsWith("ECHO"))
+        {
+            int id = xfr.getId();
+            ssn.invoke(xfr);
+            ssn.processed(id);
+        }
+        else if (body.startsWith("SINK"))
+        {
+            ssn.processed(xfr);
+        }
+        else if (body.startsWith("DROP"))
+        {
+            // do nothing
+        }
+        else if (body.startsWith("EXCP"))
+        {
+            ExecutionException exc = new ExecutionException();
+            exc.setDescription("intentional exception for testing");
+            ssn.invoke(exc);
+            ssn.close();
+        }
+        else
+        {
+            throw new IllegalArgumentException
+                ("unrecognized message: " + body);
+        }
+    }
+
+    public void exception(Session ssn, SessionException exc)
+    {
+        throw exc;
+    }
+
+    public void closed(Session ssn) {}
+
+    private void send(Session ssn, String msg)
+    {
+        send(ssn, msg, false);
+    }
+
+    private void send(Session ssn, String msg, boolean sync)
+    {
+        ssn.messageTransfer
+            ("xxx", MessageAcceptMode.NONE, MessageAcquireMode.PRE_ACQUIRED,
+             null, msg, sync ? SYNC : NONE);
     }
 
     private Connection connect(final Condition closed)
     {
-        Connection conn = IoTransport.connect("localhost", port, new ConnectionDelegate()
+        Connection conn = new Connection();
+        conn.setConnectionListener(new ConnectionListener()
         {
-            public SessionDelegate getSessionDelegate()
+            public void opened(Connection conn) {}
+            public void exception(Connection conn, ConnectionException exc)
             {
-                return new SessionDelegate() {};
+                exc.printStackTrace();
             }
-            public void exception(Throwable t)
-            {
-                t.printStackTrace();
-            }
-            public void closed()
+            public void closed(Connection conn)
             {
                 if (closed != null)
                 {
@@ -89,27 +175,94 @@ public class ConnectionTest extends TestCase
                 }
             }
         });
-
-        conn.send(new ProtocolHeader(1, 0, 10));
+        conn.connect("localhost", port, null, "guest", "guest", false);
         return conn;
     }
 
+    public void testProtocolNegotiationExceptionOverridesCloseException() throws Exception
+    {
+        // Force os.name to be windows to exercise code in IoReceiver
+        // that looks for the value of os.name
+        System.setProperty("os.name","windows");
+
+        // Start server as 0-9 to froce a ProtocolVersionException
+        startServer(new ProtocolHeader(1, 0, 9));
+        
+        Condition closed = new Condition();
+
+        try
+        {
+            connect(closed);
+            fail("ProtocolVersionException expected");
+        }
+        catch (ProtocolVersionException pve)
+        {
+            //Expected code path
+        }
+        catch (Exception e)
+        {
+            fail("ProtocolVersionException expected. Got:" + e.getMessage());
+        }
+    }
+
+    private void startServer()
+    {
+        startServer(new ProtocolHeader(1, 0, 10));
+    }
+
+    private void startServer(final ProtocolHeader protocolHeader)
+    {
+        ConnectionDelegate server = new ServerDelegate()
+        {
+            @Override
+            public void init(Connection conn, ProtocolHeader hdr)
+            {
+                conn.send(protocolHeader);
+                List<Object> utf8 = new ArrayList<Object>();
+                utf8.add("utf8");
+                conn.connectionStart(null, Collections.EMPTY_LIST, utf8);
+            }
+
+            @Override
+            public Session getSession(Connection conn, SessionAttach atc)
+            {
+                Session ssn = super.getSession(conn, atc);
+                ssn.setSessionListener(ConnectionTest.this);
+                return ssn;
+            }
+        };
+
+        try
+        {
+            _ioa = new IoAcceptor("localhost", port, ConnectionBinding.get(server));
+        }
+        catch (IOException e)
+        {
+            e.printStackTrace();
+            fail("Unable to start Server for test due to:" + e.getMessage());
+        }
+
+        _ioa.start();
+    }
+
     public void testClosedNotificationAndWriteToClosed() throws Exception
     {
+        startServer();
+
         Condition closed = new Condition();
         Connection conn = connect(closed);
+
+        Session ssn = conn.createSession(1);
+        send(ssn, "CLOSE");
+
         if (!closed.get(3000))
         {
             fail("never got notified of connection close");
         }
 
-        Channel ch = conn.getChannel(0);
-        Session ssn = new Session("test".getBytes());
-        ssn.attach(ch);
-
         try
         {
-            ssn.sessionAttach(ssn.getName());
+            conn.connectionCloseOk();
             fail("writing to a closed socket succeeded");
         }
         catch (TransportException e)
@@ -118,4 +271,176 @@ public class ConnectionTest extends TestCase
         }
     }
 
+    class FailoverConnectionListener implements ConnectionListener
+    {
+        public void opened(Connection conn) {}
+
+        public void exception(Connection conn, ConnectionException e)
+        {
+            throw e;
+        }
+
+        public void closed(Connection conn)
+        {
+            queue = true;
+            conn.connect("localhost", port, null, "guest", "guest");
+            conn.resume();
+        }
+    }
+
+    class TestSessionListener implements SessionListener
+    {
+        public void opened(Session s) {}
+        public void resumed(Session s) {}
+        public void exception(Session s, SessionException e) {}
+        public void message(Session s, MessageTransfer xfr)
+        {
+            synchronized (incoming)
+            {
+                incoming.add(xfr);
+                incoming.notifyAll();
+            }
+
+            s.processed(xfr);
+        }
+        public void closed(Session s) {}
+    }
+
+    public void testResumeNonemptyReplayBuffer() throws Exception
+    {
+        startServer();
+
+        Connection conn = new Connection();
+        conn.setConnectionListener(new FailoverConnectionListener());
+        conn.connect("localhost", port, null, "guest", "guest");
+        Session ssn = conn.createSession(1);
+        ssn.setSessionListener(new TestSessionListener());
+
+        send(ssn, "SINK 0");
+        send(ssn, "ECHO 1");
+        send(ssn, "ECHO 2");
+
+        ssn.sync();
+
+        String[] msgs = { "DROP 3", "DROP 4", "DROP 5", "CLOSE 6", "SINK 7" };
+        for (String m : msgs)
+        {
+            send(ssn, m);
+        }
+
+        ssn.sync();
+
+        assertEquals(msgs.length, messages.size());
+        for (int i = 0; i < msgs.length; i++)
+        {
+            assertEquals(msgs[i], messages.get(i).getBodyString());
+        }
+
+        queue = false;
+
+        send(ssn, "ECHO 8");
+        send(ssn, "ECHO 9");
+
+        synchronized (incoming)
+        {
+            Waiter w = new Waiter(incoming, 30000);
+            while (w.hasTime() && incoming.size() < 4)
+            {
+                w.await();
+            }
+
+            assertEquals(4, incoming.size());
+            assertEquals("ECHO 1", incoming.get(0).getBodyString());
+            assertEquals(0, incoming.get(0).getId());
+            assertEquals("ECHO 2", incoming.get(1).getBodyString());
+            assertEquals(1, incoming.get(1).getId());
+            assertEquals("ECHO 8", incoming.get(2).getBodyString());
+            assertEquals(0, incoming.get(0).getId());
+            assertEquals("ECHO 9", incoming.get(3).getBodyString());
+            assertEquals(1, incoming.get(1).getId());
+        }
+    }
+
+    public void testResumeEmptyReplayBuffer() throws InterruptedException
+    {
+        startServer();
+
+        Connection conn = new Connection();
+        conn.setConnectionListener(new FailoverConnectionListener());
+        conn.connect("localhost", port, null, "guest", "guest");
+        Session ssn = conn.createSession(1);
+        ssn.setSessionListener(new TestSessionListener());
+
+        send(ssn, "SINK 0");
+        send(ssn, "SINK 1");
+        send(ssn, "DELAYED_CLOSE 2");
+        ssn.sync();
+        Thread.sleep(6000);
+        send(ssn, "SINK 3");
+        ssn.sync();
+        System.out.println(messages);
+        assertEquals(1, messages.size());
+        assertEquals("SINK 3", messages.get(0).getBodyString());
+    }
+
+    public void testFlushExpected() throws InterruptedException
+    {
+        startServer();
+
+        Connection conn = new Connection();
+        conn.connect("localhost", port, null, "guest", "guest");
+        Session ssn = conn.createSession();
+        ssn.sessionFlush(EXPECTED);
+        send(ssn, "SINK 0");
+        ssn.sessionFlush(EXPECTED);
+        send(ssn, "SINK 1");
+        ssn.sync();
+    }
+
+    public void testHeartbeat()
+    {
+        startServer();
+        Connection conn = new Connection();
+        conn.connect("localhost", port, null, "guest", "guest");
+        conn.connectionHeartbeat();
+        conn.close();
+    }
+
+    public void testExecutionExceptionInvoke() throws Exception
+    {
+        startServer();
+
+        Connection conn = new Connection();
+        conn.connect("localhost", port, null, "guest", "guest");
+        Session ssn = conn.createSession();
+        send(ssn, "EXCP 0");
+        Thread.sleep(3000);
+        try
+        {
+            send(ssn, "SINK 1");
+        }
+        catch (SessionException exc)
+        {
+            assertNotNull(exc.getException());
+        }
+    }
+
+    public void testExecutionExceptionSync() throws Exception
+    {
+        startServer();
+
+        Connection conn = new Connection();
+        conn.connect("localhost", port, null, "guest", "guest");
+        Session ssn = conn.createSession();
+        send(ssn, "EXCP 0", true);
+        try
+        {
+            ssn.sync();
+        }
+        catch (SessionException exc)
+        {
+            assertNotNull(exc.getException());
+        }
+    }
+
 }
diff --git a/java/common/src/test/java/org/apache/qpid/transport/GenTest.java b/java/common/src/test/java/org/apache/qpid/transport/GenTest.java
new file mode 100644
index 0000000000..512a0a29a6
--- /dev/null
+++ b/java/common/src/test/java/org/apache/qpid/transport/GenTest.java
@@ -0,0 +1,44 @@
+/*
+ *
+ * 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.transport;
+
+import junit.framework.TestCase;
+
+/**
+ * GenTest
+ *
+ */
+
+public class GenTest extends TestCase
+{
+
+    public void testBooleans()
+    {
+        QueueDeclare qd = new QueueDeclare().queue("test-queue").durable(false);
+        assertEquals(qd.getQueue(), "test-queue");
+        assertFalse("durable should be false", qd.getDurable());
+        qd.setDurable(true);
+        assertTrue("durable should be true", qd.getDurable());
+        qd.setDurable(false);
+        assertFalse("durable should be false again", qd.getDurable());
+    }
+
+}
diff --git a/java/common/src/test/java/org/apache/qpid/transport/TestNetworkDriver.java b/java/common/src/test/java/org/apache/qpid/transport/TestNetworkDriver.java
new file mode 100644
index 0000000000..a4c4b59cdd
--- /dev/null
+++ b/java/common/src/test/java/org/apache/qpid/transport/TestNetworkDriver.java
@@ -0,0 +1,122 @@
+/*
+ *
+ * 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.transport;
+
+import java.net.BindException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+import org.apache.qpid.protocol.ProtocolEngine;
+import org.apache.qpid.protocol.ProtocolEngineFactory;
+import org.apache.qpid.ssl.SSLContextFactory;
+
+/**
+ * Test implementation of IoSession, which is required for some tests. Methods not being used are not implemented,
+ * so if this class is being used and some methods are to be used, then please update those.
+ */
+public class TestNetworkDriver implements NetworkDriver
+{
+    private final ConcurrentMap attributes = new ConcurrentHashMap();
+    private String _remoteAddress = "127.0.0.1";
+    private String _localAddress = "127.0.0.1";
+    private int _port = 1;
+
+    public TestNetworkDriver()
+    {
+    }
+
+    public void setRemoteAddress(String string)
+    {
+        this._remoteAddress = string;
+    }
+
+    public void setPort(int _port)
+    {
+        this._port = _port;
+    }
+
+    public int getPort()
+    {
+        return _port;
+    }
+
+    public void bind(int port, InetAddress[] addresses, ProtocolEngineFactory protocolFactory,
+            NetworkDriverConfiguration config, SSLContextFactory sslFactory) throws BindException
+    {
+        
+    }
+
+    public SocketAddress getLocalAddress()
+    {
+        return new InetSocketAddress(_localAddress, _port);
+    }
+
+    public SocketAddress getRemoteAddress()
+    {
+        return new InetSocketAddress(_remoteAddress, _port);
+    }
+
+    public void open(int port, InetAddress destination, ProtocolEngine engine, NetworkDriverConfiguration config,
+            SSLContextFactory sslFactory) throws OpenException
+    {
+        
+    }
+
+    public void setMaxReadIdle(int idleTime)
+    {
+        
+    }
+
+    public void setMaxWriteIdle(int idleTime)
+    {
+        
+    }
+
+    public void close()
+    {
+           
+    }
+
+    public void flush()
+    {
+        
+    }
+
+    public void send(ByteBuffer msg)
+    {
+        
+    }
+
+    public void setIdleTimeout(long l)
+    {
+        
+    }
+
+    public void setLocalAddress(String localAddress)
+    {
+        _localAddress = localAddress;
+    }
+
+}
diff --git a/java/common/src/test/java/org/apache/qpid/transport/codec/BBEncoderTest.java b/java/common/src/test/java/org/apache/qpid/transport/codec/BBEncoderTest.java
new file mode 100644
index 0000000000..79bf184fe2
--- /dev/null
+++ b/java/common/src/test/java/org/apache/qpid/transport/codec/BBEncoderTest.java
@@ -0,0 +1,47 @@
+/*
+ *
+ * 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.transport.codec;
+
+import junit.framework.TestCase;
+
+import java.nio.ByteBuffer;
+
+/**
+ * BBEncoderTest
+ *
+ */
+
+public class BBEncoderTest extends TestCase
+{
+
+    public void testGrow()
+    {
+        BBEncoder enc = new BBEncoder(4);
+        enc.writeInt32(0xDEADBEEF);
+        ByteBuffer buf = enc.buffer();
+        assertEquals(0xDEADBEEF, buf.getInt(0));
+        enc.writeInt32(0xBEEFDEAD);
+        buf = enc.buffer();
+        assertEquals(0xDEADBEEF, buf.getInt(0));
+        assertEquals(0xBEEFDEAD, buf.getInt(4));
+    }
+
+}
diff --git a/java/common/src/test/java/org/apache/qpid/transport/network/mina/MINANetworkDriverTest.java b/java/common/src/test/java/org/apache/qpid/transport/network/mina/MINANetworkDriverTest.java
new file mode 100644
index 0000000000..5af07d9735
--- /dev/null
+++ b/java/common/src/test/java/org/apache/qpid/transport/network/mina/MINANetworkDriverTest.java
@@ -0,0 +1,491 @@
+/*
+ *
+ * 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.transport.network.mina;
+
+import java.net.BindException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.net.UnknownHostException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import junit.framework.TestCase;
+
+import org.apache.qpid.framing.AMQDataBlock;
+import org.apache.qpid.protocol.ProtocolEngine;
+import org.apache.qpid.protocol.ProtocolEngineFactory;
+import org.apache.qpid.transport.NetworkDriver;
+import org.apache.qpid.transport.OpenException;
+
+public class MINANetworkDriverTest extends TestCase
+{
+ 
+    private static final String TEST_DATA = "YHALOTHAR";
+    private static final int TEST_PORT = 2323;
+    private NetworkDriver _server;
+    private NetworkDriver _client;
+    private CountingProtocolEngine _countingEngine; // Keeps a count of how many bytes it's read
+    private Exception _thrownEx;
+
+    @Override
+    public void setUp()
+    {
+        _server = new MINANetworkDriver();
+        _client = new MINANetworkDriver();
+        _thrownEx = null;
+        _countingEngine = new CountingProtocolEngine();
+    }
+
+    @Override
+    public void tearDown()
+    {
+        if (_server != null)
+        {
+            _server.close();
+        }
+
+        if (_client != null)
+        {
+            _client.close();
+        }
+    }
+    
+    /**
+     * Tests that a socket can't be opened if a driver hasn't been bound
+     * to the port and can be opened if a driver has been bound.
+     * @throws BindException
+     * @throws UnknownHostException
+     * @throws OpenException 
+     */
+    public void testBindOpen() throws BindException, UnknownHostException, OpenException  
+    {
+        try
+        {
+            _client.open(TEST_PORT, InetAddress.getLocalHost(), _countingEngine, null, null);
+        } 
+        catch (OpenException e)
+        {
+            _thrownEx = e;
+        }
+
+        assertNotNull("Open should have failed since no engine bound", _thrownEx);        
+        
+        _server.bind(TEST_PORT, null, new EchoProtocolEngineSingletonFactory(), null, null);
+        
+        _client.open(TEST_PORT, InetAddress.getLocalHost(), _countingEngine, null, null);
+    } 
+    
+    /**
+     * Tests that a socket can't be opened after a bound NetworkDriver has been closed
+     * @throws BindException
+     * @throws UnknownHostException
+     * @throws OpenException
+     */
+    public void testBindOpenCloseOpen() throws BindException, UnknownHostException, OpenException  
+    {
+        _server.bind(TEST_PORT, null, new EchoProtocolEngineSingletonFactory(), null, null);
+        _client.open(TEST_PORT, InetAddress.getLocalHost(), _countingEngine, null, null);
+        _client.close();
+        _server.close();
+        
+        try
+        {
+            _client.open(TEST_PORT, InetAddress.getLocalHost(), _countingEngine, null, null);
+        } 
+        catch (OpenException e)
+        {
+            _thrownEx = e;
+        }
+        assertNotNull("Open should have failed", _thrownEx);
+    }
+    
+    /**
+     * Checks that the right exception is thrown when binding a NetworkDriver to an already
+     * existing socket. 
+     */
+    public void testBindPortInUse() 
+    {
+        try
+        {
+            _server.bind(TEST_PORT, null, new EchoProtocolEngineSingletonFactory(), null, null);
+        }
+        catch (BindException e)
+        {
+            fail("First bind should not fail");
+        }
+        
+        try
+        {
+            _client.bind(TEST_PORT, null, new EchoProtocolEngineSingletonFactory(), null, null);
+        }
+        catch (BindException e)
+        {
+            _thrownEx = e;
+        }
+        assertNotNull("Second bind should throw BindException", _thrownEx);
+    } 
+    
+    /**
+     * tests that bytes sent on a network driver are received at the other end
+     * 
+     * @throws UnknownHostException
+     * @throws OpenException
+     * @throws InterruptedException 
+     * @throws BindException 
+     */
+    public void testSend() throws UnknownHostException, OpenException, InterruptedException, BindException 
+    {
+        // Open a connection from a counting engine to an echo engine
+        _server.bind(TEST_PORT, null,  new EchoProtocolEngineSingletonFactory(), null, null);
+        _client.open(TEST_PORT, InetAddress.getLocalHost(), _countingEngine, null, null);
+        
+        // Tell the counting engine how much data we're sending
+        _countingEngine.setNewLatch(TEST_DATA.getBytes().length);
+        
+        // Send the data and wait for up to 2 seconds to get it back 
+        _client.send(ByteBuffer.wrap(TEST_DATA.getBytes()));
+        _countingEngine.getLatch().await(2, TimeUnit.SECONDS);
+        
+        // Check what we got
+        assertEquals("Wrong amount of data recieved", TEST_DATA.getBytes().length, _countingEngine.getReadBytes());
+    } 
+    
+    /**
+     * Opens a connection with a low read idle and check that it gets triggered
+     * @throws BindException 
+     * @throws OpenException 
+     * @throws UnknownHostException 
+     * 
+     */
+    public void testSetReadIdle() throws BindException, UnknownHostException, OpenException 
+    {
+        // Open a connection from a counting engine to an echo engine
+        _server.bind(TEST_PORT, null,  new EchoProtocolEngineSingletonFactory(), null, null);
+        _client.open(TEST_PORT, InetAddress.getLocalHost(), _countingEngine, null, null);
+        assertFalse("Reader should not have been idle", _countingEngine.getReaderHasBeenIdle());
+        _client.setMaxReadIdle(1);
+        sleepForAtLeast(1500);
+        assertTrue("Reader should have been idle", _countingEngine.getReaderHasBeenIdle());
+    } 
+    
+    /**
+     * Opens a connection with a low write idle and check that it gets triggered
+     * @throws BindException 
+     * @throws OpenException 
+     * @throws UnknownHostException 
+     * 
+     */
+    public void testSetWriteIdle() throws BindException, UnknownHostException, OpenException 
+    {
+        // Open a connection from a counting engine to an echo engine
+        _server.bind(TEST_PORT, null,  new EchoProtocolEngineSingletonFactory(), null, null);
+        _client.open(TEST_PORT, InetAddress.getLocalHost(), _countingEngine, null, null);
+        assertFalse("Reader should not have been idle", _countingEngine.getWriterHasBeenIdle());
+        _client.setMaxWriteIdle(1);
+        sleepForAtLeast(1500);
+        assertTrue("Reader should have been idle", _countingEngine.getWriterHasBeenIdle());
+    } 
+    
+    
+    /**
+     * Creates and then closes a connection from client to server and checks that the server
+     * has its closed() method called. Then creates a new client and closes the server to check
+     * that the client has its closed() method called.  
+     * @throws BindException
+     * @throws UnknownHostException
+     * @throws OpenException
+     */
+    public void testClosed() throws BindException, UnknownHostException, OpenException 
+    {
+        // Open a connection from a counting engine to an echo engine
+        EchoProtocolEngineSingletonFactory factory = new EchoProtocolEngineSingletonFactory();
+        _server.bind(TEST_PORT, null, factory, null, null);
+        _client.open(TEST_PORT, InetAddress.getLocalHost(), _countingEngine, null, null);
+        EchoProtocolEngine serverEngine = null; 
+        while (serverEngine == null)
+        {
+            serverEngine = factory.getEngine();
+            if (serverEngine == null)
+            {
+                try
+                {
+                    Thread.sleep(10);
+                }
+                catch (InterruptedException e)
+                {
+                }
+            }
+        }
+        assertFalse("Server should not have been closed", serverEngine.getClosed());
+        serverEngine.setNewLatch(1);
+        _client.close();
+        try
+        {
+            serverEngine.getLatch().await(2, TimeUnit.SECONDS);
+        }
+        catch (InterruptedException e)
+        {
+        }
+        assertTrue("Server should have been closed", serverEngine.getClosed());
+
+        _client.open(TEST_PORT, InetAddress.getLocalHost(), _countingEngine, null, null);
+        _countingEngine.setClosed(false);
+        assertFalse("Client should not have been closed", _countingEngine.getClosed());
+        _countingEngine.setNewLatch(1);
+        _server.close();
+        try
+        {
+            _countingEngine.getLatch().await(2, TimeUnit.SECONDS);
+        }
+        catch (InterruptedException e)
+        {
+        }
+        assertTrue("Client should have been closed", _countingEngine.getClosed());
+    } 
+
+    /**
+     * Create a connection and instruct the client to throw an exception when it gets some data
+     * and that the latch gets counted down. 
+     * @throws BindException
+     * @throws UnknownHostException
+     * @throws OpenException
+     * @throws InterruptedException
+     */
+    public void testExceptionCaught() throws BindException, UnknownHostException, OpenException, InterruptedException 
+    {
+        _server.bind(TEST_PORT, null,  new EchoProtocolEngineSingletonFactory(), null, null);
+        _client.open(TEST_PORT, InetAddress.getLocalHost(), _countingEngine, null, null);
+
+
+        assertEquals("Exception should not have been thrown", 1, 
+                _countingEngine.getExceptionLatch().getCount());
+        _countingEngine.setErrorOnNextRead(true);
+        _countingEngine.setNewLatch(TEST_DATA.getBytes().length);
+        _client.send(ByteBuffer.wrap(TEST_DATA.getBytes()));
+        _countingEngine.getExceptionLatch().await(2, TimeUnit.SECONDS);
+        assertEquals("Exception should have been thrown", 0, 
+                _countingEngine.getExceptionLatch().getCount());
+    } 
+    
+    /**
+     * Opens a connection and checks that the remote address is the one that was asked for
+     * @throws BindException
+     * @throws UnknownHostException
+     * @throws OpenException
+     */
+    public void testGetRemoteAddress() throws BindException, UnknownHostException, OpenException
+    {
+        _server.bind(TEST_PORT, null,  new EchoProtocolEngineSingletonFactory(), null, null);
+        _client.open(TEST_PORT, InetAddress.getLocalHost(), _countingEngine, null, null);
+        assertEquals(new InetSocketAddress(InetAddress.getLocalHost(), TEST_PORT),
+                     _client.getRemoteAddress());
+    }
+
+    private class EchoProtocolEngineSingletonFactory implements ProtocolEngineFactory
+    {
+        EchoProtocolEngine _engine = null;
+        
+        public ProtocolEngine newProtocolEngine(NetworkDriver driver)
+        {
+            if (_engine == null)
+            {
+                _engine = new EchoProtocolEngine();
+                _engine.setNetworkDriver(driver);
+            }
+            return getEngine();
+        }
+        
+        public EchoProtocolEngine getEngine()
+        {
+            return _engine;
+        }
+    }
+    
+    public class CountingProtocolEngine implements ProtocolEngine
+    {
+
+        protected NetworkDriver _driver;
+        public ArrayList<ByteBuffer> _receivedBytes = new ArrayList<ByteBuffer>();
+        private int _readBytes;
+        private CountDownLatch _latch = new CountDownLatch(0);
+        private boolean _readerHasBeenIdle;
+        private boolean _writerHasBeenIdle;
+        private boolean _closed = false;
+        private boolean _nextReadErrors = false;
+        private CountDownLatch _exceptionLatch = new CountDownLatch(1);
+        
+        public void closed()
+        {
+            setClosed(true);
+            _latch.countDown();
+        }
+
+        public void setErrorOnNextRead(boolean b)
+        {
+            _nextReadErrors = b;
+        }
+
+        public void setNewLatch(int length)
+        {
+            _latch = new CountDownLatch(length);
+        }
+
+        public long getReadBytes()
+        {
+            return _readBytes;
+        }
+
+        public SocketAddress getRemoteAddress()
+        {
+            if (_driver != null)
+            {
+                return _driver.getRemoteAddress();
+            } 
+            else
+            {
+                return null;
+            }
+        }
+        
+        public SocketAddress getLocalAddress()
+        {            
+            if (_driver != null)
+            {
+                return _driver.getLocalAddress();
+            } 
+            else
+            {
+                return null;
+            }
+        }
+
+        public long getWrittenBytes()
+        {
+            return 0;
+        }
+
+        public void readerIdle()
+        {
+            _readerHasBeenIdle = true;
+        }
+
+        public void setNetworkDriver(NetworkDriver driver)
+        {
+            _driver = driver;
+        }
+
+        public void writeFrame(AMQDataBlock frame)
+        {
+            
+        }
+
+        public void writerIdle()
+        {
+           _writerHasBeenIdle = true;
+        }
+
+        public void exception(Throwable t)
+        {
+            _exceptionLatch.countDown();
+        }
+
+        public CountDownLatch getExceptionLatch()
+        {
+            return _exceptionLatch;
+        }
+        
+        public void received(ByteBuffer msg)
+        {
+            // increment read bytes and count down the latch for that many
+            int bytes = msg.remaining();
+            _readBytes += bytes;
+            for (int i = 0; i < bytes; i++)
+            {
+                _latch.countDown();
+            }
+            
+            // Throw an error if we've been asked too, but we can still count
+            if (_nextReadErrors)
+            {
+                throw new RuntimeException("Was asked to error");
+            }
+        }
+        
+        public CountDownLatch getLatch()
+        {
+            return _latch;
+        }
+
+        public boolean getWriterHasBeenIdle()
+        {
+            return _writerHasBeenIdle;
+        }
+
+        public boolean getReaderHasBeenIdle()
+        {
+            return _readerHasBeenIdle;
+        }
+
+        public void setClosed(boolean _closed)
+        {
+            this._closed = _closed;
+        }
+
+        public boolean getClosed()
+        {
+            return _closed;
+        }
+
+    }
+
+    private class EchoProtocolEngine extends CountingProtocolEngine
+    {
+
+        public void received(ByteBuffer msg)
+        {
+            super.received(msg);
+            msg.rewind();
+            _driver.send(msg);
+        }
+    }
+    
+    public static void sleepForAtLeast(long period)
+    {
+        long start = System.currentTimeMillis();
+        long timeLeft = period;
+        while (timeLeft > 0)
+        {
+            try
+            {
+                Thread.sleep(timeLeft);
+            }
+            catch (InterruptedException e)
+            {
+                // Ignore it
+            }
+            timeLeft = period - (System.currentTimeMillis() - start);
+        }
+    }
+}
\ No newline at end of file
diff --git a/java/common/src/test/java/org/apache/qpid/util/FileUtilsTest.java b/java/common/src/test/java/org/apache/qpid/util/FileUtilsTest.java
new file mode 100644
index 0000000000..7eba5f092e
--- /dev/null
+++ b/java/common/src/test/java/org/apache/qpid/util/FileUtilsTest.java
@@ -0,0 +1,612 @@
+/*
+ *
+ * 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.util;
+
+import junit.framework.TestCase;
+
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.util.List;
+
+public class FileUtilsTest extends TestCase
+{
+    private static final String COPY = "-Copy";
+    private static final String SUB = "-Sub";
+
+    /**
+     * Additional test for the copy method.
+     * Ensures that the directory count did increase by more than 1 after the copy.
+     */
+    public void testCopyFile()
+    {
+        final String TEST_DATA = "FileUtilsTest-testCopy-TestDataTestDataTestDataTestDataTestDataTestData";
+        String fileName = "FileUtilsTest-testCopy";
+        String fileNameCopy = fileName + COPY;
+
+        File[] beforeCopyFileList = null;
+
+        //Create initial file
+        File test = createTestFile(fileName, TEST_DATA);
+
+        try
+        {
+            //Check number of files before copy
+            beforeCopyFileList = test.getAbsoluteFile().getParentFile().listFiles();
+            int beforeCopy = beforeCopyFileList.length;
+
+            //Perform Copy
+            File destination = new File(fileNameCopy);
+            FileUtils.copy(test, destination);
+            //Ensure the JVM cleans up if cleanup failues
+            destination.deleteOnExit();
+
+            //Retrieve counts after copy
+            int afterCopy = test.getAbsoluteFile().getParentFile().listFiles().length;
+
+            int afterCopyFromCopy = new File(fileNameCopy).getAbsoluteFile().getParentFile().listFiles().length;
+
+            // Validate the copy counts
+            assertEquals("The file listing from the original and the copy differ in length.", afterCopy, afterCopyFromCopy);
+            assertEquals("The number of files did not increase.", beforeCopy + 1, afterCopy);
+            assertEquals("The number of files did not increase.", beforeCopy + 1, afterCopyFromCopy);
+
+            //Validate copy
+            // Load content
+            String copiedFileContent = FileUtils.readFileAsString(fileNameCopy);
+            assertEquals(TEST_DATA, copiedFileContent);
+        }
+        finally // Ensure clean
+        {
+            //Clean up
+            assertTrue("Unable to cleanup", FileUtils.deleteFile(fileNameCopy));
+
+            //Check file list after cleanup
+            File[] afterCleanup = new File(test.getAbsoluteFile().getParent()).listFiles();
+            checkFileLists(beforeCopyFileList, afterCleanup);
+
+            //Remove original file
+            assertTrue("Unable to cleanup", test.delete());
+        }
+    }
+
+    /**
+     * Create and Copy the following structure:
+     *
+     * testDirectory --+
+     * +-- testSubDirectory --+
+     * +-- testSubFile
+     * +-- File
+     *
+     * to testDirectory-Copy
+     *
+     * Validate that the file count in the copy is correct and contents of the copied files is correct.
+     */
+    public void testCopyRecursive()
+    {
+        final String TEST_DATA = "FileUtilsTest-testDirectoryCopy-TestDataTestDataTestDataTestDataTestDataTestData";
+        String fileName = "FileUtilsTest-testCopy";
+        String TEST_DIR = "testDirectoryCopy";
+
+        //Create Initial Structure
+        File testDir = new File(TEST_DIR);
+
+        //Check number of files before copy
+        File[] beforeCopyFileList = testDir.getAbsoluteFile().getParentFile().listFiles();
+
+        try
+        {
+            //Create Directories
+            assertTrue("Test directory already exists cannot test.", !testDir.exists());
+
+            if (!testDir.mkdir())
+            {
+                fail("Unable to make test Directory");
+            }
+
+            File testSubDir = new File(TEST_DIR + File.separator + TEST_DIR + SUB);
+            if (!testSubDir.mkdir())
+            {
+                fail("Unable to make test sub Directory");
+            }
+
+            //Create Files
+            createTestFile(testDir.toString() + File.separator + fileName, TEST_DATA);
+            createTestFile(testSubDir.toString() + File.separator + fileName + SUB, TEST_DATA);
+
+            //Ensure the JVM cleans up if cleanup failues
+            testSubDir.deleteOnExit();
+            testDir.deleteOnExit();
+
+            //Perform Copy
+            File copyDir = new File(testDir.toString() + COPY);
+            try
+            {
+                FileUtils.copyRecursive(testDir, copyDir);
+            }
+            catch (FileNotFoundException e)
+            {
+                fail(e.getMessage());
+            }
+            catch (FileUtils.UnableToCopyException e)
+            {
+                fail(e.getMessage());
+            }
+
+            //Validate Copy
+            assertEquals("Copied directory should only have one file and one directory in it.", 2, copyDir.listFiles().length);
+
+            //Validate Copy File Contents
+            String copiedFileContent = FileUtils.readFileAsString(copyDir.toString() + File.separator + fileName);
+            assertEquals(TEST_DATA, copiedFileContent);
+
+            //Validate Name of Sub Directory
+            assertTrue("Expected subdirectory is not a directory", new File(copyDir.toString() + File.separator + TEST_DIR + SUB).isDirectory());
+
+            //Assert that it contains only one item
+            assertEquals("Copied sub directory should only have one directory in it.", 1, new File(copyDir.toString() + File.separator + TEST_DIR + SUB).listFiles().length);
+
+            //Validate content of Sub file
+            copiedFileContent = FileUtils.readFileAsString(copyDir.toString() + File.separator + TEST_DIR + SUB + File.separator + fileName + SUB);
+            assertEquals(TEST_DATA, copiedFileContent);
+        }
+        finally
+        {
+            //Clean up source and copy directory.
+            assertTrue("Unable to cleanup", FileUtils.delete(testDir, true));
+            assertTrue("Unable to cleanup", FileUtils.delete(new File(TEST_DIR + COPY), true));
+
+            //Check file list after cleanup
+            File[] afterCleanup = testDir.getAbsoluteFile().getParentFile().listFiles();
+            checkFileLists(beforeCopyFileList, afterCleanup);
+        }
+    }
+
+    /**
+     * Helper method to create a test file with a string content
+     *
+     * @param fileName  The fileName to use in the creation
+     * @param test_data The data to store in the file
+     *
+     * @return The File reference
+     */
+    private File createTestFile(String fileName, String test_data)
+    {
+        File test = new File(fileName);
+
+        try
+        {
+            test.createNewFile();
+            //Ensure the JVM cleans up if cleanup failues
+            test.deleteOnExit();
+        }
+        catch (IOException e)
+        {
+            fail(e.getMessage());
+        }
+
+        BufferedWriter writer = null;
+        try
+        {
+            writer = new BufferedWriter(new FileWriter(test));
+            try
+            {
+                writer.write(test_data);
+            }
+            catch (IOException e)
+            {
+                fail(e.getMessage());
+            }
+        }
+        catch (IOException e)
+        {
+            fail(e.getMessage());
+        }
+        finally
+        {
+            try
+            {
+                if (writer != null)
+                {
+                    writer.close();
+                }
+            }
+            catch (IOException e)
+            {
+                fail(e.getMessage());
+            }
+        }
+
+        return test;
+    }
+
+    /** Test that deleteFile only deletes the specified file */
+    public void testDeleteFile()
+    {
+        File test = new File("FileUtilsTest-testDelete");
+        //Record file count in parent directory to check it is not changed by delete
+        String path = test.getAbsolutePath();
+        File[] filesBefore = new File(path.substring(0, path.lastIndexOf(File.separator))).listFiles();
+        int fileCountBefore = filesBefore.length;
+
+        try
+        {
+            test.createNewFile();
+            //Ensure the JVM cleans up if cleanup failues
+            test.deleteOnExit();
+        }
+        catch (IOException e)
+        {
+            fail(e.getMessage());
+        }
+
+        assertTrue("File does not exists", test.exists());
+        assertTrue("File is not a file", test.isFile());
+
+        //Check that file creation can be seen on disk
+        int fileCountCreated = new File(path.substring(0, path.lastIndexOf(File.separator))).listFiles().length;
+        assertEquals("File creation was no registered", fileCountBefore + 1, fileCountCreated);
+
+        //Perform Delete
+        assertTrue("Unable to cleanup", FileUtils.deleteFile("FileUtilsTest-testDelete"));
+
+        assertTrue("File exists after delete", !test.exists());
+
+        //Check that after deletion the file count is now accurate
+        File[] filesAfter = new File(path.substring(0, path.lastIndexOf(File.separator))).listFiles();
+        int fileCountAfter = filesAfter.length;
+        assertEquals("File creation was no registered", fileCountBefore, fileCountAfter);
+
+        checkFileLists(filesBefore, filesAfter);
+    }
+
+    public void testDeleteNonExistentFile()
+    {
+        File test = new File("FileUtilsTest-testDelete-" + System.currentTimeMillis());
+
+        assertTrue("File exists", !test.exists());
+        assertFalse("File is a directory", test.isDirectory());
+
+        assertTrue("Delete Succeeded ", !FileUtils.delete(test, true));
+    }
+
+    public void testDeleteNull()
+    {
+        try
+        {
+            FileUtils.delete(null, true);
+            fail("Delete with null value should throw NPE.");
+        }
+        catch (NullPointerException npe)
+        {
+            // expected path
+        }
+    }
+
+    /**
+     * Given two lists of File arrays ensure they are the same length and all entries in Before are in After
+     *
+     * @param filesBefore File[]
+     * @param filesAfter  File[]
+     */
+    private void checkFileLists(File[] filesBefore, File[] filesAfter)
+    {
+        assertNotNull("Before file list cannot be null", filesBefore);
+        assertNotNull("After file list cannot be null", filesAfter);
+
+        assertEquals("File lists are unequal", filesBefore.length, filesAfter.length);
+
+        for (File fileBefore : filesBefore)
+        {
+            boolean found = false;
+
+            for (File fileAfter : filesAfter)
+            {
+                if (fileBefore.getAbsolutePath().equals(fileAfter.getAbsolutePath()))
+                {
+                    found = true;
+                    break;
+                }
+            }
+
+            assertTrue("File'" + fileBefore.getName() + "' was not in directory afterwards", found);
+        }
+    }
+
+    public void testNonRecursiveNonEmptyDirectoryDeleteFails()
+    {
+        String directoryName = "FileUtilsTest-testRecursiveDelete";
+        File test = new File(directoryName);
+
+        //Record file count in parent directory to check it is not changed by delete
+        String path = test.getAbsolutePath();
+        File[] filesBefore = new File(path.substring(0, path.lastIndexOf(File.separator))).listFiles();
+        int fileCountBefore = filesBefore.length;
+
+        assertTrue("Directory exists", !test.exists());
+
+        test.mkdir();
+
+        //Create a file in the directory
+        String fileName = test.getAbsolutePath() + File.separatorChar + "testFile";
+        File subFile = new File(fileName);
+        try
+        {
+            subFile.createNewFile();
+            //Ensure the JVM cleans up if cleanup failues
+            subFile.deleteOnExit();
+        }
+        catch (IOException e)
+        {
+            fail(e.getMessage());
+        }
+        //Ensure the JVM cleans up if cleanup failues
+        // This must be after the subFile as the directory must be empty before
+        // the delete is performed
+        test.deleteOnExit();
+
+        //Try and delete the non-empty directory
+        assertFalse("Non Empty Directory was successfully deleted.", FileUtils.deleteDirectory(directoryName));
+
+        //Check directory is still there
+        assertTrue("Directory was deleted.", test.exists());
+
+        // Clean up
+        assertTrue("Unable to cleanup", FileUtils.delete(test, true));
+
+        //Check that after deletion the file count is now accurate
+        File[] filesAfter = new File(path.substring(0, path.lastIndexOf(File.separator))).listFiles();
+        int fileCountAfter = filesAfter.length;
+        assertEquals("File creation was no registered", fileCountBefore, fileCountAfter);
+
+        checkFileLists(filesBefore, filesAfter);
+    }
+
+    /** Test that an empty directory can be deleted with deleteDirectory */
+    public void testEmptyDirectoryDelete()
+    {
+        String directoryName = "FileUtilsTest-testRecursiveDelete";
+        File test = new File(directoryName);
+
+        //Record file count in parent directory to check it is not changed by delete
+        String path = test.getAbsolutePath();
+        File[] filesBefore = new File(path.substring(0, path.lastIndexOf(File.separator))).listFiles();
+        int fileCountBefore = filesBefore.length;
+
+        assertTrue("Directory exists", !test.exists());
+
+        test.mkdir();
+        //Ensure the JVM cleans up if cleanup failues
+        test.deleteOnExit();
+
+        //Try and delete the empty directory
+        assertTrue("Non Empty Directory was successfully deleted.", FileUtils.deleteDirectory(directoryName));
+
+        //Check directory is still there
+        assertTrue("Directory was deleted.", !test.exists());
+
+        //Check that after deletion the file count is now accurate
+        File[] filesAfter = new File(path.substring(0, path.lastIndexOf(File.separator))).listFiles();
+        int fileCountAfter = filesAfter.length;
+        assertEquals("File creation was no registered", fileCountBefore, fileCountAfter);
+
+        checkFileLists(filesBefore, filesAfter);
+
+    }
+
+    /** Test that deleteDirectory on a non empty directory to complete */
+    public void testNonEmptyDirectoryDelete()
+    {
+        String directoryName = "FileUtilsTest-testRecursiveDelete";
+        File test = new File(directoryName);
+
+        assertTrue("Directory exists", !test.exists());
+
+        //Record file count in parent directory to check it is not changed by delete
+        String path = test.getAbsolutePath();
+        File[] filesBefore = new File(path.substring(0, path.lastIndexOf(File.separator))).listFiles();
+        int fileCountBefore = filesBefore.length;
+
+        test.mkdir();
+
+        //Create a file in the directory
+        String fileName = test.getAbsolutePath() + File.separatorChar + "testFile";
+        File subFile = new File(fileName);
+        try
+        {
+            subFile.createNewFile();
+            //Ensure the JVM cleans up if cleanup failues
+            subFile.deleteOnExit();
+        }
+        catch (IOException e)
+        {
+            fail(e.getMessage());
+        }
+
+        // Ensure the JVM cleans up if cleanup failues
+        // This must be after the subFile as the directory must be empty before
+        // the delete is performed
+        test.deleteOnExit();
+
+        //Try and delete the non-empty directory non-recursively
+        assertFalse("Non Empty Directory was successfully deleted.", FileUtils.delete(test, false));
+
+        //Check directory is still there
+        assertTrue("Directory was deleted.", test.exists());
+
+        // Clean up
+        assertTrue("Unable to cleanup", FileUtils.delete(test, true));
+
+        //Check that after deletion the file count is now accurate
+        File[] filesAfter = new File(path.substring(0, path.lastIndexOf(File.separator))).listFiles();
+        int fileCountAfter = filesAfter.length;
+        assertEquals("File creation was no registered", fileCountBefore, fileCountAfter);
+
+        checkFileLists(filesBefore, filesAfter);
+
+    }
+
+    /** Test that a recursive delete successeds */
+    public void testRecursiveDelete()
+    {
+        String directoryName = "FileUtilsTest-testRecursiveDelete";
+        File test = new File(directoryName);
+
+        assertTrue("Directory exists", !test.exists());
+
+        //Record file count in parent directory to check it is not changed by delete
+        String path = test.getAbsolutePath();
+        File[] filesBefore = new File(path.substring(0, path.lastIndexOf(File.separator))).listFiles();
+        int fileCountBefore = filesBefore.length;
+
+        test.mkdir();
+
+        createSubDir(directoryName, 2, 4);
+
+        //Ensure the JVM cleans up if cleanup failues
+        // This must be after the sub dir creation as the delete order is
+        // recorded and the directory must be empty to be deleted.
+        test.deleteOnExit();
+
+        assertFalse("Non recursive delete was able to directory", FileUtils.delete(test, false));
+
+        assertTrue("File does not exist after non recursive delete", test.exists());
+
+        assertTrue("Unable to cleanup", FileUtils.delete(test, true));
+
+        assertTrue("File  exist after recursive delete", !test.exists());
+
+        //Check that after deletion the file count is now accurate
+        File[] filesAfter = new File(path.substring(0, path.lastIndexOf(File.separator))).listFiles();
+        int fileCountAfter = filesAfter.length;
+        assertEquals("File creation was no registered", fileCountBefore, fileCountAfter);
+
+        checkFileLists(filesBefore, filesAfter);
+
+    }
+
+    private void createSubDir(String path, int directories, int files)
+    {
+        File directory = new File(path);
+
+        assertTrue("Directory" + path + " does not exists", directory.exists());
+
+        for (int dir = 0; dir < directories; dir++)
+        {
+            String subDirName = path + File.separatorChar + "sub" + dir;
+            File subDir = new File(subDirName);
+
+            subDir.mkdir();
+
+            createSubDir(subDirName, directories - 1, files);
+            //Ensure the JVM cleans up if cleanup failues
+            // This must be after the sub dir creation as the delete order is
+            // recorded and the directory must be empty to be deleted.
+            subDir.deleteOnExit();
+        }
+
+        for (int file = 0; file < files; file++)
+        {
+            String subDirName = path + File.separatorChar + "file" + file;
+            File subFile = new File(subDirName);
+            try
+            {
+                subFile.createNewFile();
+                //Ensure the JVM cleans up if cleanup failues
+                subFile.deleteOnExit();
+            }
+            catch (IOException e)
+            {
+                fail(e.getMessage());
+            }
+        }
+    }
+
+    public static final String SEARCH_STRING = "testSearch";
+
+    /**
+     * Test searchFile(File file, String search) will find a match when it
+     * exists.
+     *
+     * @throws java.io.IOException if unable to perform test setup
+     */
+    public void testSearchSucceed() throws IOException
+    {
+        File _logfile = File.createTempFile("FileUtilsTest-testSearchSucceed", ".out");
+
+        prepareFileForSearchTest(_logfile);
+
+        List<String> results = FileUtils.searchFile(_logfile, SEARCH_STRING);
+
+        assertNotNull("Null result set returned", results);
+
+        assertEquals("Results do not contain expected count", 1, results.size());
+    }
+
+    /**
+     * Test searchFile(File file, String search) will not find a match when the
+     * test string does not exist.
+     *
+     * @throws java.io.IOException if unable to perform test setup
+     */
+    public void testSearchFail() throws IOException
+    {
+        File _logfile = File.createTempFile("FileUtilsTest-testSearchFail", ".out");
+
+        prepareFileForSearchTest(_logfile);
+
+        List<String> results = FileUtils.searchFile(_logfile, "Hello");
+
+        assertNotNull("Null result set returned", results);
+
+        //Validate we only got one message
+        if (results.size() > 0)
+        {
+            System.err.println("Unexpected messages");
+
+            for (String msg : results)
+            {
+                System.err.println(msg);
+            }
+        }
+
+        assertEquals("Results contains data when it was not expected",
+                     0, results.size());
+    }
+
+    /**
+     * Write the SEARCH_STRING in to the given file.
+     *
+     * @param logfile The file to write the SEARCH_STRING into
+     *
+     * @throws IOException if an error occurs
+     */
+    private void prepareFileForSearchTest(File logfile) throws IOException
+    {
+        BufferedWriter writer = new BufferedWriter(new FileWriter(logfile));
+        writer.append(SEARCH_STRING);
+        writer.flush();
+        writer.close();
+    }
+
+}
diff --git a/java/common/src/test/java/org/apache/qpid/util/SerialTest.java b/java/common/src/test/java/org/apache/qpid/util/SerialTest.java
index d3ab817b5d..b2578563e0 100644
--- a/java/common/src/test/java/org/apache/qpid/util/SerialTest.java
+++ b/java/common/src/test/java/org/apache/qpid/util/SerialTest.java
@@ -1,4 +1,25 @@
 package org.apache.qpid.util;
+/*
+ * 
+ * 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.
+ * 
+ */
+
 
 import junit.framework.TestCase;