QPID-3473 : Replace use of MINA IO with transport IO (joint work with Robbie Gemmel)

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

33 files changed, 468 insertions, 2126 deletions

diff --git a/java/common/src/main/java/org/apache/qpid/codec/AMQCodecFactory.java b/java/common/src/main/java/org/apache/qpid/codec/AMQCodecFactory.java
index 591dbd085b..c81af9760b 100644
--- a/java/common/src/main/java/org/apache/qpid/codec/AMQCodecFactory.java
+++ b/java/common/src/main/java/org/apache/qpid/codec/AMQCodecFactory.java
@@ -20,9 +20,6 @@
  */
 package org.apache.qpid.codec;
 
-import org.apache.mina.filter.codec.ProtocolCodecFactory;
-import org.apache.mina.filter.codec.ProtocolDecoder;
-import org.apache.mina.filter.codec.ProtocolEncoder;
 import org.apache.qpid.protocol.AMQVersionAwareProtocolSession;
 
 /**
@@ -31,14 +28,11 @@ import org.apache.qpid.protocol.AMQVersionAwareProtocolSession;
  *
  * <p/><table id="crc"><caption>CRC Card</caption>
  * <tr><th> Responsibilities <th> Collaborations.
- * <tr><td> Supply the protocol encoder. <td> {@link AMQEncoder}
  * <tr><td> Supply the protocol decoder. <td> {@link AMQDecoder}
  * </table>
  */
-public class AMQCodecFactory implements ProtocolCodecFactory
+public class AMQCodecFactory
 {
-    /** Holds the protocol encoder. */
-    private final AMQEncoder _encoder = new AMQEncoder();
 
     /** Holds the protocol decoder. */
     private final AMQDecoder _frameDecoder;
@@ -56,15 +50,6 @@ public class AMQCodecFactory implements ProtocolCodecFactory
         _frameDecoder = new AMQDecoder(expectProtocolInitiation, session);
     }
 
-    /**
-     * Gets the AMQP encoder.
-     *
-     * @return The AMQP encoder.
-     */
-    public ProtocolEncoder getEncoder()
-    {
-        return _encoder;
-    }
 
     /**
      * Gets the AMQP decoder.
diff --git a/java/common/src/main/java/org/apache/qpid/codec/AMQDecoder.java b/java/common/src/main/java/org/apache/qpid/codec/AMQDecoder.java
index 281c0761d9..7732ff2fd5 100644
--- a/java/common/src/main/java/org/apache/qpid/codec/AMQDecoder.java
+++ b/java/common/src/main/java/org/apache/qpid/codec/AMQDecoder.java
@@ -23,10 +23,7 @@ package org.apache.qpid.codec;
 import java.util.ArrayList;
 
 import org.apache.mina.common.ByteBuffer;
-import org.apache.mina.common.IoSession;
 import org.apache.mina.common.SimpleByteBufferAllocator;
-import org.apache.mina.filter.codec.CumulativeProtocolDecoder;
-import org.apache.mina.filter.codec.ProtocolDecoderOutput;
 
 import org.apache.qpid.framing.AMQDataBlock;
 import org.apache.qpid.framing.AMQDataBlockDecoder;
@@ -54,11 +51,8 @@ import org.apache.qpid.protocol.AMQVersionAwareProtocolSession;
  * @todo If protocol initiation decoder not needed, then don't create it. Probably not a big deal, but it adds to the
  *       per-session overhead.
  */
-public class AMQDecoder extends CumulativeProtocolDecoder
+public class AMQDecoder
 {
-
-    private static final String BUFFER = AMQDecoder.class.getName() + ".Buffer";
-
     /** Holds the 'normal' AMQP data decoder. */
     private AMQDataBlockDecoder _dataBlockDecoder = new AMQDataBlockDecoder();
 
@@ -84,98 +78,7 @@ public class AMQDecoder extends CumulativeProtocolDecoder
         _bodyFactory = new AMQMethodBodyFactory(session);
     }
 
-    /**
-     * Delegates decoding AMQP from the data buffer that Mina has retrieved from the wire, to the data or protocol
-     * intiation decoders.
-     *
-     * @param session The Mina session.
-     * @param in      The raw byte buffer.
-     * @param out     The Mina object output gatherer to write decoded objects to.
-     *
-     * @return <tt>true</tt> if the data was decoded, <tt>false<tt> if more is needed and the data should accumulate.
-     *
-     * @throws Exception If the data cannot be decoded for any reason.
-     */
-    protected boolean doDecode(IoSession session, ByteBuffer in, ProtocolDecoderOutput out) throws Exception
-    {
-
-        boolean decoded;
-        if (_expectProtocolInitiation  
-            || (firstDecode
-                && (in.remaining() > 0)
-                && (in.get(in.position()) == (byte)'A')))
-        {
-            decoded = doDecodePI(session, in, out);
-        }
-        else
-        {
-            decoded = doDecodeDataBlock(session, in, out);
-        }
-        if(firstDecode && decoded)
-        {
-            firstDecode = false;
-        }
-        return decoded;
-    }
-
-    /**
-     * Decodes AMQP data, delegating the decoding to an {@link AMQDataBlockDecoder}.
-     *
-     * @param session The Mina session.
-     * @param in      The raw byte buffer.
-     * @param out     The Mina object output gatherer to write decoded objects to.
-     *
-     * @return <tt>true</tt> if the data was decoded, <tt>false<tt> if more is needed and the data should accumulate.
-     *
-     * @throws Exception If the data cannot be decoded for any reason.
-     */
-    protected boolean doDecodeDataBlock(IoSession session, ByteBuffer in, ProtocolDecoderOutput out) throws Exception
-    {
-        int pos = in.position();
-        boolean enoughData = _dataBlockDecoder.decodable(in.buf());
-        in.position(pos);
-        if (!enoughData)
-        {
-            // returning false means it will leave the contents in the buffer and
-            // call us again when more data has been read
-            return false;
-        }
-        else
-        {
-            _dataBlockDecoder.decode(session, in, out);
 
-            return true;
-        }
-    }
-
-    /**
-     * Decodes an AMQP initiation, delegating the decoding to a {@link ProtocolInitiation.Decoder}.
-     *
-     * @param session The Mina session.
-     * @param in      The raw byte buffer.
-     * @param out     The Mina object output gatherer to write decoded objects to.
-     *
-     * @return <tt>true</tt> if the data was decoded, <tt>false<tt> if more is needed and the data should accumulate.
-     *
-     * @throws Exception If the data cannot be decoded for any reason.
-     */
-    private boolean doDecodePI(IoSession session, ByteBuffer in, ProtocolDecoderOutput out) throws Exception
-    {
-        boolean enoughData = _piDecoder.decodable(in.buf());
-        if (!enoughData)
-        {
-            // returning false means it will leave the contents in the buffer and
-            // call us again when more data has been read
-            return false;
-        }
-        else
-        {
-            ProtocolInitiation pi = new ProtocolInitiation(in.buf());
-            out.write(pi);
-
-            return true;
-        }
-    }
 
     /**
      * Sets the protocol initation flag, that determines whether decoding is handled by the data decoder of the protocol
@@ -190,97 +93,8 @@ public class AMQDecoder extends CumulativeProtocolDecoder
     }
 
 
-    /**
-     * Cumulates content of <tt>in</tt> into internal buffer and forwards
-     * decoding request to {@link #doDecode(IoSession, ByteBuffer, ProtocolDecoderOutput)}.
-     * <tt>doDecode()</tt> is invoked repeatedly until it returns <tt>false</tt>
-     * and the cumulative buffer is compacted after decoding ends.
-     *
-     * @throws IllegalStateException if your <tt>doDecode()</tt> returned
-     *                               <tt>true</tt> not consuming the cumulative buffer.
-     */
-    public void decode( IoSession session, ByteBuffer in,
-                        ProtocolDecoderOutput out ) throws Exception
-    {
-        ByteBuffer buf = ( ByteBuffer ) session.getAttribute( BUFFER );
-        // if we have a session buffer, append data to that otherwise
-        // use the buffer read from the network directly
-        if( buf != null )
-        {
-            buf.put( in );
-            buf.flip();
-        }
-        else
-        {
-            buf = in;
-        }
-
-        for( ;; )
-        {
-            int oldPos = buf.position();
-            boolean decoded = doDecode( session, buf, out );
-            if( decoded )
-            {
-                if( buf.position() == oldPos )
-                {
-                    throw new IllegalStateException(
-                            "doDecode() can't return true when buffer is not consumed." );
-                }
-
-                if( !buf.hasRemaining() )
-                {
-                    break;
-                }
-            }
-            else
-            {
-                break;
-            }
-        }
-
-        // if there is any data left that cannot be decoded, we store
-        // it in a buffer in the session and next time this decoder is
-        // invoked the session buffer gets appended to
-        if ( buf.hasRemaining() )
-        {
-            storeRemainingInSession( buf, session );
-        }
-        else
-        {
-            removeSessionBuffer( session );
-        }
-    }
-
-    /**
-     * Releases the cumulative buffer used by the specified <tt>session</tt>.
-     * Please don't forget to call <tt>super.dispose( session )</tt> when
-     * you override this method.
-     */
-    public void dispose( IoSession session ) throws Exception
-    {
-        removeSessionBuffer( session );
-    }
-
-    private void removeSessionBuffer(IoSession session)
-    {
-        ByteBuffer buf = ( ByteBuffer ) session.getAttribute( BUFFER );
-        if( buf != null )
-        {
-            buf.release();
-            session.removeAttribute( BUFFER );
-        }
-    }
-
     private static final SimpleByteBufferAllocator SIMPLE_BYTE_BUFFER_ALLOCATOR = new SimpleByteBufferAllocator();
 
