Merged revisions 620767,620858,620876,627354,627416 via svnmerge from

https://svn.apache.org/repos/asf/incubator/qpid/branches/M2.1

........
  r620767 | ritchiem | 2008-02-12 11:29:19 +0000 (Tue, 12 Feb 2008) | 9 lines
  
  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.
........
  r620858 | rgodfrey | 2008-02-12 16:44:59 +0000 (Tue, 12 Feb 2008) | 1 line
  
  QPID-787 : Allow for quoting of identifiers in selectors
........
  r620876 | ritchiem | 2008-02-12 17:36:07 +0000 (Tue, 12 Feb 2008) | 2 lines
  
  QPID-784 : Added ability to provide existing Socket to Qpid Client Libraries to use as for connection.
  Modified based on review by Robert Godfrey due to Thread safety around SocketTransportConnection.java and TransportConnection.java. Now use a safe Map to store all registered sockets in the TransportConnection.java these are then removed as used or on request.
........
  r627354 | ritchiem | 2008-02-13 11:00:13 +0000 (Wed, 13 Feb 2008) | 1 line
  
  QPID-788 : Changed MAXIMUM_STATE_WAIT_TIME to pickup value via -Damqj.MaximumStateWait=90000 or default to 30000.
........
  r627416 | rgodfrey | 2008-02-13 14:01:26 +0000 (Wed, 13 Feb 2008) | 1 line
  
  QPID-789 : FieldTable putDataInBuffer method not thread safe
........


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

11 files changed, 768 insertions, 14 deletions

diff --git a/qpid/java/broker/src/main/grammar/SelectorParser.jj b/qpid/java/broker/src/main/grammar/SelectorParser.jj
index 61638e5e26..aed3e948a5 100644
--- a/qpid/java/broker/src/main/grammar/SelectorParser.jj
+++ b/qpid/java/broker/src/main/grammar/SelectorParser.jj
@@ -172,6 +172,7 @@ TOKEN [IGNORE_CASE] :
 TOKEN [IGNORE_CASE] :

 {

     < ID : ["a"-"z", "_", "$"] (["a"-"z","0"-"9","_", "$"])* >

+    | < QUOTED_ID : "\"" ( ("\"\"") | ~["\""] )*  "\""  >

 }

 

 // ----------------------------------------------------------------------------

