QPID-784 : Added ability to provide existing Socket to Qpid Client Libraries to use as for connection.

AMQBrokerDetails.java, BrokerDetails.java And ConnectionURLTest.java augmented to allow new transport type 'socket'
New ExistingSocketConnector, which utises a given Socket() rather than creating its own from a SocketChannel. This code was taken from the Mina library v1.0.0.
Changes to AMQConnection.java, SocketTransportConnection.java were required to allow the new Socket object to be passed through to the ExistingSocketConnector.
The TransportConnection.java was updated to return an ExistingSocketConnector when the 'socket' transport is used. 
AMQConnection.makeBrokerConnection was changed when the 'socket' transport is being used. This allows the set Socket to be passed down to the ExistingSocketConnector for the transport to be run over.




git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/branches/M2.1@620767 13f79535-47bb-0310-9956-ffa450edef68

7 files changed, 596 insertions, 12 deletions

diff --git a/java/client/src/main/java/org/apache/mina/transport/socket/nio/ExistingSocketConnector.java b/java/client/src/main/java/org/apache/mina/transport/socket/nio/ExistingSocketConnector.java
new file mode 100644
index 0000000000..ab3bc28d83
--- /dev/null
+++ b/java/client/src/main/java/org/apache/mina/transport/socket/nio/ExistingSocketConnector.java
@@ -0,0 +1,478 @@
+/*
+ *  Licensed to the Apache Software Foundation (ASF) under one
+ *  or more contributor license agreements.  See the NOTICE file
+ *  distributed with this work for additional information
+ *  regarding copyright ownership.  The ASF licenses this file
+ *  to you under the Apache License, Version 2.0 (the
+ *  "License"); you may not use this file except in compliance
+ *  with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing,
+ *  software distributed under the License is distributed on an
+ *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ *  KIND, either express or implied.  See the License for the
+ *  specific language governing permissions and limitations
+ *  under the License.
+ *
+ */
+package org.apache.mina.transport.socket.nio;
+
+import edu.emory.mathcs.backport.java.util.concurrent.Executor;
+import org.apache.mina.common.ConnectFuture;
+import org.apache.mina.common.ExceptionMonitor;
+import org.apache.mina.common.IoConnector;
+import org.apache.mina.common.IoConnectorConfig;
+import org.apache.mina.common.IoHandler;
+import org.apache.mina.common.IoServiceConfig;
+import org.apache.mina.common.support.BaseIoConnector;
+import org.apache.mina.common.support.DefaultConnectFuture;
+import org.apache.mina.util.NamePreservingRunnable;
+import org.apache.mina.util.NewThreadExecutor;
+import org.apache.mina.util.Queue;
+
+import java.io.IOException;
+import java.net.ConnectException;
+import java.net.Socket;
+import java.net.SocketAddress;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.SocketChannel;
+import java.util.Iterator;
+import java.util.Set;
+
+/**
+ * {@link IoConnector} for socket transport (TCP/IP).
+ *
+ * @author The Apache Directory Project (mina-dev@directory.apache.org)
+ * @version $Rev$, $Date$
+ */
+public class ExistingSocketConnector extends BaseIoConnector
+{
+    /** @noinspection StaticNonFinalField */
+    private static volatile int nextId = 0;
+
+    private final Object lock = new Object();
+    private final int id = nextId++;
+    private final String threadName = "SocketConnector-" + id;
+    private SocketConnectorConfig defaultConfig = new SocketConnectorConfig();
+    private final Queue connectQueue = new Queue();
+    private final SocketIoProcessor[] ioProcessors;
+    private final int processorCount;
+    private final Executor executor;
+
+    /** @noinspection FieldAccessedSynchronizedAndUnsynchronized */
+    private Selector selector;
+    private Worker worker;
+    private int processorDistributor = 0;
+    private int workerTimeout = 60;  // 1 min.
+    private Socket _openSocket = null;
+
+    /** Create a connector with a single processing thread using a NewThreadExecutor */
+    public ExistingSocketConnector()
+    {
+        this(1, new NewThreadExecutor());
+    }
+
+    /**
+     * Create a connector with the desired number of processing threads
+     *
+     * @param processorCount Number of processing threads
+     * @param executor       Executor to use for launching threads
+     */
+    public ExistingSocketConnector(int processorCount, Executor executor)
+    {
+        if (processorCount < 1)
+        {
+            throw new IllegalArgumentException("Must have at least one processor");
+        }
+
+        this.executor = executor;
+        this.processorCount = processorCount;
+        ioProcessors = new SocketIoProcessor[processorCount];
+
+        for (int i = 0; i < processorCount; i++)
+        {
+            ioProcessors[i] = new SocketIoProcessor("SocketConnectorIoProcessor-" + id + "." + i, executor);
+        }
+    }
+
+    /**
+     * How many seconds to keep the connection thread alive between connection requests
+     *
+     * @return Number of seconds to keep connection thread alive
+     */
+    public int getWorkerTimeout()
+    {
+        return workerTimeout;
+    }
+
+    /**
+     * Set how many seconds the connection worker thread should remain alive once idle before terminating itself.
+     *
+     * @param workerTimeout Number of seconds to keep thread alive. Must be >=0
+     */
+    public void setWorkerTimeout(int workerTimeout)
+    {
+        if (workerTimeout < 0)
+        {
+            throw new IllegalArgumentException("Must be >= 0");
+        }
+        this.workerTimeout = workerTimeout;
+    }
+
+    public ConnectFuture connect(SocketAddress address, IoHandler handler, IoServiceConfig config)
+    {
+        return connect(address, null, handler, config);
+    }
+
+    public ConnectFuture connect(SocketAddress address, SocketAddress localAddress,
+                                 IoHandler handler, IoServiceConfig config)
+    {
+        /** Changes here from the Mina OpenSocketConnector.
+         * Ignoreing all address as they are not needed */
+
+        if (handler == null)
+        {
+            throw new NullPointerException("handler");
+        }
+
+
+        if (config == null)
+        {
+            config = getDefaultConfig();
+        }
+
+        if (_openSocket == null)
+        {
+            throw new IllegalArgumentException("Specifed Socket not active");
+        }
+
+        boolean success = false;
+
+        try
+        {
+            DefaultConnectFuture future = new DefaultConnectFuture();
+            newSession(_openSocket, handler, config, future);
+            success = true;
+            return future;
+        }
+        catch (IOException e)
+        {
+            return DefaultConnectFuture.newFailedFuture(e);
+        }
+        finally
+        {
+            if (!success && _openSocket != null)
+            {
+                try
+                {
+                    _openSocket.close();
+                }
+                catch (IOException e)
+                {
+                    ExceptionMonitor.getInstance().exceptionCaught(e);
+                }
+            }
+        }
+    }
+
+    public IoServiceConfig getDefaultConfig()
+    {
+        return defaultConfig;
+    }
+
+    /**
+     * Sets the config this connector will use by default.
+     *
+     * @param defaultConfig the default config.
+     *
+     * @throws NullPointerException if the specified value is <code>null</code>.
+     */
+    public void setDefaultConfig(SocketConnectorConfig defaultConfig)
+    {
+        if (defaultConfig == null)
+        {
+            throw new NullPointerException("defaultConfig");
+        }
+        this.defaultConfig = defaultConfig;
+    }
+
+    private synchronized void startupWorker() throws IOException
+    {
+        if (worker == null)
+        {
+            selector = Selector.open();
+            worker = new Worker();
+            executor.execute(new NamePreservingRunnable(worker));
+        }
+    }
+
+    private void registerNew()
+    {
+        if (connectQueue.isEmpty())
+        {
+            return;
+        }
+
+        for (; ;)
+        {
+            ConnectionRequest req;
+            synchronized (connectQueue)
+            {
+                req = (ConnectionRequest) connectQueue.pop();
+            }
+
+            if (req == null)
+            {
+                break;
+            }
+
+            SocketChannel ch = req.channel;
+            try
+            {
+                ch.register(selector, SelectionKey.OP_CONNECT, req);
+            }
+            catch (IOException e)
+            {
+                req.setException(e);
+            }
+        }
+    }
+
+    private void processSessions(Set keys)
+    {
+        Iterator it = keys.iterator();
+
+        while (it.hasNext())
+        {
+            SelectionKey key = (SelectionKey) it.next();
+
+            if (!key.isConnectable())
+            {
+                continue;
+            }
+
+            SocketChannel ch = (SocketChannel) key.channel();
+            ConnectionRequest entry = (ConnectionRequest) key.attachment();
+
+            boolean success = false;
+            try
+            {
+                ch.finishConnect();
+                newSession(ch, entry.handler, entry.config, entry);
+                success = true;
+            }
+            catch (Throwable e)
+            {
+                entry.setException(e);
+            }
+            finally
+            {
+                key.cancel();
+                if (!success)
+                {
+                    try
+                    {
+                        ch.close();
+                    }
+                    catch (IOException e)
+                    {
+                        ExceptionMonitor.getInstance().exceptionCaught(e);
+                    }
+                }
+            }
+        }
+
+        keys.clear();
+    }
+
+    private void processTimedOutSessions(Set keys)
+    {
+        long currentTime = System.currentTimeMillis();
+        Iterator it = keys.iterator();
+
+        while (it.hasNext())
+        {
+            SelectionKey key = (SelectionKey) it.next();
+
+            if (!key.isValid())
+            {
+                continue;
+            }
+
+            ConnectionRequest entry = (ConnectionRequest) key.attachment();
+
+            if (currentTime >= entry.deadline)
+            {
+                entry.setException(new ConnectException());
+                try
+                {
+                    key.channel().close();
+                }
+                catch (IOException e)
+                {
+                    ExceptionMonitor.getInstance().exceptionCaught(e);
+                }
+                finally
+                {
+                    key.cancel();
+                }
+            }
+        }
+    }
+
+    private void newSession(Socket socket, IoHandler handler, IoServiceConfig config, ConnectFuture connectFuture)
+            throws IOException
+    {
+        SocketSessionImpl session = new SocketSessionImpl(this,
+                                                          nextProcessor(),
+                                                          getListeners(),
+                                                          config,
+                                                          socket.getChannel(),
+                                                          handler,
+                                                          socket.getRemoteSocketAddress());
+
+        newSession(session, config, connectFuture);
+    }
+
+    private void newSession(SocketChannel ch, IoHandler handler, IoServiceConfig config, ConnectFuture connectFuture)
+            throws IOException
+
+    {
+        SocketSessionImpl session = new SocketSessionImpl(this,
+                                                          nextProcessor(),
+                                                          getListeners(),
+                                                          config,
+                                                          ch,
+                                                          handler,
+                                                          ch.socket().getRemoteSocketAddress());
+
+        newSession(session, config, connectFuture);
+    }
+
+    private void newSession(SocketSessionImpl session, IoServiceConfig config, ConnectFuture connectFuture)
+            throws IOException
+    {
+        try
+        {
+            getFilterChainBuilder().buildFilterChain(session.getFilterChain());
+            config.getFilterChainBuilder().buildFilterChain(session.getFilterChain());
+            config.getThreadModel().buildFilterChain(session.getFilterChain());
+        }
+        catch (Throwable e)
+        {
+            throw (IOException) new IOException("Failed to create a session.").initCause(e);
+        }
+        session.getIoProcessor().addNew(session);
+        connectFuture.setSession(session);
+    }
+
+    private SocketIoProcessor nextProcessor()
+    {
+        return ioProcessors[processorDistributor++ % processorCount];
+    }
+
+    public void setOpenSocket(Socket openSocket)
+    {
+        _openSocket = openSocket;
+    }
+
+    private class Worker implements Runnable
+    {
+        private long lastActive = System.currentTimeMillis();
+
+        public void run()
+        {
+            Thread.currentThread().setName(ExistingSocketConnector.this.threadName);
+
+            for (; ;)
+            {
+                try
+                {
+                    int nKeys = selector.select(1000);
+
+                    registerNew();
+
+                    if (nKeys > 0)
+                    {
+                        processSessions(selector.selectedKeys());
+                    }
+
+                    processTimedOutSessions(selector.keys());
+
+                    if (selector.keys().isEmpty())
+                    {
+                        if (System.currentTimeMillis() - lastActive > workerTimeout * 1000L)
+                        {
+                            synchronized (lock)
+                            {
+                                if (selector.keys().isEmpty() &&
+                                    connectQueue.isEmpty())
+                                {
+                                    worker = null;
+                                    try
+                                    {
+                                        selector.close();
+                                    }
+                                    catch (IOException e)
+                                    {
+                                        ExceptionMonitor.getInstance().exceptionCaught(e);
+                                    }
+                                    finally
+                                    {
+                                        selector = null;
+                                    }
+                                    break;
+                                }
+                            }
+                        }
+                    }
+                    else
+                    {
+                        lastActive = System.currentTimeMillis();
+                    }
+                }
+                catch (IOException e)
+                {
+                    ExceptionMonitor.getInstance().exceptionCaught(e);
+
+                    try
+                    {
+                        Thread.sleep(1000);
+                    }
+                    catch (InterruptedException e1)
+                    {
+                        ExceptionMonitor.getInstance().exceptionCaught(e1);
+                    }
+                }
+            }
+        }
+    }
+
+    private class ConnectionRequest extends DefaultConnectFuture
+    {
+        private final SocketChannel channel;
+        private final long deadline;
+        private final IoHandler handler;
+        private final IoServiceConfig config;
+
+        private ConnectionRequest(SocketChannel channel, IoHandler handler, IoServiceConfig config)
+        {
+            this.channel = channel;
+            long timeout;
+            if (config instanceof IoConnectorConfig)
+            {
+                timeout = ((IoConnectorConfig) config).getConnectTimeoutMillis();
+            }
+            else
+            {
+                timeout = ((IoConnectorConfig) getDefaultConfig()).getConnectTimeoutMillis();
+            }
+            this.deadline = System.currentTimeMillis() + timeout;
+            this.handler = handler;
+            this.config = config;
+        }
+    }
+}
diff --git a/java/client/src/main/java/org/apache/qpid/client/AMQBrokerDetails.java b/java/client/src/main/java/org/apache/qpid/client/AMQBrokerDetails.java
index c04380ba8c..89ce4b2c72 100644
--- a/java/client/src/main/java/org/apache/qpid/client/AMQBrokerDetails.java
+++ b/java/client/src/main/java/org/apache/qpid/client/AMQBrokerDetails.java
@@ -58,7 +58,8 @@ public class AMQBrokerDetails implements BrokerDetails
             {
                 //todo this list of valid transports should be enumerated somewhere
                 if ((!(transport.equalsIgnoreCase("vm") ||
-                       transport.equalsIgnoreCase("tcp"))))
+                       transport.equalsIgnoreCase("tcp") ||
+                       transport.equalsIgnoreCase("socket"))))
                 {
                     if (transport.equalsIgnoreCase("localhost"))
                     {
diff --git a/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java b/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java
index 39b3b80e74..acbe495550 100644
--- a/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java
+++ b/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java
@@ -30,6 +30,8 @@ import org.apache.qpid.client.failover.FailoverRetrySupport;
 import org.apache.qpid.client.protocol.AMQProtocolHandler;
 import org.apache.qpid.client.state.AMQState;
 import org.apache.qpid.client.state.AMQStateManager;
+import org.apache.qpid.client.transport.ITransportConnection;
+import org.apache.qpid.client.transport.SocketTransportConnection;
 import org.apache.qpid.client.transport.TransportConnection;
 import org.apache.qpid.exchange.ExchangeDefaults;
 import org.apache.qpid.framing.*;
@@ -62,6 +64,7 @@ import javax.naming.Referenceable;
 import javax.naming.StringRefAddr;
 import java.io.IOException;
 import java.net.ConnectException;
+import java.net.Socket;
 import java.nio.channels.UnresolvedAddressException;
 import java.text.MessageFormat;
 import java.util.*;
@@ -157,6 +160,9 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
     private static final long DEFAULT_TIMEOUT = 1000 * 30;
     private ProtocolVersion _protocolVersion;
 
+    /** The active socket that is to be used as a value for connection */
+    private Socket _openSocket;
+
     /**
      * @param broker      brokerdetails
      * @param username    username
@@ -173,7 +179,7 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
         this(new AMQConnectionURL(
                 ConnectionURL.AMQ_PROTOCOL + "://" + username + ":" + password + "@"
                 + ((clientName == null) ? "" : clientName) + "/" + virtualHost + "?brokerlist='"
-                + AMQBrokerDetails.checkTransport(broker) + "'"), null);
+                + AMQBrokerDetails.checkTransport(broker) + "'"), null, null);
     }
 
     /**
@@ -192,7 +198,7 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
         this(new AMQConnectionURL(
                 ConnectionURL.AMQ_PROTOCOL + "://" + username + ":" + password + "@"
                 + ((clientName == null) ? "" : clientName) + "/" + virtualHost + "?brokerlist='"
-                + AMQBrokerDetails.checkTransport(broker) + "'"), sslConfig);
+                + AMQBrokerDetails.checkTransport(broker) + "'"), sslConfig, null);
     }
 
     public AMQConnection(String host, int port, String username, String password, String clientName, String virtualHost)
@@ -217,26 +223,38 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
                    + "'" + "," + ConnectionURL.OPTIONS_SSL + "='true'")
                 : (ConnectionURL.AMQ_PROTOCOL + "://" + username + ":" + password + "@"
                    + ((clientName == null) ? "" : clientName) + virtualHost + "?brokerlist='tcp://" + host + ":" + port
-                   + "'" + "," + ConnectionURL.OPTIONS_SSL + "='false'")), sslConfig);
+                   + "'" + "," + ConnectionURL.OPTIONS_SSL + "='false'")), sslConfig, null);
+    }
+
+    public AMQConnection(String connection, Socket socket) throws AMQException, URLSyntaxException
+    {
+        this(new AMQConnectionURL(connection), null, socket);
     }
 
     public AMQConnection(String connection) throws AMQException, URLSyntaxException
     {
-        this(new AMQConnectionURL(connection), null);
+        this(new AMQConnectionURL(connection), null, null);
     }
 
     public AMQConnection(String connection, SSLConfiguration sslConfig) throws AMQException, URLSyntaxException
     {
-        this(new AMQConnectionURL(connection), sslConfig);
+        this(new AMQConnectionURL(connection), sslConfig, null);
     }
 
     public AMQConnection(ConnectionURL connectionURL, SSLConfiguration sslConfig) throws AMQException
     {
+        this(connectionURL, sslConfig, null);
+    }
+
+    public AMQConnection(ConnectionURL connectionURL, SSLConfiguration sslConfig, Socket socket) throws AMQException
+    {
         if (_logger.isInfoEnabled())
         {
             _logger.info("Connection:" + connectionURL);
         }
 
+        _openSocket = socket;
+
         _sslConfiguration = sslConfig;
         if (connectionURL == null)
         {
@@ -395,9 +413,26 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
                 EnumSet.of(AMQState.CONNECTION_OPEN, AMQState.CONNECTION_CLOSED);
         try
         {
-            TransportConnection.getInstance(brokerDetail).connect(_protocolHandler, brokerDetail);
-            // this blocks until the connection has been set up or when an error
-            // has prevented the connection being set up
+
+            ITransportConnection connection = TransportConnection.getInstance(brokerDetail);
+
+            if (brokerDetail.getTransport().equals(BrokerDetails.SOCKET))
+            {
+                if (_openSocket != null)
+                {
+                    ((SocketTransportConnection) connection).setOpenSocket(_openSocket);
+                }
+                else
+                {
+                    throw new IllegalArgumentException("Active Socket must be provided for broker " +
+                                                       "with 'socket' transport:" + brokerDetail);
+                }
+
+            }
+
+            connection.connect(_protocolHandler, brokerDetail);
+             // this blocks until the connection has been set up or when an error
+             // has prevented the connection being set up
 
             //_protocolHandler.attainState(AMQState.CONNECTION_OPEN);
             AMQState state = _protocolHandler.attainState(openOrClosedStates);
@@ -1292,6 +1327,11 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
         return _sessions.get(channelId);
     }
 
+    public void setOpenSocket(Socket socket)
+    {
+        _openSocket = socket;
+    }
+
     public ProtocolVersion getProtocolVersion()
     {
         return _protocolVersion;
diff --git a/java/client/src/main/java/org/apache/qpid/client/transport/SocketTransportConnection.java b/java/client/src/main/java/org/apache/qpid/client/transport/SocketTransportConnection.java
index 5482e48699..e9d6242e77 100644
--- a/java/client/src/main/java/org/apache/qpid/client/transport/SocketTransportConnection.java
+++ b/java/client/src/main/java/org/apache/qpid/client/transport/SocketTransportConnection.java
@@ -24,6 +24,7 @@ import org.apache.mina.common.ByteBuffer;
 import org.apache.mina.common.ConnectFuture;
 import org.apache.mina.common.IoConnector;
 import org.apache.mina.common.SimpleByteBufferAllocator;
+import org.apache.mina.transport.socket.nio.ExistingSocketConnector;
 import org.apache.mina.transport.socket.nio.SocketConnectorConfig;
 import org.apache.mina.transport.socket.nio.SocketSessionConfig;
 
@@ -36,6 +37,7 @@ import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.net.InetSocketAddress;
+import java.net.Socket;
 
 public class SocketTransportConnection implements ITransportConnection
 {
@@ -44,6 +46,8 @@ public class SocketTransportConnection implements ITransportConnection
 
     private SocketConnectorFactory _socketConnectorFactory;
 
+    private Socket _openSocket;
+
     static interface SocketConnectorFactory
     {
         IoConnector newSocketConnector();
@@ -54,6 +58,11 @@ public class SocketTransportConnection implements ITransportConnection
         _socketConnectorFactory = socketConnectorFactory;
     }
 
+    public void setOpenSocket(Socket openSocket)
+    {
+        _openSocket = openSocket;
+    }
+
     public void connect(AMQProtocolHandler protocolHandler, BrokerDetails brokerDetail) throws IOException
     {
         ByteBuffer.setUseDirectBuffers(Boolean.getBoolean("amqj.enableDirectBuffers"));
@@ -83,8 +92,31 @@ public class SocketTransportConnection implements ITransportConnection
         _logger.info("send-buffer-size = " + scfg.getSendBufferSize());
         scfg.setReceiveBufferSize(Integer.getInteger("amqj.receiveBufferSize", DEFAULT_BUFFER_SIZE));
         _logger.info("recv-buffer-size = " + scfg.getReceiveBufferSize());
-        final InetSocketAddress address = new InetSocketAddress(brokerDetail.getHost(), brokerDetail.getPort());
-        _logger.info("Attempting connection to " + address);
+
+        final InetSocketAddress address;
+
+        if (brokerDetail.getTransport().equals(BrokerDetails.SOCKET))
+        {
+            address = null;
+
+            if (_openSocket != null)
+            {
+                _logger.info("Using existing Socket:" + _openSocket);
+                ((ExistingSocketConnector) ioConnector).setOpenSocket(_openSocket);
+            }
+            else
+            {
+                throw new IllegalArgumentException("Active Socket must be provided for broker " +
+                                                   "with 'socket' transport:" + brokerDetail);
+            }
+        }
+        else
+        {
+            address = new InetSocketAddress(brokerDetail.getHost(), brokerDetail.getPort());
+            _logger.info("Attempting connection to " + address);
+        }
+
+
         ConnectFuture future = ioConnector.connect(address, protocolHandler);
 
         // wait for connection to complete
diff --git a/java/client/src/main/java/org/apache/qpid/client/transport/TransportConnection.java b/java/client/src/main/java/org/apache/qpid/client/transport/TransportConnection.java
index c1116ca01e..94361fccfc 100644
--- a/java/client/src/main/java/org/apache/qpid/client/transport/TransportConnection.java
+++ b/java/client/src/main/java/org/apache/qpid/client/transport/TransportConnection.java
@@ -23,6 +23,7 @@ package org.apache.qpid.client.transport;
 import org.apache.mina.common.IoConnector;
 import org.apache.mina.common.IoHandlerAdapter;
 import org.apache.mina.common.IoServiceConfig;
+import org.apache.mina.transport.socket.nio.ExistingSocketConnector;
 import org.apache.mina.transport.socket.nio.MultiThreadSocketConnector;
 import org.apache.mina.transport.socket.nio.SocketConnector;
 import org.apache.mina.transport.vmpipe.VmPipeAcceptor;
@@ -54,6 +55,7 @@ public class TransportConnection
 
     private static final int TCP = 0;
     private static final int VM = 1;
+    private static final int SOCKET = 2;
 
     private static Logger _logger = LoggerFactory.getLogger(TransportConnection.class);
 
@@ -87,7 +89,15 @@ public class TransportConnection
 
         switch (transport)
         {
-
+            case SOCKET:
+                _instance = new SocketTransportConnection(new SocketTransportConnection.SocketConnectorFactory()
+                {
+                    public IoConnector newSocketConnector()
+                    {
+                        return new ExistingSocketConnector();
+                    }
+                });
+                break;
             case TCP:
                 _instance = new SocketTransportConnection(new SocketTransportConnection.SocketConnectorFactory()
                 {
@@ -127,6 +137,11 @@ public class TransportConnection
 
     private static int getTransport(String transport)
     {
+        if (transport.equals(BrokerDetails.SOCKET))
+        {
+            return SOCKET;
+        }
+
         if (transport.equals(BrokerDetails.TCP))
         {
             return TCP;
diff --git a/java/client/src/main/java/org/apache/qpid/jms/BrokerDetails.java b/java/client/src/main/java/org/apache/qpid/jms/BrokerDetails.java
index 603b0834a3..8b353a7264 100644
--- a/java/client/src/main/java/org/apache/qpid/jms/BrokerDetails.java
+++ b/java/client/src/main/java/org/apache/qpid/jms/BrokerDetails.java
@@ -34,6 +34,7 @@ public interface BrokerDetails
     public static final String OPTIONS_CONNECT_DELAY = "connectdelay";
     public static final int DEFAULT_PORT = 5672;
 
+    public static final String SOCKET = "socket";
     public static final String TCP = "tcp";
     public static final String VM = "vm";
 
diff --git a/java/client/src/test/java/org/apache/qpid/test/unit/client/connectionurl/ConnectionURLTest.java b/java/client/src/test/java/org/apache/qpid/test/unit/client/connectionurl/ConnectionURLTest.java
index 6c872a0e10..978ce34d59 100644
--- a/java/client/src/test/java/org/apache/qpid/test/unit/client/connectionurl/ConnectionURLTest.java
+++ b/java/client/src/test/java/org/apache/qpid/test/unit/client/connectionurl/ConnectionURLTest.java
@@ -510,6 +510,23 @@ public class ConnectionURLTest extends TestCase
 
     }
 
+    public void testSocketProtocol() throws URLSyntaxException
+    {
+        String url = "amqp://guest:guest@id/test" + "?brokerlist='socket:///'";
+
+        try
+        {
+            AMQConnectionURL curl = new AMQConnectionURL(url);
+            assertNotNull(curl);
+            assertEquals(1, curl.getBrokerCount());
+            assertNotNull(curl.getBrokerDetails(0));
+            assertEquals("socket", curl.getBrokerDetails(0).getTransport());
+        }
+        catch (URLSyntaxException e)
+        {
+            fail(e.getMessage());
+        }
+    }
 
     public static junit.framework.Test suite()
     {