-    private void storeRemainingInSession(ByteBuffer buf, IoSession session)
-    {
-        ByteBuffer remainingBuf = SIMPLE_BYTE_BUFFER_ALLOCATOR.allocate( buf.remaining(), false );
-        remainingBuf.setAutoExpand( true );
-        remainingBuf.put( buf );
-        session.setAttribute( BUFFER, remainingBuf );
-    }
-
     public ArrayList<AMQDataBlock> decodeBuffer(java.nio.ByteBuffer buf) throws AMQFrameDecodingException, AMQProtocolVersionException
     {
 
diff --git a/java/common/src/main/java/org/apache/qpid/codec/AMQEncoder.java b/java/common/src/main/java/org/apache/qpid/codec/AMQEncoder.java
deleted file mode 100644
index 53f48ae1c8..0000000000
--- a/java/common/src/main/java/org/apache/qpid/codec/AMQEncoder.java
+++ /dev/null
@@ -1,66 +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.codec;
-
-import org.apache.mina.common.IoSession;
-import org.apache.mina.filter.codec.ProtocolEncoder;
-import org.apache.mina.filter.codec.ProtocolEncoderOutput;
-
-import org.apache.qpid.framing.AMQDataBlockEncoder;
-
-/**
- * AMQEncoder delegates encoding of AMQP to a data encoder.
- *
- * <p/><table id="crc"><caption>CRC Card</caption>
- * <tr><th> Responsibilities <th> Collaborations
- * <tr><td> Delegate AMQP encoding. <td> {@link AMQDataBlockEncoder}
- * </table>
- *
- * @todo This class just delegates to another, so seems to be pointless. Unless it is going to handle some
- *       responsibilities in the future, then drop it.
- */
-public class AMQEncoder implements ProtocolEncoder
-{
-    /** The data encoder that is delegated to. */
-    private AMQDataBlockEncoder _dataBlockEncoder = new AMQDataBlockEncoder();
-
-    /**
-     * Encodes AMQP.
-     *
-     * @param session The Mina session.
-     * @param message The data object to encode.
-     * @param out     The Mina writer to output the raw byte data to.
-     *
-     * @throws Exception If the data cannot be encoded for any reason.
-     */
-    public void encode(IoSession session, Object message, ProtocolEncoderOutput out) throws Exception
-    {
-        _dataBlockEncoder.encode(session, message, out);
-    }
-
-    /**
-     * Does nothing. Called by Mina to allow this to clean up resources when it is no longer needed.
-     *
-     * @param session The Mina session.
-     */
-    public void dispose(IoSession session)
-    { }
-}
diff --git a/java/common/src/main/java/org/apache/qpid/framing/AMQDataBlockDecoder.java b/java/common/src/main/java/org/apache/qpid/framing/AMQDataBlockDecoder.java
index 228867b2b0..0187fa96a9 100644
--- a/java/common/src/main/java/org/apache/qpid/framing/AMQDataBlockDecoder.java
+++ b/java/common/src/main/java/org/apache/qpid/framing/AMQDataBlockDecoder.java
@@ -21,17 +21,12 @@
 package org.apache.qpid.framing;
 
 import org.apache.mina.common.ByteBuffer;
-import org.apache.mina.common.IoSession;
-import org.apache.mina.filter.codec.ProtocolDecoderOutput;
-
-import org.apache.qpid.protocol.AMQVersionAwareProtocolSession;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 public class AMQDataBlockDecoder
 {
-    private static final String SESSION_METHOD_BODY_FACTORY = "QPID_SESSION_METHOD_BODY_FACTORY";
 
     private static final BodyFactory[] _bodiesSupported = new BodyFactory[Byte.MAX_VALUE];
 
@@ -106,19 +101,6 @@ public class AMQDataBlockDecoder
         return frame;
     }
 
-    public void decode(IoSession session, ByteBuffer in, ProtocolDecoderOutput out) throws Exception
-    {
-        AMQMethodBodyFactory bodyFactory = (AMQMethodBodyFactory) session.getAttribute(SESSION_METHOD_BODY_FACTORY);
-        if (bodyFactory == null)
-        {
-            AMQVersionAwareProtocolSession protocolSession = (AMQVersionAwareProtocolSession) session.getAttachment();
-            bodyFactory = new AMQMethodBodyFactory(protocolSession);
-            session.setAttribute(SESSION_METHOD_BODY_FACTORY, bodyFactory);
-        }
-        
-        out.write(createAndPopulateFrame(bodyFactory, in));
-    }
-
     public boolean decodable(ByteBuffer msg) throws AMQFrameDecodingException
     {
         return decodable(msg.buf());
diff --git a/java/common/src/main/java/org/apache/qpid/framing/AMQDataBlockEncoder.java b/java/common/src/main/java/org/apache/qpid/framing/AMQDataBlockEncoder.java
index 374644b4f2..d3b8ecf8bd 100644
--- a/java/common/src/main/java/org/apache/qpid/framing/AMQDataBlockEncoder.java
+++ b/java/common/src/main/java/org/apache/qpid/framing/AMQDataBlockEncoder.java
@@ -21,9 +21,6 @@
 package org.apache.qpid.framing;
 
 import org.apache.mina.common.ByteBuffer;
-import org.apache.mina.common.IoSession;
-import org.apache.mina.filter.codec.ProtocolEncoderOutput;
-import org.apache.mina.filter.codec.demux.MessageEncoder;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -31,7 +28,7 @@ import org.slf4j.LoggerFactory;
 import java.util.Collections;
 import java.util.Set;
 
-public final class AMQDataBlockEncoder implements MessageEncoder
+public final class AMQDataBlockEncoder
 {
     private static final Logger _logger = LoggerFactory.getLogger(AMQDataBlockEncoder.class);
 
@@ -40,19 +37,6 @@ public final class AMQDataBlockEncoder implements MessageEncoder
     public AMQDataBlockEncoder()
     { }
 
-    public void encode(IoSession session, Object message, ProtocolEncoderOutput out) throws Exception
-    {
-        final AMQDataBlock frame = (AMQDataBlock) message;
-
-        final ByteBuffer buffer = frame.toByteBuffer();
-
-        if (_logger.isDebugEnabled())
-        {
-            _logger.debug("Encoded frame byte-buffer is '" + EncodingUtils.convertToHexString(buffer) + "'");
-        }
-        
-        out.write(buffer);
-    }
 
     public Set getMessageTypes()
     {
diff --git a/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderBody.java b/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderBody.java
index 30db3b8be7..7526d4c756 100644
--- a/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderBody.java
+++ b/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderBody.java
@@ -138,4 +138,15 @@ public class ContentHeaderBody implements AMQBody
     {
         properties = props;
     }
+
+    @Override
+    public String toString()
+    {
+        return "ContentHeaderBody{" +
+                "classId=" + classId +
+                ", weight=" + weight +
+                ", bodySize=" + bodySize +
+                ", properties=" + properties +
+                '}';
+    }
 }
diff --git a/java/common/src/main/java/org/apache/qpid/pool/Job.java b/java/common/src/main/java/org/apache/qpid/pool/Job.java
deleted file mode 100644
index 82b600de88..0000000000
--- a/java/common/src/main/java/org/apache/qpid/pool/Job.java
+++ /dev/null
@@ -1,253 +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 java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.RejectedExecutionException;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * A Job is a continuation that batches together other continuations, specifically {@link Event}s, into one continuation.
- * The {@link Event}s themselves provide methods to process themselves, so processing a job simply consists of sequentially
- * processing all of its aggregated events.
- *
- * The constructor accepts a maximum number of events for the job, and only runs up to that maximum number when
- * processing the job, but the add method does not enforce this maximum. In other words, not all the enqueued events
- * may be processed in each run of the job, several runs may be required to clear the queue.
- *
- * <p/><table id="crc"><caption>CRC Card</caption>
- * <tr><th> Responsibilities <th> Collaborations
- * <tr><td> Aggregate many coninuations together into a single continuation.
- * <tr><td> Sequentially process aggregated continuations. <td> {@link Event}
- * <tr><td> Provide running and completion status of the aggregate continuation.
- * <tr><td> Execute a terminal continuation upon job completion. <td> {@link JobCompletionHandler}
- * </table>
- *
- * @todo Could make Job implement Runnable, FutureTask, or a custom Continuation interface, to clarify its status as a
- *       continuation. Job is a continuation that aggregates other continuations and as such is a usefull re-usable
- *       piece of code. There may be other palces than the mina filter chain where continuation batching is used within
- *       qpid, so abstracting this out could provide a usefull building block. This also opens the way to different
- *       kinds of job with a common interface, e.g. parallel or sequential jobs etc.
- *
- * @todo For better re-usability could make the completion handler optional. Only run it when one is set.
- */
-public class Job implements ReadWriteRunnable
-{
-    
-    /** Defines the maximum number of events that will be batched into a single job. */
-    public static final int MAX_JOB_EVENTS = Integer.getInteger("amqj.server.read_write_pool.max_events", 10);
-
-    /** The maximum number of events to process per run of the job. More events than this may be queued in the job. */
-    private final int _maxEvents;
-
-    /** Holds the queue of events that make up the job. */
-    private final java.util.Queue<Runnable> _eventQueue = new ConcurrentLinkedQueue<Runnable>();
-
-    /** Holds a status flag, that indicates when the job is actively running. */
-    private final AtomicBoolean _active = new AtomicBoolean();
-
-    private final boolean _readJob;
-
-    private ReferenceCountingExecutorService _poolReference;
-
-    private final static Logger _logger = LoggerFactory.getLogger(Job.class);
-    
-    public Job(ReferenceCountingExecutorService poolReference, int maxEvents, boolean readJob)
-    {
-        _poolReference = poolReference;
-        _maxEvents = maxEvents;
-        _readJob = readJob;
-    }
-
-    /**
-     * Enqueus a continuation for sequential processing by this job.
-     *
-     * @param evt The continuation to enqueue.
-     */
-    public void add(Runnable evt)
-    {
-        _eventQueue.add(evt);
-    }
-
-    /**
-     * Sequentially processes, up to the maximum number per job, the aggregated continuations in enqueued in this job.
-     */
-    boolean processAll()
-    {
-        // limit the number of events processed in one run
-        int i = _maxEvents;
-        while( --i != 0 )
-        {
-            Runnable e = _eventQueue.poll();
-            if (e == null)
-            {
-                return true;
-            }
-            else
-            {
-                e.run();
-            }
-        }
-        return false;
-    }
-
-    /**
-     * Tests if there are no more enqueued continuations to process.
-     *
-     * @return <tt>true</tt> if there are no enqueued continuations in this job, <tt>false</tt> otherwise.
-     */
-    public boolean isComplete()
-    {
-        return _eventQueue.peek() == null;
-    }
-
-    /**
-     * Marks this job as active if it is inactive. This method is thread safe.
-     *
-     * @return <tt>true</tt> if this job was inactive and has now been marked as active, <tt>false</tt> otherwise.
-     */
-    public boolean activate()
-    {
-        return _active.compareAndSet(false, true);
-    }
-
-    /**
-     * Marks this job as inactive. This method is thread safe.
-     */
-    public void deactivate()
-    {
-        _active.set(false);
-    }
-
-    /**
-     * Processes a batch of aggregated continuations, marks this job as inactive and call the terminal continuation.
-     */
-    public void run()
-    {
-        if(processAll())
-        {
-            deactivate();
-            completed();
-        }
-        else
-        {
-            notCompleted();
-        }
-    }
-
-    public boolean isRead()
-    {
-        return _readJob;
-    }
-    
-    /**
-     * Adds an {@link Event} to a {@link Job}, triggering the execution of the job if it is not already running.
-     *
-     * @param job The job.
-     * @param event   The event to hand off asynchronously.
-     */
-    public static void fireAsynchEvent(ExecutorService pool, Job job, Runnable event)
-    {
-
-        job.add(event);
-
-
-        if(pool == null)
-        {
-            return;
-        }
-
-        // rather than perform additional checks on pool to check that it hasn't shutdown.
-        // catch the RejectedExecutionException that will result from executing on a shutdown pool
-        if (job.activate())
-        {
-            try
-            {
-                pool.execute(job);
-            }
-            catch(RejectedExecutionException e)
-            {
-                _logger.warn("Thread pool shutdown while tasks still outstanding");
-            }
-        }
-
-    }
-    
-    /**
-     * Implements a terminal continuation for the {@link Job} for this filter. Whenever the Job completes its processing
-     * of a batch of events this is called. This method simply re-activates the job, if it has more events to process.
-     *
-     * @param session The Mina session to work in.
-     * @param job     The job that completed.
-     */
-    public void completed()
-    {
-        if (!isComplete())
-        {
-            final ExecutorService pool = _poolReference.getPool();
-
-            if(pool == null)
-            {
-                return;
-            }
-
-
-            // ritchiem : 2006-12-13 Do we need to perform the additional checks here?
-            // Can the pool be shutdown at this point?
-            if (activate())
-            {
-                try
-                {
-                    pool.execute(this);
-                }
-                catch(RejectedExecutionException e)
-                {
-                    _logger.warn("Thread pool shutdown while tasks still outstanding");
-                }
-
-            }
-        }
-    }
-
-    public void notCompleted()
-    {
-        final ExecutorService pool = _poolReference.getPool();
-
-        if(pool == null)
-        {
-            return;
-        }
-
-        try
-        {
-            pool.execute(this);
-        }
-        catch(RejectedExecutionException e)
-        {
-            _logger.warn("Thread pool shutdown while tasks still outstanding");
-        }
-    }
-    
-}
diff --git a/java/common/src/main/java/org/apache/qpid/protocol/AMQConstant.java b/java/common/src/main/java/org/apache/qpid/protocol/AMQConstant.java
index f0f2652ce3..f9f6ca8444 100644
--- a/java/common/src/main/java/org/apache/qpid/protocol/AMQConstant.java
+++ b/java/common/src/main/java/org/apache/qpid/protocol/AMQConstant.java
@@ -80,7 +80,7 @@ public final class AMQConstant
     /**
      * An operator intervened to close the connection for some reason. The client may retry at some later date.
      */
-    public static final AMQConstant CONTEXT_IN_USE = new AMQConstant(320, "context in use", true);
+    public static final AMQConstant CONNECTION_FORCED = new AMQConstant(320, "connection forced", true);
 
     /** The client tried to work with an unknown virtual host or cluster. */
     public static final AMQConstant INVALID_PATH = new AMQConstant(402, "invalid path", true);
diff --git a/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngine.java b/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngine.java
index 48a3df734a..fd651a2b66 100644
--- a/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngine.java
+++ b/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngine.java
@@ -21,8 +21,11 @@
 package org.apache.qpid.protocol;
 
 import java.net.SocketAddress;
+import java.nio.ByteBuffer;
 
 import org.apache.qpid.transport.Receiver;
+import org.apache.qpid.transport.Sender;
+import org.apache.qpid.transport.network.NetworkConnection;
 
 /**
  * A ProtocolEngine is a Receiver for java.nio.ByteBuffers. It takes the data passed to it in the received
@@ -53,4 +56,6 @@ public interface ProtocolEngine extends Receiver<java.nio.ByteBuffer>
    void readerIdle();
 
 
+    public void setNetworkConnection(NetworkConnection network, Sender<ByteBuffer> sender);
+
 }
\ No newline at end of file
diff --git a/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngineFactory.java b/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngineFactory.java
index 4e40b78440..7378edff0c 100644
--- a/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngineFactory.java
+++ b/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngineFactory.java
@@ -26,6 +26,6 @@ public interface ProtocolEngineFactory
 { 
  
   // Returns a new instance of a ProtocolEngine 
-  ProtocolEngine newProtocolEngine(NetworkConnection network);
+  ProtocolEngine newProtocolEngine();
    
 } 
\ No newline at end of file
diff --git a/java/common/src/main/java/org/apache/qpid/transport/ClientDelegate.java b/java/common/src/main/java/org/apache/qpid/transport/ClientDelegate.java
index c8b7ad2a5e..e421f06901 100644
--- a/java/common/src/main/java/org/apache/qpid/transport/ClientDelegate.java
+++ b/java/common/src/main/java/org/apache/qpid/transport/ClientDelegate.java
@@ -210,10 +210,7 @@ public class ClientDelegate extends ConnectionDelegate
             }
             else if (sc.getMechanismName().equals("EXTERNAL"))
             {
-                if (conn.getSecurityLayer() != null)
-                {
-                    conn.setUserID(conn.getSecurityLayer().getUserID());
-                }
+                conn.setUserID(conn.getSecurityLayer().getUserID());
             }
         }
         
diff --git a/java/common/src/main/java/org/apache/qpid/transport/Connection.java b/java/common/src/main/java/org/apache/qpid/transport/Connection.java
index 82a6cdaa67..d6ddbaa061 100644
--- a/java/common/src/main/java/org/apache/qpid/transport/Connection.java
+++ b/java/common/src/main/java/org/apache/qpid/transport/Connection.java
@@ -47,6 +47,7 @@ import org.apache.qpid.transport.network.NetworkConnection;
 import org.apache.qpid.transport.network.OutgoingNetworkTransport;
 import org.apache.qpid.transport.network.Transport;
 import org.apache.qpid.transport.network.security.SecurityLayer;
+import org.apache.qpid.transport.network.security.SecurityLayerFactory;
 import org.apache.qpid.transport.util.Logger;
 import org.apache.qpid.transport.util.Waiter;
 import org.apache.qpid.util.Strings;
@@ -72,6 +73,7 @@ public class Connection extends ConnectionInvoker
     public static final int MAX_CHANNEL_MAX = 0xFFFF;
     public static final int MIN_USABLE_CHANNEL_NUM = 0;
 
+
     public enum State { NEW, CLOSED, OPENING, OPEN, CLOSING, CLOSE_RCVD, RESUMING }
 
     static class DefaultConnectionListener implements ConnectionListener
@@ -124,9 +126,9 @@ public class Connection extends ConnectionInvoker
     private ConnectionSettings conSettings;
     private SecurityLayer securityLayer;
     private String _clientId;
-    
+
     private final AtomicBoolean connectionLost = new AtomicBoolean(false);
-    
+
     public Connection() {}
 
     public void setConnectionDelegate(ConnectionDelegate delegate)
@@ -239,12 +241,22 @@ public class Connection extends ConnectionInvoker
             userID = settings.getUsername();
             delegate = new ClientDelegate(settings);
 
-            securityLayer = new SecurityLayer(this);
+            securityLayer = SecurityLayerFactory.newInstance(getConnectionSettings());
 
             OutgoingNetworkTransport transport = Transport.getOutgoingTransportInstance(ProtocolVersion.v0_10);
-            Receiver<ByteBuffer> receiver = securityLayer.receiver(new InputHandler(new Assembler(this)));
-            NetworkConnection network = transport.connect(settings, receiver, null);
-            sender = new Disassembler(securityLayer.sender(network.getSender()), settings.getMaxFrameSize());
+            Receiver<ByteBuffer> secureReceiver = securityLayer.receiver(new InputHandler(new Assembler(this)));
+            if(secureReceiver instanceof ConnectionListener)
+            {
+                addConnectionListener((ConnectionListener)secureReceiver);
+            }
+
+            NetworkConnection network = transport.connect(settings, secureReceiver, null);
+            final Sender<ByteBuffer> secureSender = securityLayer.sender(network.getSender());
+            if(secureSender instanceof ConnectionListener)
+            {
+                addConnectionListener((ConnectionListener)secureSender);
+            }
+            sender = new Disassembler(secureSender, settings.getMaxFrameSize());
 
             send(new ProtocolHeader(1, 0, 10));
 
@@ -326,14 +338,14 @@ public class Connection extends ConnectionInvoker
             Waiter w = new Waiter(lock, timeout);
             while (w.hasTime() && state != OPEN && error == null)
             {
-                w.await();                
+                w.await();
             }
-            
+
             if (state != OPEN)
             {
                 throw new ConnectionException("Timed out waiting for connection to be ready. Current state is :" + state);
             }
-            
+
             Session ssn = _sessionFactory.newSession(this, name, expiry);
             sessions.put(name, ssn);
             map(ssn);
@@ -475,13 +487,13 @@ public class Connection extends ConnectionInvoker
                     ssn.setState(Session.State.CLOSED);
                 }
                 else
-                {                
+                {
                     map(ssn);
                     ssn.attach();
                     ssn.resume();
                 }
             }
-            
+
             for (Binary ssn_name : transactedSessions)
             {
                 sessions.remove(ssn_name);
@@ -572,12 +584,12 @@ public class Connection extends ConnectionInvoker
     {
         close(ConnectionCloseCode.NORMAL, null);
     }
-    
+
     public void mgmtClose()
     {
         close(ConnectionCloseCode.CONNECTION_FORCED, "The connection was closed using the broker's management interface.");
     }
-    
+
     public void close(ConnectionCloseCode replyCode, String replyText, Option ... _options)
     {
         synchronized (lock)
@@ -680,12 +692,12 @@ public class Connection extends ConnectionInvoker
     {
         return conSettings;
     }
-    
+
     public SecurityLayer getSecurityLayer()
     {
         return securityLayer;
     }
-    
+
     public boolean isConnectionResuming()
     {
         return connectionLost.get();
diff --git a/java/common/src/main/java/org/apache/qpid/transport/SenderClosedException.java b/java/common/src/main/java/org/apache/qpid/transport/SenderClosedException.java
new file mode 100644
index 0000000000..924c327861
--- /dev/null
+++ b/java/common/src/main/java/org/apache/qpid/transport/SenderClosedException.java
@@ -0,0 +1,52 @@
+/*
+ *
+ * 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;
+
+
+/**
+ * SenderClosedException
+ *
+ */
+
+public class SenderClosedException extends SenderException
+{
+
+    public SenderClosedException(String message, Throwable cause)
+    {
+        super(message, cause);
+    }
+
+    public SenderClosedException(String message)
+    {
+        super(message);
+    }
+
+    public SenderClosedException(Throwable cause)
+    {
+        super(cause);
+    }
+
+    public void rethrow()
+    {
+        throw new SenderClosedException(getMessage(), this);
+    }
+
+}
diff --git a/java/common/src/main/java/org/apache/qpid/transport/network/NetworkConnection.java b/java/common/src/main/java/org/apache/qpid/transport/network/NetworkConnection.java
index 1f69973b96..7384702525 100644
--- a/java/common/src/main/java/org/apache/qpid/transport/network/NetworkConnection.java
+++ b/java/common/src/main/java/org/apache/qpid/transport/network/NetworkConnection.java
@@ -29,6 +29,8 @@ public interface NetworkConnection
 {
     Sender<ByteBuffer> getSender();
 
+    void start();
+
     void close();
 
     /**
diff --git a/java/common/src/main/java/org/apache/qpid/transport/network/NetworkTransport.java b/java/common/src/main/java/org/apache/qpid/transport/network/NetworkTransport.java
index 4610c2351e..f71d39c381 100644
--- a/java/common/src/main/java/org/apache/qpid/transport/network/NetworkTransport.java
+++ b/java/common/src/main/java/org/apache/qpid/transport/network/NetworkTransport.java
@@ -27,6 +27,4 @@ package org.apache.qpid.transport.network;
 public interface NetworkTransport
 {
     public void close();
-
-    public NetworkConnection getConnection();
 }
diff --git a/java/common/src/main/java/org/apache/qpid/transport/network/OutgoingNetworkTransport.java b/java/common/src/main/java/org/apache/qpid/transport/network/OutgoingNetworkTransport.java
index 0bae46e8eb..c3c248761c 100644
--- a/java/common/src/main/java/org/apache/qpid/transport/network/OutgoingNetworkTransport.java
+++ b/java/common/src/main/java/org/apache/qpid/transport/network/OutgoingNetworkTransport.java
@@ -29,5 +29,7 @@ import org.apache.qpid.transport.Receiver;
 
 public interface OutgoingNetworkTransport extends NetworkTransport
 {
+    public NetworkConnection getConnection();
+
     public NetworkConnection connect(ConnectionSettings settings, Receiver<ByteBuffer> delegate, SSLContext sslContext);
 }
\ No newline at end of file
diff --git a/java/common/src/main/java/org/apache/qpid/transport/network/Transport.java b/java/common/src/main/java/org/apache/qpid/transport/network/Transport.java
index 2c10a30f10..da4349ba86 100644
--- a/java/common/src/main/java/org/apache/qpid/transport/network/Transport.java
+++ b/java/common/src/main/java/org/apache/qpid/transport/network/Transport.java
@@ -37,7 +37,6 @@ public class Transport
     public static final String QPID_BROKER_TRANSPORT_PROPNAME = "qpid.broker.transport";
 
     // Can't reference the class directly here, as this would preclude the ability to bundle transports separately.
-    private static final String MINA_TRANSPORT_CLASSNAME = "org.apache.qpid.transport.network.mina.MinaNetworkTransport";
     private static final String IO_TRANSPORT_CLASSNAME = "org.apache.qpid.transport.network.io.IoNetworkTransport";
 
     public static final String TCP = "tcp";
@@ -47,9 +46,9 @@ public class Transport
     static
     {
         final Map<ProtocolVersion,String> map = new HashMap<ProtocolVersion, String>();
-        map.put(ProtocolVersion.v8_0, MINA_TRANSPORT_CLASSNAME);
-        map.put(ProtocolVersion.v0_9, MINA_TRANSPORT_CLASSNAME);
-        map.put(ProtocolVersion.v0_91, MINA_TRANSPORT_CLASSNAME);
+        map.put(ProtocolVersion.v8_0, IO_TRANSPORT_CLASSNAME);
+        map.put(ProtocolVersion.v0_9, IO_TRANSPORT_CLASSNAME);
+        map.put(ProtocolVersion.v0_91, IO_TRANSPORT_CLASSNAME);
         map.put(ProtocolVersion.v0_10, IO_TRANSPORT_CLASSNAME);
 
         OUTGOING_PROTOCOL_TO_IMPLDEFAULTS_MAP = Collections.unmodifiableMap(map);
@@ -58,7 +57,7 @@ public class Transport
     public static IncomingNetworkTransport getIncomingTransportInstance()
     {
         return (IncomingNetworkTransport) loadTransportClass(
-                System.getProperty(QPID_BROKER_TRANSPORT_PROPNAME, MINA_TRANSPORT_CLASSNAME));
+                System.getProperty(QPID_BROKER_TRANSPORT_PROPNAME, IO_TRANSPORT_CLASSNAME));
     }
 
     public static OutgoingNetworkTransport getOutgoingTransportInstance(
diff --git a/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkConnection.java b/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkConnection.java
index cca1fc46c9..52cc6363a9 100644
--- a/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkConnection.java
+++ b/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkConnection.java
@@ -37,7 +37,7 @@ public class IoNetworkConnection implements NetworkConnection
     private final long _timeout;
     private final IoSender _ioSender;
     private final IoReceiver _ioReceiver;
-    
+
     public IoNetworkConnection(Socket socket, Receiver<ByteBuffer> delegate,
             int sendBufferSize, int receiveBufferSize, long timeout)
     {
@@ -45,9 +45,15 @@ public class IoNetworkConnection implements NetworkConnection
         _timeout = timeout;
 
         _ioReceiver = new IoReceiver(_socket, delegate, receiveBufferSize,_timeout);
+
         _ioSender = new IoSender(_socket, 2 * sendBufferSize, _timeout);
+
         _ioSender.registerCloseListener(_ioReceiver);
 
+    }
+
+    public void start()
+    {
         _ioReceiver.initiate();
         _ioSender.initiate();
     }
diff --git a/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkTransport.java b/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkTransport.java
index 30e2856c59..7f0f04f9c4 100644
--- a/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkTransport.java
+++ b/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkTransport.java
@@ -21,22 +21,21 @@
 package org.apache.qpid.transport.network.io;
 
 import java.io.IOException;
-import java.net.InetAddress;
-import java.net.InetSocketAddress;
-import java.net.Socket;
-import java.net.SocketException;
+import java.net.*;
 import java.nio.ByteBuffer;
 
 import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLServerSocketFactory;
 
-import org.apache.qpid.transport.ConnectionSettings;
-import org.apache.qpid.transport.Receiver;
-import org.apache.qpid.transport.TransportException;
+import org.apache.qpid.protocol.ProtocolEngine;
+import org.apache.qpid.protocol.ProtocolEngineFactory;
+import org.apache.qpid.transport.*;
+import org.apache.qpid.transport.network.IncomingNetworkTransport;
 import org.apache.qpid.transport.network.NetworkConnection;
 import org.apache.qpid.transport.network.OutgoingNetworkTransport;
 import org.apache.qpid.transport.util.Logger;
 
-public class IoNetworkTransport implements OutgoingNetworkTransport
+public class IoNetworkTransport implements OutgoingNetworkTransport, IncomingNetworkTransport
 {
     static
     {
@@ -51,12 +50,13 @@ public class IoNetworkTransport implements OutgoingNetworkTransport
     private Socket _socket;
     private IoNetworkConnection _connection;
     private long _timeout = 60000;
-    
+    private AcceptingThread _acceptor;
+
     public NetworkConnection connect(ConnectionSettings settings, Receiver<ByteBuffer> delegate, SSLContext sslContext)
     {
         int sendBufferSize = settings.getWriteBufferSize();
         int receiveBufferSize = settings.getReadBufferSize();
-        
+
         try
         {
             _socket = new Socket();
@@ -84,6 +84,7 @@ public class IoNetworkTransport implements OutgoingNetworkTransport
         try
         {
             _connection = new IoNetworkConnection(_socket, delegate, sendBufferSize, receiveBufferSize, _timeout);
+            _connection.start();
         }
         catch(Exception e)
         {
@@ -104,11 +105,134 @@ public class IoNetworkTransport implements OutgoingNetworkTransport
 
     public void close()
     {
-        _connection.close();
+        if(_connection != null)
+        {
+            _connection.close();
+        }
+        if(_acceptor != null)
+        {
+            _acceptor.close();
+        }
     }
 
     public NetworkConnection getConnection()
     {
         return _connection;
     }
+
+    public void accept(NetworkTransportConfiguration config, ProtocolEngineFactory factory, SSLContext sslContext)
+    {
+
+        try
+        {
+            _acceptor = new AcceptingThread(config, factory, sslContext);
+
+            _acceptor.start();
+        }
+        catch (IOException e)
+        {
+            throw new TransportException("Unable to start server socket", e);
+        }
+
+
+    }
+
+    private class AcceptingThread extends Thread
+    {
+        private NetworkTransportConfiguration _config;
+        private ProtocolEngineFactory _factory;
+        private SSLContext _sslContent;
+        private ServerSocket _serverSocket;
+
+        private AcceptingThread(NetworkTransportConfiguration config,
+                                ProtocolEngineFactory factory,
+                                SSLContext sslContext)
+                throws IOException
+        {
+            _config = config;
+            _factory = factory;
+            _sslContent = sslContext;
+
+            InetSocketAddress address = new InetSocketAddress(config.getHost(), config.getPort());
+
+            if(sslContext == null)
+            {
+                _serverSocket = new ServerSocket();
+            }
+            else
+            {
+                SSLServerSocketFactory socketFactory = sslContext.getServerSocketFactory();
+                _serverSocket = socketFactory.createServerSocket();
+            }
+
+            _serverSocket.bind(address);
+            _serverSocket.setReuseAddress(true);
+
+
+        }
+
+
+        /**
+            Close the underlying ServerSocket if it has not already been closed.
+         */
+        public void close()
+        {
+            if (!_serverSocket.isClosed())
+            {
+                try
+                {
+                    _serverSocket.close();
+                }
+                catch (IOException e)
+                {
+                    throw new TransportException(e);
+                }
+            }
+        }
+
+        @Override
+        public void run()
+        {
+            try
+            {
+                while (true)
+                {
+                    try
+                    {
+                        Socket socket = _serverSocket.accept();
+                        socket.setTcpNoDelay(_config.getTcpNoDelay());
+
+                        final Integer sendBufferSize = _config.getSendBufferSize();
+                        final Integer receiveBufferSize = _config.getReceiveBufferSize();
+
+                        socket.setSendBufferSize(sendBufferSize);
+                        socket.setReceiveBufferSize(receiveBufferSize);
+
+                        ProtocolEngine engine = _factory.newProtocolEngine();
+
+                        NetworkConnection connection = new IoNetworkConnection(socket, engine, sendBufferSize, receiveBufferSize, _timeout);
+
+
+                        engine.setNetworkConnection(connection, connection.getSender());
+
+                        connection.start();
+
+
+                    }
+                    catch(RuntimeException e)
+                    {
+                        LOGGER.error(e, "Error in Acceptor thread " + _config.getPort());
+                    }
+                }
+            }
+            catch (IOException e)
+            {
+                LOGGER.debug(e, "SocketException - no new connections will be accepted on port "
+                        + _config.getPort());
+            }
+        }
+
+
+    }
+
 }
diff --git a/java/common/src/main/java/org/apache/qpid/transport/network/io/IoReceiver.java b/java/common/src/main/java/org/apache/qpid/transport/network/io/IoReceiver.java
index fea87fc350..d4b5975e54 100644
--- a/java/common/src/main/java/org/apache/qpid/transport/network/io/IoReceiver.java
+++ b/java/common/src/main/java/org/apache/qpid/transport/network/io/IoReceiver.java
@@ -92,13 +92,23 @@ final class IoReceiver implements Runnable, Closeable
         {
             try
             {
-                if (shutdownBroken)
+                try
                 {
-                   socket.close();
+                    if (shutdownBroken)
+                    {
+                       socket.close();
+                    }
+                    else
+                    {
+                        socket.shutdownInput();
+                    }
                 }
-                else
+                catch(SocketException se)
                 {
-                    socket.shutdownInput();
+                    if(!socket.isClosed() && !socket.isInputShutdown())
+                    {
+                        throw se;
+                    }
                 }
                 if (block && Thread.currentThread() != receiverThread)
                 {
@@ -117,6 +127,7 @@ final class IoReceiver implements Runnable, Closeable
             {
                 throw new TransportException(e);
             }
+
         }
     }
 
diff --git a/java/common/src/main/java/org/apache/qpid/transport/network/io/IoSender.java b/java/common/src/main/java/org/apache/qpid/transport/network/io/IoSender.java
index 1bb515624c..473d4d95ff 100644
--- a/java/common/src/main/java/org/apache/qpid/transport/network/io/IoSender.java
+++ b/java/common/src/main/java/org/apache/qpid/transport/network/io/IoSender.java
@@ -31,6 +31,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
 import org.apache.qpid.common.Closeable;
 import org.apache.qpid.thread.Threading;
 import org.apache.qpid.transport.Sender;
+import org.apache.qpid.transport.SenderClosedException;
 import org.apache.qpid.transport.SenderException;
 import org.apache.qpid.transport.TransportException;
 import org.apache.qpid.transport.util.Logger;
@@ -59,7 +60,7 @@ public final class IoSender implements Runnable, Sender<ByteBuffer>
     private final AtomicBoolean closed = new AtomicBoolean(false);
     private final Thread senderThread;
     private final List<Closeable> _listeners = new ArrayList<Closeable>();
-    
+
     private volatile Throwable exception = null;
 
     public IoSender(Socket socket, int bufferSize, long timeout)
@@ -80,13 +81,13 @@ public final class IoSender implements Runnable, Sender<ByteBuffer>
         try
         {
             //Create but deliberately don't start the thread.
-            senderThread = Threading.getThreadFactory().createThread(this);                      
+            senderThread = Threading.getThreadFactory().createThread(this);
         }
         catch(Exception e)
         {
             throw new Error("Error creating IOSender thread",e);
         }
-        
+
         senderThread.setDaemon(true);
         senderThread.setName(String.format("IoSender - %s", socket.getRemoteSocketAddress()));
     }
@@ -110,7 +111,7 @@ public final class IoSender implements Runnable, Sender<ByteBuffer>
     {
         if (closed.get())
         {
-            throw new SenderException("sender is closed", exception);
+            throw new SenderClosedException("sender is closed", exception);
         }
 
         final int size = buffer.length;
@@ -143,7 +144,7 @@ public final class IoSender implements Runnable, Sender<ByteBuffer>
 
                     if (closed.get())
                     {
-                        throw new SenderException("sender is closed", exception);
+                        throw new SenderClosedException("sender is closed", exception);
                     }
 
                     if (head - tail >= size)
@@ -255,7 +256,7 @@ public final class IoSender implements Runnable, Sender<ByteBuffer>
 
     public void run()
     {
-        final int size = buffer.length;       
+        final int size = buffer.length;
         while (true)
         {
             final int hd = head;
diff --git a/java/common/src/main/java/org/apache/qpid/transport/network/mina/MinaNetworkConnection.java b/java/common/src/main/java/org/apache/qpid/transport/network/mina/MinaNetworkConnection.java
deleted file mode 100644
index 0f433f6eeb..0000000000
--- a/java/common/src/main/java/org/apache/qpid/transport/network/mina/MinaNetworkConnection.java
+++ /dev/null
@@ -1,81 +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.transport.network.mina;
-
-import java.net.SocketAddress;
-import java.nio.ByteBuffer;
-
-import org.apache.mina.common.IdleStatus;
-import org.apache.mina.common.IoSession;
-import org.apache.qpid.transport.Sender;
-import org.apache.qpid.transport.network.NetworkConnection;
-
-public class MinaNetworkConnection implements NetworkConnection
-{
-    private IoSession _session;
-    private Sender<ByteBuffer> _sender;
-    
-    public MinaNetworkConnection(IoSession session)
-    {
-        _session = session;
-        _sender = new MinaSender(_session);
-    }
-
-    public Sender<ByteBuffer> getSender()
-    {
-        return _sender;
-    }
-    
-    public void close()
-    {
-        _session.close();
-    }
-
-    public SocketAddress getRemoteAddress()
-    {
-        return _session.getRemoteAddress();
-    }
-
-    public SocketAddress getLocalAddress()
-    {
-        return _session.getLocalAddress();
-    }
-
-    public long getReadBytes()
-    {
-        return _session.getReadBytes();
-    }
-
-    public long getWrittenBytes()
-    {
-        return _session.getWrittenBytes();
-    }
-
-    public void setMaxWriteIdle(int sec)
-    {
-        _session.setIdleTime(IdleStatus.WRITER_IDLE, sec);
-    }
-
-    public void setMaxReadIdle(int sec)
-    {
-        _session.setIdleTime(IdleStatus.READER_IDLE, sec);
-    }
-}
diff --git a/java/common/src/main/java/org/apache/qpid/transport/network/mina/MinaNetworkHandler.java b/java/common/src/main/java/org/apache/qpid/transport/network/mina/MinaNetworkHandler.java
deleted file mode 100644
index 0e4492e31b..0000000000
--- a/java/common/src/main/java/org/apache/qpid/transport/network/mina/MinaNetworkHandler.java
+++ /dev/null
@@ -1,157 +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.transport.network.mina;
-
-import javax.net.ssl.SSLContext;
-
-import org.apache.mina.common.ByteBuffer;
-import org.apache.mina.common.IdleStatus;
-import org.apache.mina.common.IoHandlerAdapter;
-import org.apache.mina.common.IoSession;
-import org.apache.mina.common.SimpleByteBufferAllocator;
-import org.apache.mina.filter.SSLFilter;
-import org.apache.mina.util.SessionUtil;
-import org.apache.qpid.protocol.ProtocolEngine;
-import org.apache.qpid.protocol.ProtocolEngineFactory;
-import org.apache.qpid.transport.network.NetworkConnection;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class MinaNetworkHandler extends IoHandlerAdapter
-{
-    private static final Logger LOGGER = LoggerFactory.getLogger(MinaNetworkHandler.class);
-
-    private ProtocolEngineFactory _factory;
-    private SSLContext _sslContext = null;
-    private boolean _useClientMode;
-
-    static
-    {
-        boolean directBuffers = Boolean.getBoolean("amqj.enableDirectBuffers");
-        LOGGER.debug("Using " + (directBuffers ? "direct" : "heap") + " buffers");
-        ByteBuffer.setUseDirectBuffers(directBuffers);
-
-        //override the MINA defaults to prevent use of the PooledByteBufferAllocator
-        ByteBuffer.setAllocator(new SimpleByteBufferAllocator());
-    }
-
-    public MinaNetworkHandler(SSLContext sslContext, ProtocolEngineFactory factory)
-    {
-        _sslContext = sslContext;
-        _factory = factory;
-        if(_factory == null)
-        {
-            _useClientMode = true;
-        }
-    }
-
-    public MinaNetworkHandler(SSLContext sslContext)
-    {
-        this(sslContext, null);
-    }
-
-    public void messageReceived(IoSession session, Object message)
-    {
-        ProtocolEngine engine = (ProtocolEngine) session.getAttachment();
-        ByteBuffer buf = (ByteBuffer) message;
-        try
-        {
-            engine.received(buf.buf());
-        }
-        catch (RuntimeException re)
-        {
-            engine.exception(re);
-        }
-    }
-
-    public void exceptionCaught(IoSession ioSession, Throwable throwable) throws Exception
-    {
-        ProtocolEngine engine = (ProtocolEngine) ioSession.getAttachment();
-        if(engine != null)
-        {
-            LOGGER.error("Exception caught by Mina", throwable);
-            engine.exception(throwable);
-        }
-        else
-        {
-            LOGGER.error("Exception caught by Mina but without protocol engine to handle it", throwable);
-        }
-    }
-
-    public void sessionCreated(IoSession ioSession) throws Exception
-    {
-        if(LOGGER.isDebugEnabled())
-        {
-            LOGGER.debug("Created session: " + ioSession.getRemoteAddress());
-        }
-
-        SessionUtil.initialize(ioSession);
-
-        if (_sslContext != null)
-        {
-            SSLFilter sslFilter = new SSLFilter(_sslContext);
-            sslFilter.setUseClientMode(_useClientMode);
-
-            ioSession.getFilterChain().addFirst("sslFilter",sslFilter);
-        }
-
-        if (_factory != null)
-        {
-           NetworkConnection netConn = new MinaNetworkConnection(ioSession);
-
-           ProtocolEngine engine = _factory.newProtocolEngine(netConn);
-           ioSession.setAttachment(engine);
-        }
-    }
-
-    public void sessionClosed(IoSession ioSession) throws Exception
-    {
-        if(LOGGER.isDebugEnabled())
-        {
-            LOGGER.debug("closed: " + ioSession.getRemoteAddress());
-        }
-
-        ProtocolEngine engine = (ProtocolEngine) ioSession.getAttachment();
-        if(engine != null)
-        {
-            engine.closed();
-        }
-        else
-        {
-            LOGGER.error("Unable to close ProtocolEngine as none was present");
-        }
-    }
-
-   
-    public void sessionIdle(IoSession session, IdleStatus status) throws Exception
-    {
-        if (IdleStatus.WRITER_IDLE.equals(status))
-        {
-            ((ProtocolEngine) session.getAttachment()).writerIdle();
-        }
-        else if (IdleStatus.READER_IDLE.equals(status))
-        {
-            ((ProtocolEngine) session.getAttachment()).readerIdle();
-        }
-    }
-
-}
diff --git a/java/common/src/main/java/org/apache/qpid/transport/network/mina/MinaNetworkTransport.java b/java/common/src/main/java/org/apache/qpid/transport/network/mina/MinaNetworkTransport.java
deleted file mode 100644
index 85b42da2b2..0000000000
--- a/java/common/src/main/java/org/apache/qpid/transport/network/mina/MinaNetworkTransport.java
+++ /dev/null
@@ -1,221 +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.transport.network.mina;
-
-import static org.apache.qpid.transport.ConnectionSettings.WILDCARD_ADDRESS;
-
-import java.io.IOException;
-import java.net.InetSocketAddress;
-import java.net.SocketAddress;
-
-import javax.net.ssl.SSLContext;
-
-import org.apache.mina.common.ConnectFuture;
-import org.apache.mina.common.ExecutorThreadModel;
-import org.apache.mina.common.IoConnector;
-import org.apache.mina.common.IoSession;
-import org.apache.mina.transport.socket.nio.SocketAcceptor;
-import org.apache.mina.transport.socket.nio.SocketAcceptorConfig;
-import org.apache.mina.transport.socket.nio.SocketConnector;
-import org.apache.mina.transport.socket.nio.SocketConnectorConfig;
-import org.apache.mina.transport.socket.nio.SocketSessionConfig;
-import org.apache.mina.util.NewThreadExecutor;
-
-import org.apache.qpid.protocol.ProtocolEngineFactory;
-import org.apache.qpid.ssl.SSLContextFactory;
-import org.apache.qpid.thread.QpidThreadExecutor;
-import org.apache.qpid.transport.ConnectionSettings;
-import org.apache.qpid.transport.NetworkTransportConfiguration;
-import org.apache.qpid.transport.Receiver;
-import org.apache.qpid.transport.SocketConnectorFactory;
-import org.apache.qpid.transport.TransportException;
-import org.apache.qpid.transport.network.IncomingNetworkTransport;
-import org.apache.qpid.transport.network.NetworkConnection;
-import org.apache.qpid.transport.network.OutgoingNetworkTransport;
-import org.apache.qpid.transport.network.Transport;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class MinaNetworkTransport implements OutgoingNetworkTransport, IncomingNetworkTransport
-{
-    private static final int UNKNOWN = -1;
-    private static final int TCP = 0;
-
-    public NetworkConnection _connection;
-    private SocketAcceptor _acceptor;
-    private InetSocketAddress _address;
-
-    public NetworkConnection connect(ConnectionSettings settings,
-            Receiver<java.nio.ByteBuffer> delegate, SSLContext sslContext)
-    {
-        int transport = getTransport(settings.getProtocol());
-        
-        IoConnectorCreator stc;
-        switch(transport)
-        {
-            case TCP:
-                stc = new IoConnectorCreator(new SocketConnectorFactory()
-                {
-                    public IoConnector newConnector()
-                    {
-                        return new SocketConnector(1, new QpidThreadExecutor()); // non-blocking connector
-                    }
-                });
-                _connection = stc.connect(delegate, settings, sslContext);
-                break;
-            case UNKNOWN:
-            default:
-                    throw new TransportException("Unknown protocol: " + settings.getProtocol());
-        }
-
-        return _connection;
-    }
-
-    private static int getTransport(String transport)
-    {
-        if (transport.equals(Transport.TCP))
-        {
-            return TCP;
-        }
-
-        return UNKNOWN;
-    }
-
-    public void close()
-    {
-        if(_connection != null)
-        {
-            _connection.close();
-        }
-        if (_acceptor != null)
-        {
-            _acceptor.unbindAll();
-        }
-    }
-
-    public NetworkConnection getConnection()
-    {
-        return _connection;
-    }
-
-    public void accept(final NetworkTransportConfiguration config, final ProtocolEngineFactory factory,
-            final SSLContext sslContext)
-    {
-        int processors = config.getConnectorProcessors();
-        
-        if (Transport.TCP.equalsIgnoreCase(config.getTransport()))
-        {
-            _acceptor = new SocketAcceptor(processors, new NewThreadExecutor());
-    
-            SocketAcceptorConfig sconfig = (SocketAcceptorConfig) _acceptor.getDefaultConfig();
-            sconfig.setThreadModel(ExecutorThreadModel.getInstance("MinaNetworkTransport(Acceptor)"));
-            SocketSessionConfig sc = (SocketSessionConfig) sconfig.getSessionConfig();
-            sc.setTcpNoDelay(config.getTcpNoDelay());
-            sc.setSendBufferSize(config.getSendBufferSize());
-            sc.setReceiveBufferSize(config.getReceiveBufferSize());
-
-            if (config.getHost().equals(WILDCARD_ADDRESS))
-            {
-                _address = new InetSocketAddress(config.getPort());
-            }
-            else
-            {
-                _address = new InetSocketAddress(config.getHost(), config.getPort());
-            }
-        }
-        else
-        {
-            throw new TransportException("Unknown transport: " + config.getTransport());
-        }
-
-        try
-        {
-            _acceptor.bind(_address, new MinaNetworkHandler(sslContext, factory));
-        }
-        catch (IOException e)
-        {
-            throw new TransportException("Could not bind to " + _address, e);
-        }
-    }
-
-
-    private static class IoConnectorCreator
-    {
-        private static final Logger LOGGER = LoggerFactory.getLogger(IoConnectorCreator.class);
-        
-        private static final int CLIENT_DEFAULT_BUFFER_SIZE = 32 * 1024;
-
-        private SocketConnectorFactory _ioConnectorFactory;
-        
-        public IoConnectorCreator(SocketConnectorFactory socketConnectorFactory)
-        {
-            _ioConnectorFactory = socketConnectorFactory;
-        }
-        
-        public NetworkConnection connect(Receiver<java.nio.ByteBuffer> receiver, ConnectionSettings settings, SSLContext sslContext)
-        {
-            final IoConnector ioConnector = _ioConnectorFactory.newConnector();
-            final SocketAddress address;
-            final String protocol = settings.getProtocol();
-            final int port = settings.getPort();
-
-            if (Transport.TCP.equalsIgnoreCase(protocol))
-            {
-                address = new InetSocketAddress(settings.getHost(), port);
-            }
-            else
-            {
-                throw new TransportException("Unknown transport: " + protocol);
-            }
-
-            LOGGER.info("Attempting connection to " + address);
-
-            if (ioConnector instanceof SocketConnector)
-            {
-                SocketConnectorConfig cfg = (SocketConnectorConfig) ioConnector.getDefaultConfig();
-                cfg.setThreadModel(ExecutorThreadModel.getInstance("MinaNetworkTransport(Client)"));
-
-                SocketSessionConfig scfg = (SocketSessionConfig) cfg.getSessionConfig();
-                scfg.setTcpNoDelay(true);
-                scfg.setSendBufferSize(CLIENT_DEFAULT_BUFFER_SIZE);
-                scfg.setReceiveBufferSize(CLIENT_DEFAULT_BUFFER_SIZE);
-
-                // Don't have the connector's worker thread wait around for other
-                // connections (we only use one SocketConnector per connection
-                // at the moment anyway). This allows short-running
-                // clients (like unit tests) to complete quickly.
-                ((SocketConnector) ioConnector).setWorkerTimeout(0);
-            }
-
-            ConnectFuture future = ioConnector.connect(address, new MinaNetworkHandler(sslContext), ioConnector.getDefaultConfig());
-            future.join();
-            if (!future.isConnected())
-            {
-                throw new TransportException("Could not open connection");
-            }
-
-            IoSession session = future.getSession();
-            session.setAttachment(receiver);
-
-            return new MinaNetworkConnection(session);
-        }
-    }
-}
diff --git a/java/common/src/main/java/org/apache/qpid/transport/network/mina/MinaSender.java b/java/common/src/main/java/org/apache/qpid/transport/network/mina/MinaSender.java
deleted file mode 100644
index be114e2fa1..0000000000
--- a/java/common/src/main/java/org/apache/qpid/transport/network/mina/MinaSender.java
+++ /dev/null
@@ -1,79 +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.transport.network.mina;
-
-import org.apache.mina.common.ByteBuffer;
-import org.apache.mina.common.CloseFuture;
-import org.apache.mina.common.IoSession;
-import org.apache.mina.common.WriteFuture;
-import org.apache.qpid.transport.Sender;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * MinaSender
- */
-public class MinaSender implements Sender<java.nio.ByteBuffer>
-{
-    private static final Logger _log = LoggerFactory.getLogger(MinaSender.class);
-    
-    private final IoSession _session;
-    private WriteFuture _lastWrite;
-
-    public MinaSender(IoSession session)
-    {
-        _session = session;
-    }
-
-    public synchronized void send(java.nio.ByteBuffer msg)
-    {
-        _log.debug("sending data:");
-        ByteBuffer mina = ByteBuffer.allocate(msg.limit());
-        mina.put(msg);
-        mina.flip();
-        _lastWrite = _session.write(mina);
-        _log.debug("sent data:");
-    }
-
-    public synchronized void flush()
-    {
-        if (_lastWrite != null)
-        {
-            _lastWrite.join();
-        }
-    }
-
-    public void close()
-    {
-        // MINA will sometimes throw away in-progress writes when you ask it to close
-        flush();
-        CloseFuture closed = _session.close();
-        closed.join();
-    }
-    
-    public void setIdleTimeout(int i)
-    {
-        //TODO:
-        //We are instead using the setMax[Read|Write]IdleTime methods in 
-        //MinaNetworkConnection for this. Should remove this method from
-        //sender interface, but currently being used by IoSender for 0-10.
-    }
-}
diff --git a/java/common/src/main/java/org/apache/qpid/transport/network/security/SecurityLayer.java b/java/common/src/main/java/org/apache/qpid/transport/network/security/SecurityLayer.java
index e80f8904a3..9fd65c6e51 100644
--- a/java/common/src/main/java/org/apache/qpid/transport/network/security/SecurityLayer.java
+++ b/java/common/src/main/java/org/apache/qpid/transport/network/security/SecurityLayer.java
@@ -27,7 +27,6 @@ import javax.net.ssl.SSLEngine;
 
 import org.apache.qpid.ssl.SSLContextFactory;
 import org.apache.qpid.transport.Connection;
-import org.apache.qpid.transport.ConnectionListener;
 import org.apache.qpid.transport.ConnectionSettings;
 import org.apache.qpid.transport.Receiver;
 import org.apache.qpid.transport.Sender;
@@ -38,148 +37,12 @@ import org.apache.qpid.transport.network.security.ssl.SSLReceiver;
 import org.apache.qpid.transport.network.security.ssl.SSLSender;
 import org.apache.qpid.transport.network.security.ssl.SSLUtil;
 
-public class SecurityLayer
+public interface SecurityLayer
 {
-    ConnectionSettings settings;
-    Connection con;
-    SSLSecurityLayer sslLayer;
-    SASLSecurityLayer saslLayer;
 
-    public SecurityLayer(Connection con)
-    {
-        this.con = con;
-        this.settings = con.getConnectionSettings();
-        if (settings.isUseSSL())
-        {
-            sslLayer = new SSLSecurityLayer();
-        }
-        if (settings.isUseSASLEncryption())
-        {
-            saslLayer = new SASLSecurityLayer();
-        }
-    }
+    public Sender<ByteBuffer> sender(Sender<ByteBuffer> delegate);
+    public Receiver<ByteBuffer> receiver(Receiver<ByteBuffer> delegate);
+    public String getUserID();
 
-    public Sender<ByteBuffer> sender(Sender<ByteBuffer> delegate)
-    {
-        Sender<ByteBuffer> sender = delegate;
-        
-        if (settings.isUseSSL())
-        {
-            sender = sslLayer.sender(sender);
-        }     
-        
-        if (settings.isUseSASLEncryption())
-        {
-            sender = saslLayer.sender(sender);
-        }
-        
-        return sender;
-    }
-    
-    public Receiver<ByteBuffer> receiver(Receiver<ByteBuffer> delegate)
-    {
-        Receiver<ByteBuffer> receiver = delegate;
-        
-        if (settings.isUseSSL())
-        {
-            receiver = sslLayer.receiver(receiver);
-        }        
-        
-        if (settings.isUseSASLEncryption())
-        {
-            receiver = saslLayer.receiver(receiver);
-        }
-        
-        return receiver;
-    }
-    
-    public String getUserID()
-    {
-        if (settings.isUseSSL())
-        {
-            return sslLayer.getUserID();
-        }
-        else
-        {
-            return null;
-        }
-    }
-    
-    class SSLSecurityLayer
-    {
-        final SSLEngine _engine;
-        final SSLStatus _sslStatus = new SSLStatus();
-                
-        public SSLSecurityLayer() 
-        {
-            SSLContext sslCtx;
-            try
-            {
-                sslCtx = SSLContextFactory
-                        .buildClientContext(settings.getTrustStorePath(),
-                                settings.getTrustStorePassword(),
-                                settings.getTrustStoreCertType(),
-                                settings.getKeyStorePath(),
-                                settings.getKeyStorePassword(),
-                                settings.getKeyStoreCertType(),
-                                settings.getCertAlias());
-            }
-            catch (Exception e)
-            {
-                throw new TransportException("Error creating SSL Context", e);
-            }
-            
-            try
-            {
-                _engine = sslCtx.createSSLEngine();
-                _engine.setUseClientMode(true);
-            }
-            catch(Exception e)
-            {
-                throw new TransportException("Error creating SSL Engine", e);
-            }
-        }
-        
-        public SSLSender sender(Sender<ByteBuffer> delegate)
-        {
-            SSLSender sender = new SSLSender(_engine, delegate, _sslStatus);
-            sender.setConnectionSettings(settings);
-            return sender;
-        }
-        
-        public SSLReceiver receiver(Receiver<ByteBuffer> delegate)
-        {
-            SSLReceiver receiver = new SSLReceiver(_engine, delegate, _sslStatus);
-            receiver.setConnectionSettings(settings);
-            return receiver;
-        }
-        
-        public String getUserID()
-        {
-            return SSLUtil.retriveIdentity(_engine);
-        }
-        
-    }
-    
-    class SASLSecurityLayer
-    {
-        public SASLSecurityLayer() 
-        {
-        }
-        
-        public SASLSender sender(Sender<ByteBuffer> delegate)
-        {
-            SASLSender sender = new SASLSender(delegate);
-            con.addConnectionListener((ConnectionListener)sender);
-            return sender;
-        }
-        
-        public SASLReceiver receiver(Receiver<ByteBuffer> delegate)
-        {
-            SASLReceiver receiver = new SASLReceiver(delegate);
-            con.addConnectionListener((ConnectionListener)receiver);
-            return receiver;
-        }
-        
-    }
 }
+
diff --git a/java/common/src/main/java/org/apache/qpid/transport/network/security/SecurityLayerFactory.java b/java/common/src/main/java/org/apache/qpid/transport/network/security/SecurityLayerFactory.java
new file mode 100644
index 0000000000..17f89c34ef
--- /dev/null
+++ b/java/common/src/main/java/org/apache/qpid/transport/network/security/SecurityLayerFactory.java
@@ -0,0 +1,161 @@
+package org.apache.qpid.transport.network.security;
+
+import org.apache.qpid.ssl.SSLContextFactory;
+import org.apache.qpid.transport.*;
+import org.apache.qpid.transport.network.security.sasl.SASLReceiver;
+import org.apache.qpid.transport.network.security.sasl.SASLSender;
+import org.apache.qpid.transport.network.security.ssl.SSLReceiver;
+import org.apache.qpid.transport.network.security.ssl.SSLSender;
+import org.apache.qpid.transport.network.security.ssl.SSLUtil;
+
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLEngine;
+import java.nio.ByteBuffer;
+
+public class SecurityLayerFactory
+{
+    public static SecurityLayer newInstance(ConnectionSettings settings)
+    {
+
+        SecurityLayer layer = NullSecurityLayer.getInstance();
+
+        if (settings.isUseSSL())
+        {
+            layer = new SSLSecurityLayer(settings, layer);
+        }
+        if (settings.isUseSASLEncryption())
+        {
+            layer = new SASLSecurityLayer(layer);
+        }
+
+        return layer;
+
+    }
+
+    static class SSLSecurityLayer implements SecurityLayer
+    {
+
+        private final SSLEngine _engine;
+        private final SSLStatus _sslStatus = new SSLStatus();
+        private String _hostname;
+        private SecurityLayer _layer;
+
+
+        public SSLSecurityLayer(ConnectionSettings settings, SecurityLayer layer)
+        {
+
+            SSLContext sslCtx;
+            _layer = layer;
+            try
+            {
+                sslCtx = SSLContextFactory
+                        .buildClientContext(settings.getTrustStorePath(),
+                                settings.getTrustStorePassword(),
+                                settings.getTrustStoreCertType(),
+                                settings.getKeyStorePath(),
+                                settings.getKeyStorePassword(),
+                                settings.getKeyStoreCertType(),
+                                settings.getCertAlias());
+            }
+            catch (Exception e)
+            {
+                throw new TransportException("Error creating SSL Context", e);
+            }
+
+            if(settings.isVerifyHostname())
+            {
+                _hostname = settings.getHost();
+            }
+
+            try
+            {
+                _engine = sslCtx.createSSLEngine();
+                _engine.setUseClientMode(true);
+            }
+            catch(Exception e)
+            {
+                throw new TransportException("Error creating SSL Engine", e);
+            }
+
+        }
+
+        public Sender<ByteBuffer> sender(Sender<ByteBuffer> delegate)
+        {
+            SSLSender sender = new SSLSender(_engine, _layer.sender(delegate), _sslStatus);
+            sender.setHostname(_hostname);
+            return sender;
+        }
+
+        public Receiver<ByteBuffer> receiver(Receiver<ByteBuffer> delegate)
+        {
+            SSLReceiver receiver = new SSLReceiver(_engine, _layer.receiver(delegate), _sslStatus);
+            receiver.setHostname(_hostname);
+            return receiver;
+        }
+
+        public String getUserID()
+        {
+            return SSLUtil.retriveIdentity(_engine);
+        }
+    }
+
+
+    static class SASLSecurityLayer implements SecurityLayer
+    {
+
+        private SecurityLayer _layer;
+
+        SASLSecurityLayer(SecurityLayer layer)
+        {
+            _layer = layer;
+        }
+
+        public SASLSender sender(Sender<ByteBuffer> delegate)
+        {
+            SASLSender sender = new SASLSender(_layer.sender(delegate));
+            return sender;
+        }
+
+        public SASLReceiver receiver(Receiver<ByteBuffer> delegate)
+        {
+            SASLReceiver receiver = new SASLReceiver(_layer.receiver(delegate));
+            return receiver;
+        }
+
+        public String getUserID()
+        {
+            return _layer.getUserID();
+        }
+    }
+
+
+    static class NullSecurityLayer implements SecurityLayer
+    {
+
+        private static final NullSecurityLayer INSTANCE = new NullSecurityLayer();
+
+        private NullSecurityLayer()
+        {
+        }
+
+        public Sender<ByteBuffer> sender(Sender<ByteBuffer> delegate)
+        {
+            return delegate;
+        }
+
+        public Receiver<ByteBuffer> receiver(Receiver<ByteBuffer> delegate)
+        {
+            return delegate;
+        }
+
+        public String getUserID()
+        {
+            return null;
+        }
+
+        public static NullSecurityLayer getInstance()
+        {
+            return INSTANCE;
+        }
+    }
+}
diff --git a/java/common/src/main/java/org/apache/qpid/transport/network/security/ssl/SSLReceiver.java b/java/common/src/main/java/org/apache/qpid/transport/network/security/ssl/SSLReceiver.java
index 878f0b2352..8ad40bbfd3 100644
--- a/java/common/src/main/java/org/apache/qpid/transport/network/security/ssl/SSLReceiver.java
+++ b/java/common/src/main/java/org/apache/qpid/transport/network/security/ssl/SSLReceiver.java
@@ -43,10 +43,11 @@ public class SSLReceiver implements Receiver<ByteBuffer>
     private final int sslBufSize;
     private final ByteBuffer localBuffer;
     private final SSLStatus _sslStatus;
-    private ConnectionSettings settings;
     private ByteBuffer appData;
     private boolean dataCached = false;
 
+    private String _hostname;
+
     public SSLReceiver(final SSLEngine engine, final Receiver<ByteBuffer> delegate, final SSLStatus sslStatus)
     {
         this.engine = engine;
@@ -57,9 +58,9 @@ public class SSLReceiver implements Receiver<ByteBuffer>
         _sslStatus = sslStatus;
     }
 
-    public void setConnectionSettings(ConnectionSettings settings)
+    public void setHostname(String hostname)
     {
-        this.settings = settings;
+        _hostname = hostname;
     }
     
     public void closed()
@@ -166,9 +167,9 @@ public class SSLReceiver implements Receiver<ByteBuffer>
                         handshakeStatus = engine.getHandshakeStatus();
 
                     case FINISHED:
-                        if (this.settings != null && this.settings.isVerifyHostname() )
+                        if (_hostname != null)
                         {
-                            SSLUtil.verifyHostname(engine, this.settings.getHost());
+                            SSLUtil.verifyHostname(engine, _hostname);
                         }
                             
                     case NEED_WRAP:                        
diff --git a/java/common/src/main/java/org/apache/qpid/transport/network/security/ssl/SSLSender.java b/java/common/src/main/java/org/apache/qpid/transport/network/security/ssl/SSLSender.java
index 5e0ee93cb8..6f5aa6d86e 100644
--- a/java/common/src/main/java/org/apache/qpid/transport/network/security/ssl/SSLSender.java
+++ b/java/common/src/main/java/org/apache/qpid/transport/network/security/ssl/SSLSender.java
@@ -44,8 +44,9 @@ public class SSLSender implements Sender<ByteBuffer>
     private final ByteBuffer netData;
     private final long timeout;
     private final SSLStatus _sslStatus;
-    private ConnectionSettings settings;
-    
+
+    private String _hostname;
+
     private final AtomicBoolean closed = new AtomicBoolean(false);
 
 
@@ -59,9 +60,9 @@ public class SSLSender implements Sender<ByteBuffer>
         _sslStatus = sslStatus;
     }
     
-    public void setConnectionSettings(ConnectionSettings settings)
+    public void setHostname(String hostname)
     {
-        this.settings = settings;
+        _hostname = hostname;
     }
 
     public void close()
@@ -237,9 +238,9 @@ public class SSLSender implements Sender<ByteBuffer>
                     break;
 
                 case FINISHED:
-                    if (this.settings != null && this.settings.isVerifyHostname() )
+                    if (_hostname != null)
                     {
-                        SSLUtil.verifyHostname(engine, this.settings.getHost());
+                        SSLUtil.verifyHostname(engine, _hostname);
                     }
                     
                 case NOT_HANDSHAKING:
diff --git a/java/common/src/test/java/org/apache/qpid/session/TestSession.java b/java/common/src/test/java/org/apache/qpid/session/TestSession.java
deleted file mode 100644
index aafc91b03b..0000000000
--- a/java/common/src/test/java/org/apache/qpid/session/TestSession.java
+++ /dev/null
@@ -1,277 +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.session;
-
-import org.apache.mina.common.*;
-
-import java.net.SocketAddress;
-import java.util.Set;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.ConcurrentHashMap;
-
-public class TestSession implements IoSession
-{
-    private final ConcurrentMap attributes = new ConcurrentHashMap();
-
-    public TestSession()
-    {
-    }
-
-    public IoService getService()
-    {
-        return null;  //TODO
-    }
-
-    public IoServiceConfig getServiceConfig()
-    {
-        return null;  //TODO
-    }
-
-    public IoHandler getHandler()
-    {
-        return null;  //TODO
-    }
-
-    public IoSessionConfig getConfig()
-    {
-        return null;  //TODO
-    }
-
-    public IoFilterChain getFilterChain()
-    {
-        return null;  //TODO
-    }
-
-    public WriteFuture write(Object message)
-    {
-        return null;  //TODO
-    }
-
-    public CloseFuture close()
-    {
-        return null;  //TODO
-    }
-
-    public Object getAttachment()
-    {
-        return getAttribute("");
-    }
-
-    public Object setAttachment(Object attachment)
-    {
-        return setAttribute("",attachment);
-    }
-
-    public Object getAttribute(String key)
-    {
-        return attributes.get(key);
-    }
-
-    public Object setAttribute(String key, Object value)
-    {
-        return attributes.put(key,value);
-    }
-
-    public Object setAttribute(String key)
-    {
-        return attributes.put(key, Boolean.TRUE);
-    }
-
-    public Object removeAttribute(String key)
-    {
-        return attributes.remove(key);
-    }
-
-    public boolean containsAttribute(String key)
-    {
-        return attributes.containsKey(key);
-    }
-
-    public Set getAttributeKeys()
-    {
-        return attributes.keySet();
-    }
-
-    public TransportType getTransportType()
-    {
-        return null;  //TODO
-    }
-
-    public boolean isConnected()
-    {
-        return false;  //TODO
-    }
-
-    public boolean isClosing()
-    {
-        return false;  //TODO
-    }
-
-    public CloseFuture getCloseFuture()
-    {
-        return null;  //TODO
-    }
-
-    public SocketAddress getRemoteAddress()
-    {
-        return null;  //TODO
-    }
-
-    public SocketAddress getLocalAddress()
-    {
-        return null;  //TODO
-    }
-
-    public SocketAddress getServiceAddress()
-    {
-        return null;  //TODO
-    }
-
-    public int getIdleTime(IdleStatus status)
-    {
-        return 0;  //TODO
-    }
-
-    public long getIdleTimeInMillis(IdleStatus status)
-    {
-        return 0;  //TODO
-    }
-
-    public void setIdleTime(IdleStatus status, int idleTime)
-    {
-        //TODO
-    }
-
-    public int getWriteTimeout()
-    {
-        return 0;  //TODO
-    }
-
-    public long getWriteTimeoutInMillis()
-    {
-        return 0;  //TODO
-    }
-
-    public void setWriteTimeout(int writeTimeout)
-    {
-        //TODO
-    }
-
-    public TrafficMask getTrafficMask()
-    {
-        return null;  //TODO
-    }
-
-    public void setTrafficMask(TrafficMask trafficMask)
-    {
-        //TODO
-    }
-
-    public void suspendRead()
-    {
-        //TODO
-    }
-
-    public void suspendWrite()
-    {
-        //TODO
-    }
-
-    public void resumeRead()
-    {
-        //TODO
-    }
-
-    public void resumeWrite()
-    {
-        //TODO
-    }
-
-    public long getReadBytes()
-    {
-        return 0;  //TODO
-    }
-
-    public long getWrittenBytes()
-    {
-        return 0;  //TODO
-    }
-
-    public long getReadMessages()
-    {
-        return 0;
-    }
-
-    public long getWrittenMessages()
-    {
-        return 0;
-    }
-
-    public long getWrittenWriteRequests()
-    {
-        return 0;  //TODO
-    }
-
-    public int getScheduledWriteRequests()
-    {
-        return 0;  //TODO
-    }
-
-    public int getScheduledWriteBytes()
-    {
-        return 0;  //TODO
-    }
-
-    public long getCreationTime()
-    {
-        return 0;  //TODO
-    }
-
-    public long getLastIoTime()
-    {
-        return 0;  //TODO
-    }
-
-    public long getLastReadTime()
-    {
-        return 0;  //TODO
-    }
-
-    public long getLastWriteTime()
-    {
-        return 0;  //TODO
-    }
-
-    public boolean isIdle(IdleStatus status)
-    {
-        return false;  //TODO
-    }
-
-    public int getIdleCount(IdleStatus status)
-    {
-        return 0;  //TODO
-    }
-
-    public long getLastIdleTime(IdleStatus status)
-    {
-        return 0;  //TODO
-    }
-}
diff --git a/java/common/src/test/java/org/apache/qpid/transport/TestNetworkConnection.java b/java/common/src/test/java/org/apache/qpid/transport/TestNetworkConnection.java
index 8686c17414..8533c64fab 100644
--- a/java/common/src/test/java/org/apache/qpid/transport/TestNetworkConnection.java
+++ b/java/common/src/test/java/org/apache/qpid/transport/TestNetworkConnection.java
@@ -49,10 +49,12 @@ public class TestNetworkConnection implements NetworkConnection
         _sender = new MockSender();
     }
 
+
+
     public void bind(int port, InetAddress[] addresses, ProtocolEngineFactory protocolFactory,
             NetworkTransportConfiguration config, SSLContextFactory sslFactory) throws BindException
     {
-        
+
     }
 
     public SocketAddress getLocalAddress()
@@ -68,37 +70,37 @@ public class TestNetworkConnection implements NetworkConnection
     public void open(int port, InetAddress destination, ProtocolEngine engine, NetworkTransportConfiguration 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(int i)
     {
-        
+
     }
 
     public void setPort(int port)
@@ -135,4 +137,8 @@ public class TestNetworkConnection implements NetworkConnection
     {
         return _sender;
     }
+
+    public void start()
+    {
+    }
 }
diff --git a/java/common/src/test/java/org/apache/qpid/transport/network/TransportTest.java b/java/common/src/test/java/org/apache/qpid/transport/network/TransportTest.java
index d2fab7d163..7039b904e3 100644
--- a/java/common/src/test/java/org/apache/qpid/transport/network/TransportTest.java
+++ b/java/common/src/test/java/org/apache/qpid/transport/network/TransportTest.java
@@ -33,7 +33,6 @@ import org.apache.qpid.transport.NetworkTransportConfiguration;
 import org.apache.qpid.transport.Receiver;
 import org.apache.qpid.transport.TransportException;
 import org.apache.qpid.transport.network.io.IoNetworkTransport;
-import org.apache.qpid.transport.network.mina.MinaNetworkTransport;
 
 public class TransportTest extends QpidTestCase
 {
@@ -44,7 +43,7 @@ public class TransportTest extends QpidTestCase
     {
         final OutgoingNetworkTransport networkTransport = Transport.getOutgoingTransportInstance(ProtocolVersion.v8_0);
         assertNotNull(networkTransport);
-        assertTrue(networkTransport instanceof MinaNetworkTransport);
+        assertTrue(networkTransport instanceof IoNetworkTransport);
     }
 
     public void testGloballyOverriddenOutgoingTransportForv0_8() throws Exception
@@ -76,7 +75,7 @@ public class TransportTest extends QpidTestCase
     {
         final IncomingNetworkTransport networkTransport = Transport.getIncomingTransportInstance();
         assertNotNull(networkTransport);
-        assertTrue(networkTransport instanceof MinaNetworkTransport);
+        assertTrue(networkTransport instanceof IoNetworkTransport);
     }
 
     public void testOverriddenGetIncomingTransport() throws Exception
diff --git a/java/common/src/test/java/org/apache/qpid/transport/network/mina/MinaNetworkHandlerTest.java b/java/common/src/test/java/org/apache/qpid/transport/network/mina/MinaNetworkHandlerTest.java
deleted file mode 100644
index 976b141fc0..0000000000
--- a/java/common/src/test/java/org/apache/qpid/transport/network/mina/MinaNetworkHandlerTest.java
+++ /dev/null
@@ -1,540 +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.transport.network.mina;
-
-import static org.apache.qpid.transport.ConnectionSettings.WILDCARD_ADDRESS;
-
-import java.net.InetAddress;
-import java.net.InetSocketAddress;
-import java.net.SocketAddress;
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-
-import org.apache.qpid.framing.AMQDataBlock;
-import org.apache.qpid.protocol.ProtocolEngine;
-import org.apache.qpid.protocol.ProtocolEngineFactory;
-import org.apache.qpid.protocol.ServerProtocolEngine;
-import org.apache.qpid.test.utils.QpidTestCase;
-import org.apache.qpid.transport.ConnectionSettings;
-import org.apache.qpid.transport.NetworkTransportConfiguration;
-import org.apache.qpid.transport.TransportException;
-import org.apache.qpid.transport.network.IncomingNetworkTransport;
-import org.apache.qpid.transport.network.NetworkConnection;
-import org.apache.qpid.transport.network.OutgoingNetworkTransport;
-import org.apache.qpid.transport.network.Transport;
-
-public class MinaNetworkHandlerTest extends QpidTestCase
-{
- 
-    private static final String TEST_DATA = "YHALOTHAR";
-    private int _testPort;
-    private IncomingNetworkTransport _server;
-    private OutgoingNetworkTransport _client;
-    private CountingProtocolEngine _countingEngine; // Keeps a count of how many bytes it's read
-    private Exception _thrownEx;
-    private ConnectionSettings _clientSettings;
-    private NetworkConnection _network;
-    private TestNetworkTransportConfiguration _brokerSettings;
-
-    @Override
-    public void setUp() throws Exception
-    {
-        String host = InetAddress.getLocalHost().getHostName();
-        _testPort = findFreePort();
-
-        _clientSettings = new ConnectionSettings();
-        _clientSettings.setHost(host);
-        _clientSettings.setPort(_testPort);
-
-        _brokerSettings = new TestNetworkTransportConfiguration(_testPort, host);
-
-        _server = new MinaNetworkTransport();
-        _client = new MinaNetworkTransport();
-        _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.
-     */
-    public void testBindOpen() throws Exception
-    {
-        try
-        {
-            _client.connect(_clientSettings, _countingEngine, null);
-        } 
-        catch (TransportException e)
-        {
-            _thrownEx = e;
-        }
-
-        assertNotNull("Open should have failed since no engine bound", _thrownEx);        
-        
-        _server.accept(_brokerSettings, null, null);
-        
-        _client.connect(_clientSettings, _countingEngine, null);
-    } 
-    
-    /**
-     * Tests that a socket can't be opened after a bound NetworkDriver has been closed
-     */
-    public void testBindOpenCloseOpen() throws Exception
-    {
-        _server.accept(_brokerSettings, new EchoProtocolEngineSingletonFactory(), null);
-        _client.connect(_clientSettings, _countingEngine, null);
-        _client.close();
-        _server.close();
-        
-        try
-        {
-            _client.connect(_clientSettings, _countingEngine, null);
-        } 
-        catch (TransportException 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.accept(_brokerSettings, new EchoProtocolEngineSingletonFactory(), null);
-        }
-        catch (TransportException e)
-        {
-            fail("First bind should not fail");
-        }
-        
-        try
-        {
-            IncomingNetworkTransport second = new MinaNetworkTransport();
-            second.accept(_brokerSettings, new EchoProtocolEngineSingletonFactory(), null);
-        }
-        catch (TransportException e)
-        {
-            _thrownEx = e;
-        }
-        assertNotNull("Second bind should throw BindException", _thrownEx);
-    }
-
-    /**
-     * Tests that binding to the wildcard address succeeds and a client can
-     * connect via localhost.
-     */
-    public void testWildcardBind() throws Exception
-    {
-        TestNetworkTransportConfiguration serverSettings = 
-            new TestNetworkTransportConfiguration(_testPort, WILDCARD_ADDRESS);
-
-        _server.accept(serverSettings, null, null);
-
-        try
-        {
-            _client.connect(_clientSettings, _countingEngine, null);
-        } 
-        catch (TransportException e)
-        {
-            fail("Open should have succeeded since we used a wildcard bind");        
-        }
-    }
-
-    /**
-     * tests that bytes sent on a network driver are received at the other end
-     */
-    public void testSend() throws Exception 
-    {
-        // Open a connection from a counting engine to an echo engine
-        _server.accept(_brokerSettings, new EchoProtocolEngineSingletonFactory(), null);
-        _network = _client.connect(_clientSettings, _countingEngine, 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 
-        _network.getSender().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
-     * 
-     */
-    public void testSetReadIdle() throws Exception
-    {
-        // Open a connection from a counting engine to an echo engine
-        _server.accept(_brokerSettings, new EchoProtocolEngineSingletonFactory(), null);
-        _network = _client.connect(_clientSettings, _countingEngine, null);
-        assertFalse("Reader should not have been idle", _countingEngine.getReaderHasBeenIdle());
-        _network.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
-     * 
-     */
-    public void testSetWriteIdle() throws Exception
-    {
-        // Open a connection from a counting engine to an echo engine
-        _server.accept(_brokerSettings, new EchoProtocolEngineSingletonFactory(), null);
-        _network = _client.connect(_clientSettings, _countingEngine, null);
-        assertFalse("Reader should not have been idle", _countingEngine.getWriterHasBeenIdle());
-        _network.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.  
-     */
-    public void testClosed() throws Exception
-    {
-        // Open a connection from a counting engine to an echo engine
-        EchoProtocolEngineSingletonFactory factory = new EchoProtocolEngineSingletonFactory();
-        _server.accept(_brokerSettings, factory, null);
-        _network = _client.connect(_clientSettings, _countingEngine, 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.connect(_clientSettings, _countingEngine, 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. 
-     */
-    public void testExceptionCaught() throws Exception
-    {
-        _server.accept(_brokerSettings, new EchoProtocolEngineSingletonFactory(), null);
-        _network = _client.connect(_clientSettings, _countingEngine, null);
-
-
-        assertEquals("Exception should not have been thrown", 1, 
-                _countingEngine.getExceptionLatch().getCount());
-        _countingEngine.setErrorOnNextRead(true);
-        _countingEngine.setNewLatch(TEST_DATA.getBytes().length);
-        _network.getSender().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
-     */
-    public void testGetRemoteAddress() throws Exception
-    {
-        _server.accept(_brokerSettings, new EchoProtocolEngineSingletonFactory(), null);
-        _network = _client.connect(_clientSettings, _countingEngine, null);
-        assertEquals(new InetSocketAddress(InetAddress.getLocalHost(), _testPort),
-                     _network.getRemoteAddress());
-    }
-
-    private class EchoProtocolEngineSingletonFactory implements ProtocolEngineFactory
-    {
-        private EchoProtocolEngine _engine = null;
-        
-        public ProtocolEngine newProtocolEngine(NetworkConnection network)
-        {
-            if (_engine == null)
-            {
-                _engine = new EchoProtocolEngine(network);
-            }
-            return getEngine();
-        }
-        
-        public EchoProtocolEngine getEngine()
-        {
-            return _engine;
-        }
-    }
-    
-    public class CountingProtocolEngine implements ServerProtocolEngine
-    {
-        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()
-        {
-            return _network.getRemoteAddress();
-        }
-        
-        public SocketAddress getLocalAddress()
-        {            
-            return _network.getLocalAddress();
-        }
-
-        public long getWrittenBytes()
-        {
-            return 0;
-        }
-
-        public void readerIdle()
-        {
-            _readerHasBeenIdle = true;
-        }
-
-        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;
-        }
-
-        public long getConnectionId()
-        {
-            return -1;
-        }
-
-    }
-
-    private class EchoProtocolEngine extends CountingProtocolEngine
-    {
-        private NetworkConnection _echoNetwork;
-
-        public EchoProtocolEngine(NetworkConnection network)
-        {
-            _echoNetwork = network;
-        }
-
-        public void received(ByteBuffer msg)
-        {
-            super.received(msg);
-            msg.rewind();
-            _echoNetwork.getSender().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);
-        }
-    }
-
-    private static class TestNetworkTransportConfiguration implements NetworkTransportConfiguration
-    {
-        private int _port;
-        private String _host;
-
-        public TestNetworkTransportConfiguration(final int port, final String host)
-        {
-            _port = port;
-            _host = host;
-        }
-
-        public Boolean getTcpNoDelay()
-        {
-            return true;
-        }
-
-        public Integer getReceiveBufferSize()
-        {
-            return 32768;
-        }
-
-        public Integer getSendBufferSize()
-        {
-            return 32768;
-        }
-
-        public Integer getPort()
-        {
-            return _port;
-        }
-
-        public String getHost()
-        {
-            return _host;
-        }
-
-        public String getTransport()
-        {
-            return Transport.TCP;
-        }
-
-        public Integer getConnectorProcessors()
-        {
-            return 4;
-        }
-        
-    }
-}
\ No newline at end of file