@@ -589,6 +590,7 @@ String stringLitteral() :
 PropertyExpression variable() :

 {

     Token t;

+    StringBuffer rc = new StringBuffer();

     PropertyExpression left=null;

 }

 {

@@ -597,6 +599,21 @@ PropertyExpression variable() :
         {

             left = new PropertyExpression(t.image);

         }

+        |

+        t = <QUOTED_ID>

+        {

+            // Decode the sting value.

+            String image = t.image;

+            for( int i=1; i < image.length()-1; i++ ) {

+                char c = image.charAt(i);

+                if( c == '"' )

+                    i++;

+                rc.append(c);

+            }

+            return new PropertyExpression(rc.toString());

+        }

+

+

     )

     {

         return left;

diff --git a/qpid/java/client/example/src/main/java/org/apache/qpid/example/transport/ExistingSocketConnectorDemo.java b/qpid/java/client/example/src/main/java/org/apache/qpid/example/transport/ExistingSocketConnectorDemo.java
new file mode 100644
index 0000000000..d7eb138523
--- /dev/null
+++ b/qpid/java/client/example/src/main/java/org/apache/qpid/example/transport/ExistingSocketConnectorDemo.java
@@ -0,0 +1,171 @@
+/*
+ *
+ * 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.example.transport;
+
+import org.apache.qpid.AMQException;
+import org.apache.qpid.client.AMQConnection;
+import org.apache.qpid.client.transport.TransportConnection;
+import org.apache.qpid.jms.ConnectionListener;
+import org.apache.qpid.url.URLSyntaxException;
+
+import javax.jms.Connection;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.nio.channels.SocketChannel;
+import java.util.UUID;
+
+/**
+ * This is a simple application that demonstrates how you can use the Qpid AMQP interfaces to use existing sockets as
+ * the transport for the Client API.
+ *
+ * The Demo here runs twice:
+ * 1. Just to show a simple publish and receive.
+ * 2. To demonstrate how to use existing sockets and utilise the underlying client failover mechnaism.
+ */
+public class ExistingSocketConnectorDemo implements ConnectionListener
+{
+    private static boolean DEMO_FAILOVER = false;
+
+    public static void main(String[] args) throws IOException, URLSyntaxException, AMQException, JMSException
+    {
+        System.out.println("Testing socket connection to localhost:5672.");
+
+        new ExistingSocketConnectorDemo();
+
+        System.out.println("Testing socket connection failover between localhost:5672 and localhost:5673.");
+
+        DEMO_FAILOVER = true;
+
+        new ExistingSocketConnectorDemo();
+    }
+
+    Connection _connection;
+    MessageProducer _producer;
+    Session _session;
+
+    String Socket1_ID = UUID.randomUUID().toString();
+    String Socket2_ID = UUID.randomUUID().toString();
+
+
+
+    /** Here we can see the broker we are connecting to is set to be 'socket:///' signifying we will provide the socket. */
+    public final String CONNECTION = "amqp://guest:guest@id/test?brokerlist='socket://" + Socket1_ID + ";socket://" + Socket2_ID + "'";
+
+
+    public ExistingSocketConnectorDemo() throws IOException, URLSyntaxException, AMQException, JMSException
+    {
+
+        Socket socket = SocketChannel.open().socket();
+        socket.connect(new InetSocketAddress("localhost", 5672));
+
+        TransportConnection.registerOpenSocket(Socket1_ID, socket);
+
+
+        _connection = new AMQConnection(CONNECTION);
+
+        _session = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+
+        MessageConsumer consumer = _session.createConsumer(_session.createQueue("Queue"));
+
+        _producer = _session.createProducer(_session.createQueue("Queue"));
+
+        _connection.start();
+
+        if (!DEMO_FAILOVER)
+        {
+            _producer.send(_session.createTextMessage("Simple Test"));
+        }
+        else
+        {
+            // Using the Qpid interfaces we can set a listener that allows us to demonstrate failover
+            ((AMQConnection) _connection).setConnectionListener(this);
+
+            System.out.println("Testing failover: Please ensure second broker running on localhost:5673 and shutdown broker on 5672.");
+        }
+
+        //We do a blocking receive here so that we can demonstrate failover.
+        Message message = consumer.receive();
+
+        System.out.println("Recevied :" + message);
+
+        _connection.close();
+    }
+
+    // ConnectionListener Interface
+
+    public void bytesSent(long count)
+    {
+        //not used in this example
+    }
+    public void bytesReceived(long count)
+    {
+        //not used in this example
+    }
+
+    public boolean preFailover(boolean redirect)
+    {
+        /**
+         * This method is called before the underlying client library starts to reconnect. This gives us the opportunity
+         * to set a new socket for the failover to occur on.
+         */
+        try
+        {
+            Socket socket = SocketChannel.open().socket();
+
+            socket.connect(new InetSocketAddress("localhost", 5673));
+
+            // This is the new method to pass in an open socket for the connection to use.
+            TransportConnection.registerOpenSocket(Socket2_ID, socket);
+        }
+        catch (IOException e)
+        {
+            e.printStackTrace();
+            return false;
+        }
+        return true;
+    }
+
+    public boolean preResubscribe()
+    {
+        //not used in this example - but must return true to allow the resubscription of existing clients.
+        return true;
+    }
+
+    public void failoverComplete()
+    {
+        // Now that failover has completed we can send a message that the receiving thread will pick up
+        try
+        {
+            _producer.send(_session.createTextMessage("Simple Failover Test"));
+        }
+        catch (JMSException e)
+        {
+            e.printStackTrace();
+        }
+    }
+}
diff --git a/qpid/java/client/src/main/java/org/apache/mina/transport/socket/nio/ExistingSocketConnector.java b/qpid/java/client/src/main/java/org/apache/mina/transport/socket/nio/ExistingSocketConnector.java
new file mode 100644
index 0000000000..ab3bc28d83
--- /dev/null
+++ b/qpid/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/qpid/java/client/src/main/java/org/apache/qpid/client/AMQBrokerDetails.java b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQBrokerDetails.java
index b97b1fa253..8c074470ef 100644
--- a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQBrokerDetails.java
+++ b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQBrokerDetails.java
@@ -59,7 +59,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"))
                     {
@@ -157,7 +158,10 @@ public class AMQBrokerDetails implements BrokerDetails
             }
             else
             {
-                setPort(port);
+                if (!_transport.equalsIgnoreCase(SOCKET))
+                {
+                    setPort(port);
+                }
             }
 
             String queryString = connection.getQuery();
@@ -264,13 +268,16 @@ public class AMQBrokerDetails implements BrokerDetails
         sb.append(_transport);
         sb.append("://");
 
-        if (!(_transport.equalsIgnoreCase("vm")))
+        if (!(_transport.equalsIgnoreCase(VM)))
         {
             sb.append(_host);
         }
 
-        sb.append(':');
-        sb.append(_port);
+        if (!(_transport.equalsIgnoreCase(SOCKET)))
+        {
+            sb.append(':');
+            sb.append(_port);
+        }
 
         sb.append(printOptionsURL());
 
diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/state/AMQStateManager.java b/qpid/java/client/src/main/java/org/apache/qpid/client/state/AMQStateManager.java
index 5a0489f906..aeb9e0589c 100644
--- a/qpid/java/client/src/main/java/org/apache/qpid/client/state/AMQStateManager.java
+++ b/qpid/java/client/src/main/java/org/apache/qpid/client/state/AMQStateManager.java
@@ -53,7 +53,7 @@ public class AMQStateManager implements AMQMethodListener
 
     private final CopyOnWriteArraySet _stateListeners = new CopyOnWriteArraySet();
     private final Object _stateLock = new Object();
-    private static final long MAXIMUM_STATE_WAIT_TIME = 30000L;
+    private static final long MAXIMUM_STATE_WAIT_TIME = Long.parseLong(System.getProperty("amqj.MaximumStateWait", "30000"));
 
     public AMQStateManager()
     {
diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/transport/SocketTransportConnection.java b/qpid/java/client/src/main/java/org/apache/qpid/client/transport/SocketTransportConnection.java
index 5482e48699..b2f7ae8395 100644
--- a/qpid/java/client/src/main/java/org/apache/qpid/client/transport/SocketTransportConnection.java
+++ b/qpid/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,9 @@ import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 
 public class SocketTransportConnection implements ITransportConnection
 {
@@ -83,8 +87,34 @@ 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;
+
+            Socket socket = TransportConnection.removeOpenSocket(brokerDetail.getHost());
+
+            if (socket != null)
+            {
+                _logger.info("Using existing Socket:" + socket);
+
+                ((ExistingSocketConnector) ioConnector).setOpenSocket(socket);
+            }
+            else
+            {
+                throw new IllegalArgumentException("Active Socket must be provided for broker " +
+                                                   "with 'socket://<SocketID>' 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/qpid/java/client/src/main/java/org/apache/qpid/client/transport/TransportConnection.java b/qpid/java/client/src/main/java/org/apache/qpid/client/transport/TransportConnection.java
index 8cb3103a5a..b48b862f78 100644
--- a/qpid/java/client/src/main/java/org/apache/qpid/client/transport/TransportConnection.java
+++ b/qpid/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;
@@ -36,6 +37,9 @@ import org.slf4j.LoggerFactory;
 import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.net.Socket;
+
 
 /**
  * The TransportConnection is a helper class responsible for connecting to an AMQ server. It sets up the underlying
@@ -54,11 +58,24 @@ 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);
 
     private static final String DEFAULT_QPID_SERVER = "org.apache.qpid.server.protocol.AMQPFastProtocolHandler";
 
+    private static Map<String, Socket> _openSocketRegister = new ConcurrentHashMap<String, Socket>();
+
+    public static void registerOpenSocket(String socketID, Socket openSocket)
+    {
+        _openSocketRegister.put(socketID, openSocket);
+    }
+
+    public static Socket removeOpenSocket(String socketID)
+    {
+        return _openSocketRegister.remove(socketID);
+    }
+
     public static ITransportConnection getInstance(BrokerDetails details) throws AMQTransportConnectionException
     {
         int transport = getTransport(details.getTransport());
@@ -87,7 +104,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 +152,11 @@ public class TransportConnection
 
     private static int getTransport(String transport)
     {
+        if (transport.equals(BrokerDetails.SOCKET))
+        {
+            return SOCKET;
+        }
+
         if (transport.equals(BrokerDetails.TCP))
         {
             return TCP;
@@ -294,7 +324,7 @@ public class TransportConnection
         synchronized (_inVmPipeAddress)
         {
             _inVmPipeAddress.clear();
-        }        
+        }
         _acceptor = null;
         _currentInstance = -1;
         _currentVMPort = -1;
diff --git a/qpid/java/client/src/main/java/org/apache/qpid/jms/BrokerDetails.java b/qpid/java/client/src/main/java/org/apache/qpid/jms/BrokerDetails.java
index 2892f75516..72ba16086d 100644
--- a/qpid/java/client/src/main/java/org/apache/qpid/jms/BrokerDetails.java
+++ b/qpid/java/client/src/main/java/org/apache/qpid/jms/BrokerDetails.java
@@ -36,6 +36,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/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/SelectorTest.java b/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/SelectorTest.java
index ed4f6041df..2e8bbb0b28 100644
--- a/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/SelectorTest.java
+++ b/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/SelectorTest.java
@@ -71,7 +71,7 @@ public class SelectorTest extends QpidTestCase  implements MessageListener
         connection.start();
 
         String selector = null;
-        // selector = "Cost = 2 AND JMSDeliveryMode=" + DeliveryMode.NON_PERSISTENT;
+         selector = "Cost = 2 AND \"property-with-hyphen\" = 'wibble'";
         // selector = "JMSType = Special AND Cost = 2 AND AMQMessageID > 0 AND JMSDeliveryMode=" + DeliveryMode.NON_PERSISTENT;
 
         _session = (AMQSession) connection.createSession(false, AMQSession.NO_ACKNOWLEDGE);
@@ -86,6 +86,7 @@ public class SelectorTest extends QpidTestCase  implements MessageListener
             Message msg = _session.createTextMessage("Message");
             msg.setJMSPriority(1);
             msg.setIntProperty("Cost", 2);
+            msg.setStringProperty("property-with-hyphen","wibble");
             msg.setJMSType("Special");
 
             _logger.info("Sending Message:" + msg);
diff --git a/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/connectionurl/ConnectionURLTest.java b/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/connectionurl/ConnectionURLTest.java
index bfbba61913..31f95559f3 100644
--- a/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/connectionurl/ConnectionURLTest.java
+++ b/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/connectionurl/ConnectionURLTest.java
@@ -473,6 +473,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()
     {
diff --git a/qpid/java/common/src/main/java/org/apache/qpid/framing/FieldTable.java b/qpid/java/common/src/main/java/org/apache/qpid/framing/FieldTable.java
index 55968ffe19..7f851ab264 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/framing/FieldTable.java
+++ b/qpid/java/common/src/main/java/org/apache/qpid/framing/FieldTable.java
@@ -898,13 +898,15 @@ public class FieldTable
         if (_encodedForm != null)
         {
 
-            if (_encodedForm.position() != 0)
+            ByteBuffer encodedForm = _encodedForm.duplicate();
+
+            if (encodedForm.position() != 0)
             {
-                _encodedForm.flip();
+                encodedForm.flip();
             }
             // _encodedForm.limit((int)getEncodedSize());
 
-            buffer.put(_encodedForm);
+            buffer.put(encodedForm);
         }
         else if (_properties != null)
         {