activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From cshan...@apache.org
Subject [3/3] activemq git commit: https://issues.apache.org/jira/browse/AMQ-5889
Date Tue, 11 Aug 2015 20:03:14 GMT
https://issues.apache.org/jira/browse/AMQ-5889

Adding support for auto detection of wire protocols over a transport.
OpenWire, AMQP, STOMP, and MQTT can all be detected and the broker
will properly handle each one over a given Transport.  Currently
auto TCP, NIO, SSL, and NIO+SSL transports can handle auto-detection
of the wire format and client but support could be added in the
future for other transports like websockets.


Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/04ee70a1
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/04ee70a1
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/04ee70a1

Branch: refs/heads/master
Commit: 04ee70a161b463f69692debea623d754a5c781c2
Parents: 9f50ce3
Author: Christopher L. Shannon (cshannon) <christopher.l.shannon@gmail.com>
Authored: Thu Jul 16 18:23:24 2015 +0000
Committer: Christopher L. Shannon (cshannon) <christopher.l.shannon@gmail.com>
Committed: Tue Aug 11 19:39:29 2015 +0000

----------------------------------------------------------------------
 activemq-amqp/pom.xml                           |  21 ++
 .../transport/amqp/AmqpNioSslTransport.java     |  43 ++-
 .../amqp/AmqpNioSslTransportFactory.java        |  10 +
 .../transport/amqp/AmqpNioTransport.java        |  33 +-
 .../transport/amqp/AmqpNioTransportFactory.java |   7 +
 .../transport/amqp/AmqpTestSupport.java         |  53 ++++
 .../amqp/auto/JMSClientAutoNioPlusSslTest.java  |  51 ++++
 .../amqp/auto/JMSClientAutoNioTest.java         |  51 ++++
 .../amqp/auto/JMSClientAutoPlusSslTest.java     |  51 ++++
 .../transport/amqp/auto/JMSClientAutoTest.java  |  50 ++++
 activemq-broker/pom.xml                         |  21 ++
 .../activemq/broker/TransportConnection.java    |   1 +
 .../transport/auto/AutoSslTransportFactory.java | 117 ++++++++
 .../transport/auto/AutoSslTransportServer.java  | 149 ++++++++++
 .../transport/auto/AutoTcpTransportFactory.java | 110 +++++++
 .../transport/auto/AutoTcpTransportServer.java  | 298 +++++++++++++++++++
 .../transport/auto/AutoTransportUtils.java      |  62 ++++
 .../auto/nio/AutoNIOSSLTransportServer.java     | 122 ++++++++
 .../transport/auto/nio/AutoNIOTransport.java    |  85 ++++++
 .../auto/nio/AutoNioSslTransportFactory.java    | 131 ++++++++
 .../auto/nio/AutoNioTransportFactory.java       | 114 +++++++
 .../protocol/AmqpProtocolVerifier.java          |  36 +++
 .../protocol/MqttProtocolVerifier.java          |  45 +++
 .../protocol/OpenWireProtocolVerifier.java      |  67 +++++
 .../transport/protocol/ProtocolVerifier.java    |   9 +
 .../protocol/StompProtocolVerifier.java         |  39 +++
 .../transport/nio/AutoInitNioSSLTransport.java  | 234 +++++++++++++++
 .../services/org/apache/activemq/transport/auto |  17 ++
 .../org/apache/activemq/transport/auto+nio      |  17 ++
 .../org/apache/activemq/transport/auto+nio+ssl  |  17 ++
 .../org/apache/activemq/transport/auto+ssl      |  17 ++
 activemq-camel/pom.xml                          |  22 +-
 activemq-client/pom.xml                         |  21 ++
 .../activemq/ActiveMQConnectionFactory.java     |  18 +-
 .../activemq/transport/TransportFactory.java    |   7 +
 .../activemq/transport/nio/NIOSSLTransport.java | 109 +++++--
 .../transport/nio/NIOSSLTransportFactory.java   |  16 +
 .../transport/nio/NIOSSLTransportServer.java    |   2 +-
 .../activemq/transport/nio/NIOTransport.java    |  16 +-
 .../transport/nio/NIOTransportFactory.java      |  20 ++
 .../activemq/transport/tcp/SslTransport.java    |  22 +-
 .../transport/tcp/TcpBufferedInputStream.java   |  24 +-
 .../activemq/transport/tcp/TcpTransport.java    |  65 +++-
 .../transport/tcp/TcpTransportFactory.java      |  13 +
 .../transport/tcp/TcpTransportServer.java       |  44 ++-
 activemq-http/pom.xml                           |  22 +-
 activemq-jaas/pom.xml                           |  21 ++
 activemq-jms-pool/pom.xml                       |  21 ++
 activemq-kahadb-store/pom.xml                   |  23 +-
 activemq-karaf-itest/pom.xml                    |  21 ++
 activemq-leveldb-store/pom.xml                  |  21 ++
 activemq-mqtt/pom.xml                           |  21 ++
 .../transport/mqtt/MQTTNIOSSLTransport.java     |  23 +-
 .../mqtt/MQTTNIOSSLTransportFactory.java        |  12 +
 .../transport/mqtt/MQTTNIOTransport.java        |  34 ++-
 .../transport/mqtt/MQTTNIOTransportFactory.java |   7 +
 .../transport/mqtt/auto/MQTTAutoNioSslTest.java |  35 +++
 .../transport/mqtt/auto/MQTTAutoNioTest.java    |  35 +++
 .../transport/mqtt/auto/MQTTAutoSslTest.java    |  35 +++
 .../transport/mqtt/auto/MQTTAutoTest.java       |  35 +++
 activemq-partition/pom.xml                      |  21 ++
 activemq-pool/pom.xml                           |  24 ++
 activemq-ra/pom.xml                             |  21 ++
 activemq-runtime-config/pom.xml                 |  21 ++
 activemq-spring/pom.xml                         |  21 ++
 activemq-stomp/pom.xml                          |  21 ++
 .../transport/stomp/StompNIOSSLTransport.java   |  24 +-
 .../stomp/StompNIOSSLTransportFactory.java      |  10 +
 .../transport/stomp/StompNIOTransport.java      |  38 ++-
 .../stomp/StompNIOTransportFactory.java         |  13 +
 .../transport/stomp/StompTestSupport.java       |  56 +++-
 .../stomp/auto/StompAutoNioSslTest.java         |  44 +++
 .../transport/stomp/auto/StompAutoNioTest.java  |  40 +++
 .../transport/stomp/auto/StompAutoSslTest.java  |  44 +++
 .../transport/stomp/auto/StompAutoTest.java     |  40 +++
 activemq-unit-tests/pom.xml                     |  21 ++
 .../auto/AutoNIOSslTransportBrokerTest.java     |  67 +++++
 .../auto/AutoNIOTransportBrokerTest.java        |  38 +++
 .../auto/AutoSslTransportBrokerTest.java        |  64 ++++
 .../transport/auto/AutoTransportBrokerTest.java |  38 +++
 .../auto/AutoTransportConfigureTest.java        | 156 ++++++++++
 .../auto/failover/AutoFailoverClusterTest.java  |  27 ++
 .../auto/failover/AutoFailoverTimeoutTest.java  |  27 ++
 .../auto/failover/FailoverAutoRandomTest.java   |  28 ++
 .../FailoverAutoTransportBrokerTest.java        |  43 +++
 .../AutoNIOJmsDurableTopicSendReceiveTest.java  |  29 ++
 .../auto/nio/AutoNIOJmsSendAndReceiveTest.java  |  32 ++
 .../AutoNIOPersistentSendAndReceiveTest.java    |  27 ++
 .../transport/auto/nio/AutoNIOSSLBasicTest.java |  27 ++
 .../transport/failover/FailoverClusterTest.java |  12 +-
 .../transport/failover/FailoverRandomTest.java  |  22 +-
 .../transport/failover/FailoverTimeoutTest.java |   4 +
 .../activemq/transport/nio/NIOSSLBasicTest.java |  10 +-
 .../nio/NIOSSLTransportBrokerTest.java          |   4 +-
 activemq-web-demo/pom.xml                       |  21 ++
 activemq-web/pom.xml                            |  21 ++
 assembly/pom.xml                                |  21 ++
 97 files changed, 3912 insertions(+), 108 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/activemq/blob/04ee70a1/activemq-amqp/pom.xml
----------------------------------------------------------------------
diff --git a/activemq-amqp/pom.xml b/activemq-amqp/pom.xml
index 4d3c8ef..5f10825 100644
--- a/activemq-amqp/pom.xml
+++ b/activemq-amqp/pom.xml
@@ -205,6 +205,27 @@
         </plugins>
       </build>
     </profile>
+    <profile>
+      <id>activemq.tests-autoTransport</id>
+      <activation>
+        <property>
+          <name>activemq.tests</name>
+          <value>autoTransport</value>
+        </property>
+      </activation>
+      <build>
+        <plugins>
+          <plugin>
+            <artifactId>maven-surefire-plugin</artifactId>
+            <configuration>
+              <includes>
+                <include>**/auto/*Test.java</include>
+              </includes>
+            </configuration>
+          </plugin>
+        </plugins>
+      </build>
+    </profile>
 
       <profile>
           <id>activemq.tests.windows.excludes</id>

http://git-wip-us.apache.org/repos/asf/activemq/blob/04ee70a1/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpNioSslTransport.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpNioSslTransport.java b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpNioSslTransport.java
index 3276be9..88d9284 100644
--- a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpNioSslTransport.java
+++ b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpNioSslTransport.java
@@ -23,6 +23,7 @@ import java.net.UnknownHostException;
 import java.nio.ByteBuffer;
 
 import javax.net.SocketFactory;
+import javax.net.ssl.SSLEngine;
 
 import org.apache.activemq.transport.nio.NIOSSLTransport;
 import org.apache.activemq.wireformat.WireFormat;
@@ -38,7 +39,14 @@ public class AmqpNioSslTransport extends NIOSSLTransport {
     }
 
     public AmqpNioSslTransport(WireFormat wireFormat, Socket socket) throws IOException {
-        super(wireFormat, socket);
+        super(wireFormat, socket, null, null, null);
+
+        frameReader.setWireFormat((AmqpWireFormat) wireFormat);
+    }
+
+    public AmqpNioSslTransport(WireFormat wireFormat, Socket socket,
+            SSLEngine engine, InitBuffer initBuffer, ByteBuffer inputBuffer) throws IOException {
+        super(wireFormat, socket, engine, initBuffer, inputBuffer);
 
         frameReader.setWireFormat((AmqpWireFormat) wireFormat);
     }
@@ -55,4 +63,37 @@ public class AmqpNioSslTransport extends NIOSSLTransport {
     protected void processCommand(ByteBuffer plain) throws Exception {
         frameReader.parse(plain);
     }
+
+    /* (non-Javadoc)
+     * @see org.apache.activemq.transport.nio.NIOSSLTransport#secureRead(java.nio.ByteBuffer)
+     */
+
+    @Override
+    protected void doInit() {
+        if (initBuffer != null) {
+            nextFrameSize = -1;
+            serviceRead();
+
+        }
+    }
+
+    @Override
+    protected int secureRead(ByteBuffer plain) throws Exception {
+        if (initBuffer != null) {
+            initBuffer.buffer.flip();
+            if (initBuffer.buffer.hasRemaining()) {
+                plain.flip();
+                for (int i =0; i < 8; i++) {
+                    plain.put(initBuffer.buffer.get());
+                }
+                plain.flip();
+                processCommand(plain);
+                initBuffer.buffer.clear();
+                return 8;
+            }
+        }
+        return super.secureRead(plain);
+    }
+
+
 }
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/activemq/blob/04ee70a1/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpNioSslTransportFactory.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpNioSslTransportFactory.java b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpNioSslTransportFactory.java
index 5e2fa06..7858a56 100644
--- a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpNioSslTransportFactory.java
+++ b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpNioSslTransportFactory.java
@@ -21,15 +21,18 @@ import java.net.Socket;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.net.UnknownHostException;
+import java.nio.ByteBuffer;
 
 import javax.net.ServerSocketFactory;
 import javax.net.SocketFactory;
 import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLEngine;
 
 import org.apache.activemq.broker.SslContext;
 import org.apache.activemq.transport.Transport;
 import org.apache.activemq.transport.TransportServer;
 import org.apache.activemq.transport.tcp.TcpTransport;
+import org.apache.activemq.transport.tcp.TcpTransport.InitBuffer;
 import org.apache.activemq.transport.tcp.TcpTransportServer;
 import org.apache.activemq.wireformat.WireFormat;
 
@@ -62,6 +65,13 @@ public class AmqpNioSslTransportFactory extends AmqpNioTransportFactory {
     }
 
     @Override
+    public TcpTransport createTransport(WireFormat wireFormat, Socket socket,
+            SSLEngine engine, InitBuffer initBuffer, ByteBuffer inputBuffer)
+            throws IOException {
+        return new AmqpNioSslTransport(wireFormat, socket, engine, initBuffer, inputBuffer);
+    }
+
+    @Override
     public TransportServer doBind(URI location) throws IOException {
         if (SslContext.getCurrentSslContext() != null) {
             try {

http://git-wip-us.apache.org/repos/asf/activemq/blob/04ee70a1/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpNioTransport.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpNioTransport.java b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpNioTransport.java
index 21d40eb..dfa1569 100644
--- a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpNioTransport.java
+++ b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpNioTransport.java
@@ -16,6 +16,7 @@
  */
 package org.apache.activemq.transport.amqp;
 
+import java.io.ByteArrayInputStream;
 import java.io.DataOutputStream;
 import java.io.EOFException;
 import java.io.IOException;
@@ -63,6 +64,12 @@ public class AmqpNioTransport extends TcpTransport {
         frameReader.setWireFormat((AmqpWireFormat) wireFormat);
     }
 
+    public AmqpNioTransport(WireFormat wireFormat, Socket socket, InitBuffer initBuffer) throws IOException {
+        super(wireFormat, socket, initBuffer);
+
+        frameReader.setWireFormat((AmqpWireFormat) wireFormat);
+    }
+
     @Override
     protected void initializeStreams() throws IOException {
         channel = socket.getChannel();
@@ -91,6 +98,17 @@ public class AmqpNioTransport extends TcpTransport {
         NIOOutputStream outPutStream = new NIOOutputStream(channel, 8 * 1024);
         this.dataOut = new DataOutputStream(outPutStream);
         this.buffOut = outPutStream;
+
+        try {
+            if (initBuffer != null) {
+                processBuffer(initBuffer.buffer, initBuffer.readSize);
+            }
+        } catch (IOException e) {
+            onException(e);
+        } catch (Throwable e) {
+            onException(IOExceptionSupport.create(e));
+        }
+
     }
 
     boolean magicRead = false;
@@ -101,6 +119,7 @@ public class AmqpNioTransport extends TcpTransport {
             while (isStarted()) {
                 // read channel
                 int readSize = channel.read(inputBuffer);
+
                 // channel is closed, cleanup
                 if (readSize == -1) {
                     onException(new EOFException());
@@ -112,11 +131,7 @@ public class AmqpNioTransport extends TcpTransport {
                     break;
                 }
 
-                receiveCounter += readSize;
-
-                inputBuffer.flip();
-                frameReader.parse(inputBuffer);
-                inputBuffer.clear();
+                processBuffer(inputBuffer, readSize);
             }
         } catch (IOException e) {
             onException(e);
@@ -125,6 +140,14 @@ public class AmqpNioTransport extends TcpTransport {
         }
     }
 
+    protected void processBuffer(ByteBuffer buffer, int readSize) throws Exception {
+        receiveCounter += readSize;
+
+        buffer.flip();
+        frameReader.parse(buffer);
+        buffer.clear();
+    }
+
     @Override
     protected void doStart() throws Exception {
         connect();

http://git-wip-us.apache.org/repos/asf/activemq/blob/04ee70a1/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpNioTransportFactory.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpNioTransportFactory.java b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpNioTransportFactory.java
index d54d58f..4b6d9bd 100644
--- a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpNioTransportFactory.java
+++ b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpNioTransportFactory.java
@@ -33,6 +33,7 @@ import org.apache.activemq.transport.MutexTransport;
 import org.apache.activemq.transport.Transport;
 import org.apache.activemq.transport.nio.NIOTransportFactory;
 import org.apache.activemq.transport.tcp.TcpTransport;
+import org.apache.activemq.transport.tcp.TcpTransport.InitBuffer;
 import org.apache.activemq.transport.tcp.TcpTransportServer;
 import org.apache.activemq.util.IntrospectionSupport;
 import org.apache.activemq.wireformat.WireFormat;
@@ -64,6 +65,12 @@ public class AmqpNioTransportFactory extends NIOTransportFactory implements Brok
         return new AmqpNioTransport(wf, socketFactory, location, localLocation);
     }
 
+    @Override
+    public TcpTransport createTransport(WireFormat wireFormat, Socket socket,
+            InitBuffer initBuffer) throws IOException {
+        return new AmqpNioTransport(wireFormat, socket, initBuffer);
+    }
+
     @SuppressWarnings("rawtypes")
     @Override
     public Transport serverConfigure(Transport transport, WireFormat format, HashMap options) throws Exception {

http://git-wip-us.apache.org/repos/asf/activemq/blob/04ee70a1/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AmqpTestSupport.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AmqpTestSupport.java b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AmqpTestSupport.java
index 91909d4..45af469 100644
--- a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AmqpTestSupport.java
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AmqpTestSupport.java
@@ -79,6 +79,15 @@ public class AmqpTestSupport {
     protected URI amqpNioPlusSslURI;
     protected int amqpNioPlusSslPort;
 
+    protected URI autoURI;
+    protected int autoPort;
+    protected URI autoSslURI;
+    protected int autoSslPort;
+    protected URI autoNioURI;
+    protected int autoNioPort;
+    protected URI autoNioPlusSslURI;
+    protected int autoNioPlusSslPort;
+
     protected URI openwireURI;
     protected int openwirePort;
 
@@ -176,6 +185,34 @@ public class AmqpTestSupport {
             amqpNioPlusSslURI = connector.getPublishableConnectURI();
             LOG.debug("Using amqp+nio+ssl port " + amqpNioPlusSslPort);
         }
+        if (isUseAutoConnector()) {
+            connector = brokerService.addConnector(
+                "auto://0.0.0.0:" + autoPort + getAdditionalConfig());
+            autoPort = connector.getConnectUri().getPort();
+            autoURI = connector.getPublishableConnectURI();
+            LOG.debug("Using auto port " + autoPort);
+        }
+        if (isUseAutoSslConnector()) {
+            connector = brokerService.addConnector(
+                "auto+ssl://0.0.0.0:" + autoSslPort + getAdditionalConfig());
+            autoSslPort = connector.getConnectUri().getPort();
+            autoSslURI = connector.getPublishableConnectURI();
+            LOG.debug("Using auto+ssl port " + autoSslPort);
+        }
+        if (isUseAutoNioConnector()) {
+            connector = brokerService.addConnector(
+                "auto+nio://0.0.0.0:" + autoNioPort + getAdditionalConfig());
+            autoNioPort = connector.getConnectUri().getPort();
+            autoNioURI = connector.getPublishableConnectURI();
+            LOG.debug("Using auto+nio port " + autoNioPort);
+        }
+        if (isUseAutoNioPlusSslConnector()) {
+            connector = brokerService.addConnector(
+                "auto+nio+ssl://0.0.0.0:" + autoNioPlusSslPort + getAdditionalConfig());
+            autoNioPlusSslPort = connector.getConnectUri().getPort();
+            autoNioPlusSslURI = connector.getPublishableConnectURI();
+            LOG.debug("Using auto+nio+ssl port " + autoNioPlusSslPort);
+        }
     }
 
     protected boolean isPersistent() {
@@ -206,6 +243,22 @@ public class AmqpTestSupport {
         return false;
     }
 
+    protected boolean isUseAutoConnector() {
+        return false;
+    }
+
+    protected boolean isUseAutoSslConnector() {
+        return false;
+    }
+
+    protected boolean isUseAutoNioConnector() {
+        return false;
+    }
+
+    protected boolean isUseAutoNioPlusSslConnector() {
+        return false;
+    }
+
     protected String getAmqpTransformer() {
         return "jms";
     }

http://git-wip-us.apache.org/repos/asf/activemq/blob/04ee70a1/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/auto/JMSClientAutoNioPlusSslTest.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/auto/JMSClientAutoNioPlusSslTest.java b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/auto/JMSClientAutoNioPlusSslTest.java
new file mode 100644
index 0000000..0bf2e38
--- /dev/null
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/auto/JMSClientAutoNioPlusSslTest.java
@@ -0,0 +1,51 @@
+/**
+ * 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.activemq.transport.amqp.auto;
+
+import java.net.URI;
+
+import org.apache.activemq.transport.amqp.JMSClientNioPlusSslTest;
+import org.apache.activemq.transport.amqp.JMSClientSslTest;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Test the JMS client when connected to the NIO+SSL transport.
+ */
+public class JMSClientAutoNioPlusSslTest extends JMSClientSslTest {
+    protected static final Logger LOG = LoggerFactory.getLogger(JMSClientAutoNioPlusSslTest.class);
+
+    @Override
+    protected URI getBrokerURI() {
+        return autoNioPlusSslURI;
+    }
+
+    @Override
+    protected boolean isUseTcpConnector() {
+        return false;
+    }
+
+    @Override
+    protected boolean isUseAutoNioPlusSslConnector() {
+        return true;
+    }
+
+    @Override
+    protected String getTargetConnectorName() {
+        return "auto+nio+ssl";
+    }
+}

http://git-wip-us.apache.org/repos/asf/activemq/blob/04ee70a1/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/auto/JMSClientAutoNioTest.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/auto/JMSClientAutoNioTest.java b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/auto/JMSClientAutoNioTest.java
new file mode 100644
index 0000000..5346a79
--- /dev/null
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/auto/JMSClientAutoNioTest.java
@@ -0,0 +1,51 @@
+/**
+ * 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.activemq.transport.amqp.auto;
+
+import java.net.URI;
+
+import org.apache.activemq.transport.amqp.JMSClientNioTest;
+import org.apache.activemq.transport.amqp.JMSClientTest;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Test the JMS client when connected to the NIO transport.
+ */
+public class JMSClientAutoNioTest extends JMSClientTest {
+    protected static final Logger LOG = LoggerFactory.getLogger(JMSClientAutoNioTest.class);
+
+    @Override
+    protected URI getBrokerURI() {
+        return autoNioURI;
+    }
+
+    @Override
+    protected boolean isUseTcpConnector() {
+        return false;
+    }
+
+    @Override
+    protected boolean isUseAutoNioConnector() {
+        return true;
+    }
+
+    @Override
+    protected String getTargetConnectorName() {
+        return "auto+nio";
+    }
+}

http://git-wip-us.apache.org/repos/asf/activemq/blob/04ee70a1/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/auto/JMSClientAutoPlusSslTest.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/auto/JMSClientAutoPlusSslTest.java b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/auto/JMSClientAutoPlusSslTest.java
new file mode 100644
index 0000000..deb90fb
--- /dev/null
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/auto/JMSClientAutoPlusSslTest.java
@@ -0,0 +1,51 @@
+/**
+ * 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.activemq.transport.amqp.auto;
+
+import java.net.URI;
+
+import org.apache.activemq.transport.amqp.JMSClientNioPlusSslTest;
+import org.apache.activemq.transport.amqp.JMSClientSslTest;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Test the JMS client when connected to the NIO+SSL transport.
+ */
+public class JMSClientAutoPlusSslTest extends JMSClientSslTest {
+    protected static final Logger LOG = LoggerFactory.getLogger(JMSClientAutoPlusSslTest.class);
+
+    @Override
+    protected URI getBrokerURI() {
+        return autoSslURI;
+    }
+
+    @Override
+    protected boolean isUseTcpConnector() {
+        return false;
+    }
+
+    @Override
+    protected boolean isUseAutoSslConnector() {
+        return true;
+    }
+
+    @Override
+    protected String getTargetConnectorName() {
+        return "auto+ssl";
+    }
+}

http://git-wip-us.apache.org/repos/asf/activemq/blob/04ee70a1/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/auto/JMSClientAutoTest.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/auto/JMSClientAutoTest.java b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/auto/JMSClientAutoTest.java
new file mode 100644
index 0000000..77683e4
--- /dev/null
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/auto/JMSClientAutoTest.java
@@ -0,0 +1,50 @@
+/**
+ * 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.activemq.transport.amqp.auto;
+
+import java.net.URI;
+
+import org.apache.activemq.transport.amqp.JMSClientTest;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Test the JMS client when connected to the NIO transport.
+ */
+public class JMSClientAutoTest extends JMSClientTest {
+    protected static final Logger LOG = LoggerFactory.getLogger(JMSClientAutoTest.class);
+
+    @Override
+    protected boolean isUseTcpConnector() {
+        return false;
+    }
+
+    @Override
+    protected boolean isUseAutoConnector() {
+        return true;
+    }
+
+    @Override
+    protected URI getBrokerURI() {
+        return autoURI;
+    }
+
+    @Override
+    protected String getTargetConnectorName() {
+        return "auto";
+    }
+}

http://git-wip-us.apache.org/repos/asf/activemq/blob/04ee70a1/activemq-broker/pom.xml
----------------------------------------------------------------------
diff --git a/activemq-broker/pom.xml b/activemq-broker/pom.xml
index bc35400..5f8545b 100755
--- a/activemq-broker/pom.xml
+++ b/activemq-broker/pom.xml
@@ -238,5 +238,26 @@
         </plugins>
       </build>
     </profile>
+	<profile>
+      <id>activemq.tests-autoTransport</id>
+      <activation>
+        <property>
+          <name>activemq.tests</name>
+          <value>autoTransport</value>
+        </property>
+      </activation>
+      <build>
+        <plugins>
+          <plugin>
+            <artifactId>maven-surefire-plugin</artifactId>
+            <configuration>
+              <excludes>
+                <exclude>**</exclude>
+              </excludes>
+            </configuration>
+          </plugin>
+        </plugins>
+      </build>
+    </profile>
   </profiles>
 </project>

http://git-wip-us.apache.org/repos/asf/activemq/blob/04ee70a1/activemq-broker/src/main/java/org/apache/activemq/broker/TransportConnection.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/TransportConnection.java b/activemq-broker/src/main/java/org/apache/activemq/broker/TransportConnection.java
index 2727503..a9d36b5 100755
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/TransportConnection.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/TransportConnection.java
@@ -239,6 +239,7 @@ public class TransportConnection implements Connection, Task, CommandVisitor {
         }
         if (!stopping.get() && !pendingStop) {
             transportException.set(e);
+            e.printStackTrace();
             if (TRANSPORTLOG.isDebugEnabled()) {
                 TRANSPORTLOG.debug(this + " failed: " + e, e);
             } else if (TRANSPORTLOG.isWarnEnabled() && !expected(e)) {

http://git-wip-us.apache.org/repos/asf/activemq/blob/04ee70a1/activemq-broker/src/main/java/org/apache/activemq/broker/transport/auto/AutoSslTransportFactory.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/transport/auto/AutoSslTransportFactory.java b/activemq-broker/src/main/java/org/apache/activemq/broker/transport/auto/AutoSslTransportFactory.java
new file mode 100644
index 0000000..7c65311
--- /dev/null
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/transport/auto/AutoSslTransportFactory.java
@@ -0,0 +1,117 @@
+/**
+ * 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.activemq.broker.transport.auto;
+
+import java.io.IOException;
+import java.net.Socket;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+import javax.net.ServerSocketFactory;
+import javax.net.ssl.SSLServerSocketFactory;
+
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.BrokerServiceAware;
+import org.apache.activemq.transport.Transport;
+import org.apache.activemq.transport.TransportServer;
+import org.apache.activemq.transport.tcp.SslTransportFactory;
+import org.apache.activemq.transport.tcp.SslTransportServer;
+import org.apache.activemq.transport.tcp.TcpTransport;
+import org.apache.activemq.util.IOExceptionSupport;
+import org.apache.activemq.util.IntrospectionSupport;
+import org.apache.activemq.util.URISupport;
+import org.apache.activemq.wireformat.WireFormat;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class AutoSslTransportFactory extends SslTransportFactory implements BrokerServiceAware  {
+    private static final Logger LOG = LoggerFactory.getLogger(AutoSslTransportFactory.class);
+
+
+    protected BrokerService brokerService;
+    /* (non-Javadoc)
+     * @see org.apache.activemq.broker.BrokerServiceAware#setBrokerService(org.apache.activemq.broker.BrokerService)
+     */
+    @Override
+    public void setBrokerService(BrokerService brokerService) {
+        this.brokerService = brokerService;
+    }
+
+    private Set<String> enabledProtocols;
+
+    /**
+     * Overriding to use SslTransportServer and allow for proper reflection.
+     */
+    @Override
+    public TransportServer doBind(final URI location) throws IOException {
+        try {
+            Map<String, String> options = new HashMap<String, String>(URISupport.parseParameters(location));
+
+            Map<String, Object> autoProperties = IntrospectionSupport.extractProperties(options, "auto.");
+            this.enabledProtocols = AutoTransportUtils.parseProtocols((String) autoProperties.get("protocols"));
+
+            ServerSocketFactory serverSocketFactory = createServerSocketFactory();
+            AutoSslTransportServer server = createAutoSslTransportServer(location, (SSLServerSocketFactory)serverSocketFactory);
+            if (options.get("allowLinkStealing") != null){
+                allowLinkStealingSet = true;
+            }
+            IntrospectionSupport.setProperties(server, options);
+            server.setAutoTransportOptions(IntrospectionSupport.extractProperties(options, "auto."));
+            server.setTransportOption(IntrospectionSupport.extractProperties(options, "transport."));
+            server.setWireFormatOptions(AutoTransportUtils.extractWireFormatOptions(options));
+            server.bind();
+
+            return server;
+        } catch (URISyntaxException e) {
+            throw IOExceptionSupport.create(e);
+        }
+    }
+
+    boolean allowLinkStealingSet = false;
+
+    /**
+     * Allows subclasses of SslTransportFactory to create custom instances of
+     * SslTransportServer.
+     *
+     * @param location
+     * @param serverSocketFactory
+     * @return
+     * @throws IOException
+     * @throws URISyntaxException
+     */
+   // @Override
+    protected AutoSslTransportServer createAutoSslTransportServer(final URI location, SSLServerSocketFactory serverSocketFactory) throws IOException, URISyntaxException {
+        AutoSslTransportServer server = new AutoSslTransportServer(this, location, serverSocketFactory,
+                this.brokerService, enabledProtocols) {
+            @Override
+            protected TcpTransport createTransport(Socket socket, WireFormat format)
+                    throws IOException {
+                if (format.getClass().toString().contains("MQTT") && !allowLinkStealingSet) {
+                    this.setAllowLinkStealing(true);
+                }
+                return super.createTransport(socket, format);
+            }
+        };
+        return server;
+    }
+
+
+}

http://git-wip-us.apache.org/repos/asf/activemq/blob/04ee70a1/activemq-broker/src/main/java/org/apache/activemq/broker/transport/auto/AutoSslTransportServer.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/transport/auto/AutoSslTransportServer.java b/activemq-broker/src/main/java/org/apache/activemq/broker/transport/auto/AutoSslTransportServer.java
new file mode 100644
index 0000000..9954523
--- /dev/null
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/transport/auto/AutoSslTransportServer.java
@@ -0,0 +1,149 @@
+/**
+ * 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.activemq.broker.transport.auto;
+
+import java.io.IOException;
+import java.net.Socket;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.Set;
+
+import javax.net.ServerSocketFactory;
+import javax.net.ssl.SSLServerSocket;
+import javax.net.ssl.SSLServerSocketFactory;
+import javax.net.ssl.SSLSocket;
+
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.transport.Transport;
+import org.apache.activemq.transport.tcp.SslTransport;
+import org.apache.activemq.transport.tcp.SslTransportFactory;
+import org.apache.activemq.transport.tcp.TcpTransport;
+import org.apache.activemq.transport.tcp.TcpTransportFactory;
+import org.apache.activemq.wireformat.WireFormat;
+
+/**
+ *  An SSL TransportServer.
+ *
+ *  Allows for client certificate authentication (refer to setNeedClientAuth for
+ *      details).
+ *  NOTE: Client certificate authentication is disabled by default.
+ *
+ */
+public class AutoSslTransportServer extends AutoTcpTransportServer {
+
+
+
+    // Specifies if sockets created from this server should needClientAuth.
+    private boolean needClientAuth;
+
+    // Specifies if sockets created from this server should wantClientAuth.
+    private boolean wantClientAuth;
+
+//    /**
+//     * Creates a ssl transport server for the specified url using the provided
+//     * serverSocketFactory
+//     *
+//     * @param transportFactory The factory used to create transports when connections arrive.
+//     * @param location The location of the broker to bind to.
+//     * @param serverSocketFactory The factory used to create this server.
+//     * @throws IOException passed up from TcpTransportFactory.
+//     * @throws URISyntaxException passed up from TcpTransportFactory.
+//     */
+//    public SslTransportServer(SslTransportFactory transportFactory, URI location, SSLServerSocketFactory serverSocketFactory) throws IOException, URISyntaxException {
+//        super(transportFactory, location, serverSocketFactory);
+//    }
+
+    public AutoSslTransportServer(SslTransportFactory transportFactory,
+            URI location, SSLServerSocketFactory serverSocketFactory,
+            BrokerService brokerService, Set<String> enabledProtocols) throws IOException, URISyntaxException {
+        super(transportFactory, location, serverSocketFactory, brokerService, enabledProtocols);
+        // TODO Auto-generated constructor stub
+    }
+
+    /**
+     * Sets whether client authentication should be required
+     * Must be called before {@link #bind()}
+     * Note: Calling this method clears the wantClientAuth flag
+     * in the underlying implementation.
+     */
+    public void setNeedClientAuth(boolean needAuth) {
+        this.needClientAuth = needAuth;
+    }
+
+    /**
+     * Returns whether client authentication should be required.
+     */
+    public boolean getNeedClientAuth() {
+        return this.needClientAuth;
+    }
+
+    /**
+     * Returns whether client authentication should be requested.
+     */
+    public boolean getWantClientAuth() {
+        return this.wantClientAuth;
+    }
+
+    /**
+     * Sets whether client authentication should be requested.
+     * Must be called before {@link #bind()}
+     * Note: Calling this method clears the needClientAuth flag
+     * in the underlying implementation.
+     */
+    public void setWantClientAuth(boolean wantAuth) {
+        this.wantClientAuth = wantAuth;
+    }
+
+    /**
+     * Binds this socket to the previously specified URI.
+     *
+     * Overridden to allow for proper handling of needClientAuth.
+     *
+     * @throws IOException passed up from TcpTransportServer.
+     */
+    @Override
+    public void bind() throws IOException {
+        super.bind();
+        if (needClientAuth) {
+            ((SSLServerSocket)this.serverSocket).setNeedClientAuth(true);
+        } else if (wantClientAuth) {
+            ((SSLServerSocket)this.serverSocket).setWantClientAuth(true);
+        }
+    }
+
+    /**
+     * Used to create Transports for this server.
+     *
+     * Overridden to allow the use of SslTransports (instead of TcpTransports).
+     *
+     * @param socket The incoming socket that will be wrapped into the new Transport.
+     * @param format The WireFormat being used.
+     * @return The newly return (SSL) Transport.
+     * @throws IOException
+     */
+    @Override
+    protected TcpTransport createTransport(Socket socket, WireFormat format) throws IOException {
+        return new SslTransport(format, (SSLSocket)socket, this.initBuffer);
+    }
+
+    @Override
+    public boolean isSslServer() {
+        return true;
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/activemq/blob/04ee70a1/activemq-broker/src/main/java/org/apache/activemq/broker/transport/auto/AutoTcpTransportFactory.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/transport/auto/AutoTcpTransportFactory.java b/activemq-broker/src/main/java/org/apache/activemq/broker/transport/auto/AutoTcpTransportFactory.java
new file mode 100644
index 0000000..5731b85
--- /dev/null
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/transport/auto/AutoTcpTransportFactory.java
@@ -0,0 +1,110 @@
+/**
+ * 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.activemq.broker.transport.auto;
+
+import java.io.IOException;
+import java.net.Socket;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+import javax.net.ServerSocketFactory;
+
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.BrokerServiceAware;
+import org.apache.activemq.openwire.OpenWireFormatFactory;
+import org.apache.activemq.transport.MutexTransport;
+import org.apache.activemq.transport.Transport;
+import org.apache.activemq.transport.TransportFactory;
+import org.apache.activemq.transport.TransportFilter;
+import org.apache.activemq.transport.TransportServer;
+import org.apache.activemq.transport.tcp.TcpTransport;
+import org.apache.activemq.transport.tcp.TcpTransportFactory;
+import org.apache.activemq.util.FactoryFinder;
+import org.apache.activemq.util.IOExceptionSupport;
+import org.apache.activemq.util.IntrospectionSupport;
+import org.apache.activemq.util.URISupport;
+import org.apache.activemq.wireformat.WireFormat;
+import org.apache.activemq.wireformat.WireFormatFactory;
+
+/**
+ *
+ *
+ */
+public class AutoTcpTransportFactory extends TcpTransportFactory implements BrokerServiceAware {
+
+    protected BrokerService brokerService;
+    /* (non-Javadoc)
+     * @see org.apache.activemq.broker.BrokerServiceAware#setBrokerService(org.apache.activemq.broker.BrokerService)
+     */
+    @Override
+    public void setBrokerService(BrokerService brokerService) {
+        this.brokerService = brokerService;
+    }
+
+
+    @Override
+    public TransportServer doBind(final URI location) throws IOException {
+        try {
+            Map<String, String> options = new HashMap<String, String>(URISupport.parseParameters(location));
+
+            Map<String, Object> autoProperties = IntrospectionSupport.extractProperties(options, "auto.");
+            this.enabledProtocols = AutoTransportUtils.parseProtocols((String) autoProperties.get("protocols"));
+
+            ServerSocketFactory serverSocketFactory = createServerSocketFactory();
+            AutoTcpTransportServer server = createTcpTransportServer(location, serverSocketFactory);
+            //server.setWireFormatFactory(createWireFormatFactory(options));
+            server.setWireFormatFactory(new OpenWireFormatFactory());
+            if (options.get("allowLinkStealing") != null){
+                allowLinkStealingSet = true;
+            }
+            IntrospectionSupport.setProperties(server, options);
+            server.setTransportOption(IntrospectionSupport.extractProperties(options, "transport."));
+            server.setWireFormatOptions(AutoTransportUtils.extractWireFormatOptions(options));
+            server.bind();
+
+            return server;
+        } catch (URISyntaxException e) {
+            throw IOExceptionSupport.create(e);
+        }
+    }
+
+    boolean allowLinkStealingSet = false;
+    private Set<String> enabledProtocols;
+
+    @Override
+    protected AutoTcpTransportServer createTcpTransportServer(final URI location, ServerSocketFactory serverSocketFactory) throws IOException, URISyntaxException {
+        AutoTcpTransportServer server = new AutoTcpTransportServer(this, location, serverSocketFactory, brokerService, enabledProtocols) {
+
+            @Override
+            protected TcpTransport createTransport(Socket socket, WireFormat format)
+                    throws IOException {
+                if (format.getClass().toString().contains("MQTT") && !allowLinkStealingSet) {
+                    this.setAllowLinkStealing(true);
+                }
+                return super.createTransport(socket, format);
+            }
+
+        };
+
+        return server;
+    }
+}

http://git-wip-us.apache.org/repos/asf/activemq/blob/04ee70a1/activemq-broker/src/main/java/org/apache/activemq/broker/transport/auto/AutoTcpTransportServer.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/transport/auto/AutoTcpTransportServer.java b/activemq-broker/src/main/java/org/apache/activemq/broker/transport/auto/AutoTcpTransportServer.java
new file mode 100644
index 0000000..65a0fe5
--- /dev/null
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/transport/auto/AutoTcpTransportServer.java
@@ -0,0 +1,298 @@
+/**
+ * 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.activemq.broker.transport.auto;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.Socket;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.nio.ByteBuffer;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import javax.net.ServerSocketFactory;
+
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.BrokerServiceAware;
+import org.apache.activemq.broker.transport.protocol.AmqpProtocolVerifier;
+import org.apache.activemq.broker.transport.protocol.MqttProtocolVerifier;
+import org.apache.activemq.broker.transport.protocol.OpenWireProtocolVerifier;
+import org.apache.activemq.broker.transport.protocol.ProtocolVerifier;
+import org.apache.activemq.broker.transport.protocol.StompProtocolVerifier;
+import org.apache.activemq.openwire.OpenWireFormatFactory;
+import org.apache.activemq.transport.Transport;
+import org.apache.activemq.transport.TransportFactory;
+import org.apache.activemq.transport.TransportServer;
+import org.apache.activemq.transport.tcp.TcpTransport;
+import org.apache.activemq.transport.tcp.TcpTransport.InitBuffer;
+import org.apache.activemq.transport.tcp.TcpTransportFactory;
+import org.apache.activemq.transport.tcp.TcpTransportServer;
+import org.apache.activemq.util.FactoryFinder;
+import org.apache.activemq.util.IOExceptionSupport;
+import org.apache.activemq.util.IntrospectionSupport;
+import org.apache.activemq.wireformat.WireFormat;
+import org.apache.activemq.wireformat.WireFormatFactory;
+import org.fusesource.hawtbuf.Buffer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A TCP based implementation of {@link TransportServer}
+ */
+public class AutoTcpTransportServer extends TcpTransportServer {
+
+    private static final Logger LOG = LoggerFactory.getLogger(AutoTcpTransportServer.class);
+
+    protected Map<String, Map<String, Object>> wireFormatOptions;
+    protected Map<String, Object> autoTransportOptions;
+    protected Set<String> enabledProtocols;
+    protected final Map<String, ProtocolVerifier> protocolVerifiers = new ConcurrentHashMap<String, ProtocolVerifier>();
+
+    protected BrokerService brokerService;
+
+    private static final FactoryFinder TRANSPORT_FACTORY_FINDER = new FactoryFinder("META-INF/services/org/apache/activemq/transport/");
+    private final ConcurrentMap<String, TransportFactory> transportFactories = new ConcurrentHashMap<String, TransportFactory>();
+
+    private static final FactoryFinder WIREFORMAT_FACTORY_FINDER = new FactoryFinder("META-INF/services/org/apache/activemq/wireformat/");
+
+    public WireFormatFactory findWireFormatFactory(String scheme, Map<String, Map<String, Object>> options) throws IOException {
+        WireFormatFactory wff = null;
+        try {
+            wff = (WireFormatFactory)WIREFORMAT_FACTORY_FINDER.newInstance(scheme);
+            if (options != null) {
+                IntrospectionSupport.setProperties(wff, options.get(AutoTransportUtils.ALL));
+                IntrospectionSupport.setProperties(wff, options.get(scheme));
+            }
+            if (wff instanceof OpenWireFormatFactory) {
+                protocolVerifiers.put(AutoTransportUtils.OPENWIRE, new OpenWireProtocolVerifier((OpenWireFormatFactory) wff));
+            }
+            return wff;
+        } catch (Throwable e) {
+           throw IOExceptionSupport.create("Could not create wire format factory for: " + scheme + ", reason: " + e, e);
+        }
+    }
+
+    public TransportFactory findTransportFactory(String scheme, Map<String, ?> options) throws IOException {
+        scheme = append(scheme, "nio");
+        scheme = append(scheme, "ssl");
+
+        if (scheme.isEmpty()) {
+            scheme = "tcp";
+        }
+
+        TransportFactory tf = transportFactories.get(scheme);
+        if (tf == null) {
+            // Try to load if from a META-INF property.
+            try {
+                tf = (TransportFactory)TRANSPORT_FACTORY_FINDER.newInstance(scheme);
+                if (options != null)
+                    IntrospectionSupport.setProperties(tf, options);
+                transportFactories.put(scheme, tf);
+            } catch (Throwable e) {
+                throw IOExceptionSupport.create("Transport scheme NOT recognized: [" + scheme + "]", e);
+            }
+        }
+        return tf;
+    }
+
+    protected String append(String currentScheme, String scheme) {
+        if (this.getBindLocation().getScheme().contains(scheme)) {
+            if (!currentScheme.isEmpty()) {
+                currentScheme += "+";
+            }
+            currentScheme += scheme;
+        }
+        return currentScheme;
+    }
+
+    /**
+     * @param transportFactory
+     * @param location
+     * @param serverSocketFactory
+     * @throws IOException
+     * @throws URISyntaxException
+     */
+    public AutoTcpTransportServer(TcpTransportFactory transportFactory,
+            URI location, ServerSocketFactory serverSocketFactory, BrokerService brokerService,
+            Set<String> enabledProtocols)
+            throws IOException, URISyntaxException {
+        super(transportFactory, location, serverSocketFactory);
+        service = Executors.newCachedThreadPool();
+        this.brokerService = brokerService;
+        this.enabledProtocols = enabledProtocols;
+        initProtocolVerifiers();
+    }
+
+    @Override
+    public void setWireFormatFactory(WireFormatFactory factory) {
+        super.setWireFormatFactory(factory);
+        initOpenWireProtocolVerifier();
+    }
+
+    protected void initProtocolVerifiers() {
+        initOpenWireProtocolVerifier();
+
+        if (isAllProtocols() || enabledProtocols.contains(AutoTransportUtils.AMQP)) {
+            protocolVerifiers.put(AutoTransportUtils.AMQP, new AmqpProtocolVerifier());
+        }
+        if (isAllProtocols() || enabledProtocols.contains(AutoTransportUtils.STOMP)) {
+            protocolVerifiers.put(AutoTransportUtils.STOMP, new StompProtocolVerifier());
+        }
+        if (isAllProtocols()|| enabledProtocols.contains(AutoTransportUtils.MQTT)) {
+            protocolVerifiers.put(AutoTransportUtils.MQTT, new MqttProtocolVerifier());
+        }
+    }
+
+    protected void initOpenWireProtocolVerifier() {
+        if (isAllProtocols() || enabledProtocols.contains(AutoTransportUtils.OPENWIRE)) {
+            OpenWireProtocolVerifier owpv;
+            if (wireFormatFactory instanceof OpenWireFormatFactory) {
+                owpv = new OpenWireProtocolVerifier((OpenWireFormatFactory) wireFormatFactory);
+            } else {
+                owpv = new OpenWireProtocolVerifier(new OpenWireFormatFactory());
+            }
+            protocolVerifiers.put(AutoTransportUtils.OPENWIRE, owpv);
+        }
+    }
+
+    protected boolean isAllProtocols() {
+        return enabledProtocols == null || enabledProtocols.isEmpty();
+    }
+
+
+    protected final ExecutorService service;
+
+
+    /**
+     * This holds the initial buffer that has been read to detect the protocol.
+     */
+    public InitBuffer initBuffer;
+
+    @Override
+    protected void handleSocket(final Socket socket) {
+        final AutoTcpTransportServer server = this;
+
+        //This needs to be done in a new thread because
+        //the socket might be waiting on the client to send bytes
+        //doHandleSocket can't complete until the protocol can be detected
+        service.submit(new Runnable() {
+            @Override
+            public void run() {
+                server.doHandleSocket(socket);
+            }
+        });
+    }
+
+    @Override
+    protected TransportInfo configureTransport(final TcpTransportServer server, final Socket socket) throws Exception {
+         InputStream is = socket.getInputStream();
+
+         //We need to peak at the first 8 bytes of the buffer to detect the protocol
+         Buffer magic = new Buffer(8);
+         magic.readFrom(is);
+
+         ProtocolInfo protocolInfo = detectProtocol(magic.getData());
+
+         initBuffer = new InitBuffer(8, ByteBuffer.allocate(8));
+         initBuffer.buffer.put(magic.getData());
+
+         if (protocolInfo.detectedTransportFactory instanceof BrokerServiceAware) {
+             ((BrokerServiceAware) protocolInfo.detectedTransportFactory).setBrokerService(brokerService);
+         }
+
+        WireFormat format = protocolInfo.detectedWireFormatFactory.createWireFormat();
+        Transport transport = createTransport(socket, format, protocolInfo.detectedTransportFactory);
+
+        return new TransportInfo(format, transport, protocolInfo.detectedTransportFactory);
+    }
+
+    @Override
+    protected TcpTransport createTransport(Socket socket, WireFormat format) throws IOException {
+        return new TcpTransport(format, socket, this.initBuffer);
+    }
+
+    /**
+     * @param socket
+     * @param format
+     * @param detectedTransportFactory
+     * @return
+     */
+    protected TcpTransport createTransport(Socket socket, WireFormat format,
+            TcpTransportFactory detectedTransportFactory) throws IOException {
+        return createTransport(socket, format);
+    }
+
+    public void setWireFormatOptions(Map<String, Map<String, Object>> wireFormatOptions) {
+        this.wireFormatOptions = wireFormatOptions;
+    }
+
+    public void setEnabledProtocols(Set<String> enabledProtocols) {
+        this.enabledProtocols = enabledProtocols;
+    }
+
+    public void setAutoTransportOptions(Map<String, Object> autoTransportOptions) {
+        this.autoTransportOptions = autoTransportOptions;
+        if (autoTransportOptions.get("protocols") != null)
+            this.enabledProtocols = AutoTransportUtils.parseProtocols((String) autoTransportOptions.get("protocols"));
+    }
+
+    protected ProtocolInfo detectProtocol(byte[] buffer) throws IOException {
+        TcpTransportFactory detectedTransportFactory = transportFactory;
+        WireFormatFactory detectedWireFormatFactory = wireFormatFactory;
+
+        boolean found = false;
+        for (String scheme : protocolVerifiers.keySet()) {
+            if (protocolVerifiers.get(scheme).isProtocol(buffer)) {
+                LOG.debug("Detected " + scheme);
+                detectedWireFormatFactory = findWireFormatFactory(scheme, wireFormatOptions);
+
+                if (scheme.equals("default")) {
+                    scheme = "";
+                }
+
+                detectedTransportFactory = (TcpTransportFactory) findTransportFactory(scheme, transportOptions);
+                found = true;
+                break;
+            }
+        }
+
+        if (!found) {
+            throw new IllegalStateException("Could not detect wire format");
+        }
+
+        return new ProtocolInfo(detectedTransportFactory, detectedWireFormatFactory);
+
+    }
+
+    protected class ProtocolInfo {
+        public final TcpTransportFactory detectedTransportFactory;
+        public final WireFormatFactory detectedWireFormatFactory;
+
+        public ProtocolInfo(TcpTransportFactory detectedTransportFactory,
+                WireFormatFactory detectedWireFormatFactory) {
+            super();
+            this.detectedTransportFactory = detectedTransportFactory;
+            this.detectedWireFormatFactory = detectedWireFormatFactory;
+        }
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/activemq/blob/04ee70a1/activemq-broker/src/main/java/org/apache/activemq/broker/transport/auto/AutoTransportUtils.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/transport/auto/AutoTransportUtils.java b/activemq-broker/src/main/java/org/apache/activemq/broker/transport/auto/AutoTransportUtils.java
new file mode 100644
index 0000000..453292d
--- /dev/null
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/transport/auto/AutoTransportUtils.java
@@ -0,0 +1,62 @@
+/**
+ * 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.activemq.broker.transport.auto;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.activemq.util.IntrospectionSupport;
+
+/**
+ *
+ *
+ */
+public class AutoTransportUtils {
+
+    //wireformats
+    public static String ALL = "all";
+    public static String OPENWIRE = "default";
+    public static String STOMP = "stomp";
+    public static String AMQP = "amqp";
+    public static String MQTT = "mqtt";
+
+    //transports
+    public static String AUTO = "auto";
+
+    public static Map<String, Map<String, Object>> extractWireFormatOptions(Map<String, String> options ) {
+        Map<String, Map<String, Object>> wireFormatOptions = new HashMap<>();
+        if (options != null) {
+            wireFormatOptions.put(OPENWIRE, IntrospectionSupport.extractProperties(options, "wireFormat.default."));
+            wireFormatOptions.put(STOMP, IntrospectionSupport.extractProperties(options, "wireFormat.stomp."));
+            wireFormatOptions.put(AMQP, IntrospectionSupport.extractProperties(options, "wireFormat.amqp."));
+            wireFormatOptions.put(MQTT, IntrospectionSupport.extractProperties(options, "wireFormat.mqtt."));
+            wireFormatOptions.put(ALL, IntrospectionSupport.extractProperties(options, "wireFormat."));
+        }
+        return wireFormatOptions;
+    }
+
+    public static Set<String> parseProtocols(String protocolString) {
+        Set<String> protocolSet = new HashSet<>();;
+        if (protocolString != null && !protocolString.isEmpty()) {
+            protocolSet.addAll(Arrays.asList(protocolString.split(",")));
+        }
+        return protocolSet;
+    }
+}

http://git-wip-us.apache.org/repos/asf/activemq/blob/04ee70a1/activemq-broker/src/main/java/org/apache/activemq/broker/transport/auto/nio/AutoNIOSSLTransportServer.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/transport/auto/nio/AutoNIOSSLTransportServer.java b/activemq-broker/src/main/java/org/apache/activemq/broker/transport/auto/nio/AutoNIOSSLTransportServer.java
new file mode 100644
index 0000000..7ac3a97
--- /dev/null
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/transport/auto/nio/AutoNIOSSLTransportServer.java
@@ -0,0 +1,122 @@
+package org.apache.activemq.broker.transport.auto.nio;
+
+import java.io.IOException;
+import java.net.Socket;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.nio.ByteBuffer;
+import java.util.Set;
+
+import javax.net.ServerSocketFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLEngine;
+
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.BrokerServiceAware;
+import org.apache.activemq.broker.transport.auto.AutoTcpTransportServer;
+import org.apache.activemq.transport.Transport;
+import org.apache.activemq.transport.nio.AutoInitNioSSLTransport;
+import org.apache.activemq.transport.nio.NIOSSLTransport;
+import org.apache.activemq.transport.tcp.TcpTransport;
+import org.apache.activemq.transport.tcp.TcpTransport.InitBuffer;
+import org.apache.activemq.transport.tcp.TcpTransportFactory;
+import org.apache.activemq.transport.tcp.TcpTransportServer;
+import org.apache.activemq.wireformat.WireFormat;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class AutoNIOSSLTransportServer extends AutoTcpTransportServer {
+
+    private static final Logger LOG = LoggerFactory.getLogger(AutoNIOSSLTransportServer.class);
+
+    private SSLContext context;
+
+    public AutoNIOSSLTransportServer(SSLContext context, TcpTransportFactory transportFactory, URI location, ServerSocketFactory serverSocketFactory,
+            BrokerService brokerService, Set<String> enabledProtocols) throws IOException, URISyntaxException {
+        super(transportFactory, location, serverSocketFactory, brokerService, enabledProtocols);
+
+        this.context = context;
+    }
+
+    private boolean needClientAuth;
+    private boolean wantClientAuth;
+
+    protected Transport createTransport(Socket socket, WireFormat format, SSLEngine engine,
+            InitBuffer initBuffer, ByteBuffer inputBuffer, TcpTransportFactory detectedFactory) throws IOException {
+        NIOSSLTransport transport = new NIOSSLTransport(format, socket, engine, initBuffer, inputBuffer);
+        if (context != null) {
+            transport.setSslContext(context);
+        }
+
+        transport.setNeedClientAuth(needClientAuth);
+        transport.setWantClientAuth(wantClientAuth);
+
+
+        return transport;
+    }
+
+    @Override
+    protected TcpTransport createTransport(Socket socket, WireFormat format) throws IOException {
+        throw new UnsupportedOperationException("method not supported");
+    }
+
+    @Override
+    public boolean isSslServer() {
+        return true;
+    }
+
+    public boolean isNeedClientAuth() {
+        return this.needClientAuth;
+    }
+
+    public void setNeedClientAuth(boolean value) {
+        this.needClientAuth = value;
+    }
+
+    public boolean isWantClientAuth() {
+        return this.wantClientAuth;
+    }
+
+    public void setWantClientAuth(boolean value) {
+        this.wantClientAuth = value;
+    }
+
+
+    @Override
+    protected TransportInfo configureTransport(final TcpTransportServer server, final Socket socket) throws Exception {
+
+        //The SSLEngine needs to be initialized and handshake done to get the first command and detect the format
+        AutoInitNioSSLTransport in = new AutoInitNioSSLTransport(wireFormatFactory.createWireFormat(), socket);
+        if (context != null) {
+            in.setSslContext(context);
+        }
+        in.start();
+        SSLEngine engine = in.getSslSession();
+
+        //Wait for handshake to finish initializing
+        byte[] read = null;
+        do {
+            in.serviceRead();
+        } while((read = in.read) == null);
+
+        in.stop();
+
+        initBuffer = new InitBuffer(in.readSize, ByteBuffer.allocate(read.length));
+        initBuffer.buffer.put(read);
+
+        ProtocolInfo protocolInfo = detectProtocol(read);
+
+        if (protocolInfo.detectedTransportFactory instanceof BrokerServiceAware) {
+            ((BrokerServiceAware) protocolInfo.detectedTransportFactory).setBrokerService(brokerService);
+        }
+
+        WireFormat format = protocolInfo.detectedWireFormatFactory.createWireFormat();
+        Transport transport = createTransport(socket, format, engine, initBuffer, in.getInputBuffer(), protocolInfo.detectedTransportFactory);
+
+        return new TransportInfo(format, transport, protocolInfo.detectedTransportFactory);
+    }
+
+
+}
+
+

http://git-wip-us.apache.org/repos/asf/activemq/blob/04ee70a1/activemq-broker/src/main/java/org/apache/activemq/broker/transport/auto/nio/AutoNIOTransport.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/transport/auto/nio/AutoNIOTransport.java b/activemq-broker/src/main/java/org/apache/activemq/broker/transport/auto/nio/AutoNIOTransport.java
new file mode 100644
index 0000000..e1b6e71
--- /dev/null
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/transport/auto/nio/AutoNIOTransport.java
@@ -0,0 +1,85 @@
+/**
+ * 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.activemq.broker.transport.auto.nio;
+
+import java.io.IOException;
+import java.net.Socket;
+import java.net.URI;
+import java.net.UnknownHostException;
+
+import javax.net.SocketFactory;
+
+import org.apache.activemq.transport.nio.NIOTransport;
+import org.apache.activemq.wireformat.WireFormat;
+
+/**
+ *
+ *
+ */
+public class AutoNIOTransport extends NIOTransport {
+
+    public AutoNIOTransport(WireFormat format, Socket socket,
+            InitBuffer initBuffer) throws IOException {
+        super(format, socket, initBuffer);
+    }
+
+    public AutoNIOTransport(WireFormat wireFormat, Socket socket)
+            throws IOException {
+        super(wireFormat, socket);
+    }
+
+    public AutoNIOTransport(WireFormat wireFormat, SocketFactory socketFactory,
+            URI remoteLocation, URI localLocation) throws UnknownHostException,
+            IOException {
+        super(wireFormat, socketFactory, remoteLocation, localLocation);
+    }
+
+
+    boolean doneInitBuffer = false;
+
+    /**
+     * Read from the initial buffer if it is set
+     */
+    @Override
+    protected int readFromBuffer() throws IOException {
+        int readSize = 0;
+        if (!doneInitBuffer) {
+            if (initBuffer == null) {
+                throw new IOException("Null initBuffer");
+            }
+            if (nextFrameSize == -1) {
+                readSize = 4;
+                this.initBuffer.buffer.flip();
+                for (int i = 0; i < 4; i++) {
+                    currentBuffer.put(initBuffer.buffer.get());
+                }
+            } else {
+                for (int i = 0; i < 4; i++) {
+                    currentBuffer.put(initBuffer.buffer.get());
+                }
+                readSize = 4;
+                doneInitBuffer = true;
+            }
+
+        } else {
+            readSize += channel.read(currentBuffer);
+        }
+        return readSize;
+    }
+
+
+}

http://git-wip-us.apache.org/repos/asf/activemq/blob/04ee70a1/activemq-broker/src/main/java/org/apache/activemq/broker/transport/auto/nio/AutoNioSslTransportFactory.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/transport/auto/nio/AutoNioSslTransportFactory.java b/activemq-broker/src/main/java/org/apache/activemq/broker/transport/auto/nio/AutoNioSslTransportFactory.java
new file mode 100644
index 0000000..8142a2a
--- /dev/null
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/transport/auto/nio/AutoNioSslTransportFactory.java
@@ -0,0 +1,131 @@
+/**
+ * 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.activemq.broker.transport.auto.nio;
+
+import java.io.IOException;
+import java.net.Socket;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+import javax.net.ServerSocketFactory;
+import javax.net.ssl.SSLEngine;
+
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.BrokerServiceAware;
+import org.apache.activemq.broker.SslContext;
+import org.apache.activemq.broker.transport.auto.AutoTcpTransportServer;
+import org.apache.activemq.broker.transport.auto.AutoTransportUtils;
+import org.apache.activemq.openwire.OpenWireFormatFactory;
+import org.apache.activemq.transport.Transport;
+import org.apache.activemq.transport.TransportServer;
+import org.apache.activemq.transport.nio.NIOSSLTransport;
+import org.apache.activemq.transport.nio.NIOSSLTransportFactory;
+import org.apache.activemq.transport.tcp.TcpTransport.InitBuffer;
+import org.apache.activemq.transport.tcp.TcpTransportFactory;
+import org.apache.activemq.util.IOExceptionSupport;
+import org.apache.activemq.util.IntrospectionSupport;
+import org.apache.activemq.util.URISupport;
+import org.apache.activemq.wireformat.WireFormat;
+
+/**
+ *
+ *
+ */
+public class AutoNioSslTransportFactory extends NIOSSLTransportFactory implements BrokerServiceAware {
+    protected BrokerService brokerService;
+
+    /* (non-Javadoc)
+     * @see org.apache.activemq.broker.BrokerServiceAware#setBrokerService(org.apache.activemq.broker.BrokerService)
+     */
+    @Override
+    public void setBrokerService(BrokerService brokerService) {
+        this.brokerService = brokerService;
+    }
+
+    @Override
+    protected AutoNIOSSLTransportServer createTcpTransportServer(URI location, ServerSocketFactory serverSocketFactory) throws IOException, URISyntaxException {
+        return new AutoNIOSSLTransportServer(context, this, location, serverSocketFactory, brokerService, enabledProtocols) {
+
+            @Override
+            protected Transport createTransport(Socket socket, WireFormat format, SSLEngine engine, InitBuffer initBuffer,
+                     ByteBuffer inputBuffer, TcpTransportFactory detectedFactory) throws IOException {
+                NIOSSLTransport nioSslTransport = (NIOSSLTransport) detectedFactory.createTransport(
+                        format, socket, engine, initBuffer, inputBuffer);
+
+                if (format.getClass().toString().contains("MQTT")) {
+                    if (!allowLinkStealingSet) {
+                        this.setAllowLinkStealing(true);
+                    }
+                }
+
+                if (context != null) {
+                    nioSslTransport.setSslContext(context);
+                }
+
+                nioSslTransport.setNeedClientAuth(isNeedClientAuth());
+                nioSslTransport.setWantClientAuth(isWantClientAuth());
+
+                return nioSslTransport;
+            }
+
+        };
+
+    }
+
+    boolean allowLinkStealingSet = false;
+    private Set<String> enabledProtocols;
+
+    @Override
+    public TransportServer doBind(final URI location) throws IOException {
+        try {
+            if (SslContext.getCurrentSslContext() != null) {
+                try {
+                    context = SslContext.getCurrentSslContext().getSSLContext();
+                } catch (Exception e) {
+                    throw new IOException(e);
+                }
+            }
+
+            Map<String, String> options = new HashMap<String, String>(URISupport.parseParameters(location));
+
+            Map<String, Object> autoProperties = IntrospectionSupport.extractProperties(options, "auto.");
+            this.enabledProtocols = AutoTransportUtils.parseProtocols((String) autoProperties.get("protocols"));
+
+            ServerSocketFactory serverSocketFactory = createServerSocketFactory();
+            AutoTcpTransportServer server = createTcpTransportServer(location, serverSocketFactory);
+            server.setWireFormatFactory(new OpenWireFormatFactory());
+            if (options.get("allowLinkStealing") != null){
+                allowLinkStealingSet = true;
+            }
+            IntrospectionSupport.setProperties(server, options);
+            server.setAutoTransportOptions(IntrospectionSupport.extractProperties(options, "auto."));
+            server.setTransportOption(IntrospectionSupport.extractProperties(options, "transport."));
+            server.setWireFormatOptions(AutoTransportUtils.extractWireFormatOptions(options));
+            server.bind();
+
+            return server;
+        } catch (URISyntaxException e) {
+            throw IOExceptionSupport.create(e);
+        }
+    }
+
+
+}

http://git-wip-us.apache.org/repos/asf/activemq/blob/04ee70a1/activemq-broker/src/main/java/org/apache/activemq/broker/transport/auto/nio/AutoNioTransportFactory.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/transport/auto/nio/AutoNioTransportFactory.java b/activemq-broker/src/main/java/org/apache/activemq/broker/transport/auto/nio/AutoNioTransportFactory.java
new file mode 100644
index 0000000..0922ae6
--- /dev/null
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/transport/auto/nio/AutoNioTransportFactory.java
@@ -0,0 +1,114 @@
+/**
+ * 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.activemq.broker.transport.auto.nio;
+
+import java.io.IOException;
+import java.net.Socket;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+import javax.net.ServerSocketFactory;
+
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.BrokerServiceAware;
+import org.apache.activemq.broker.transport.auto.AutoTcpTransportServer;
+import org.apache.activemq.broker.transport.auto.AutoTransportUtils;
+import org.apache.activemq.openwire.OpenWireFormatFactory;
+import org.apache.activemq.transport.Transport;
+import org.apache.activemq.transport.TransportServer;
+import org.apache.activemq.transport.nio.NIOTransport;
+import org.apache.activemq.transport.nio.NIOTransportFactory;
+import org.apache.activemq.transport.tcp.TcpTransport;
+import org.apache.activemq.transport.tcp.TcpTransportFactory;
+import org.apache.activemq.util.IOExceptionSupport;
+import org.apache.activemq.util.IntrospectionSupport;
+import org.apache.activemq.util.URISupport;
+import org.apache.activemq.wireformat.WireFormat;
+
+/**
+ *
+ *
+ */
+public class AutoNioTransportFactory extends NIOTransportFactory implements BrokerServiceAware {
+    protected BrokerService brokerService;
+    /* (non-Javadoc)
+     * @see org.apache.activemq.broker.BrokerServiceAware#setBrokerService(org.apache.activemq.broker.BrokerService)
+     */
+    @Override
+    public void setBrokerService(BrokerService brokerService) {
+        this.brokerService = brokerService;
+    }
+
+    @Override
+    protected AutoTcpTransportServer createTcpTransportServer(URI location, ServerSocketFactory serverSocketFactory) throws IOException, URISyntaxException {
+        return new AutoTcpTransportServer(this, location, serverSocketFactory, brokerService, enabledProtocols) {
+            @Override
+            protected TcpTransport createTransport(Socket socket, WireFormat format, TcpTransportFactory detectedTransportFactory) throws IOException {
+                TcpTransport nioTransport = null;
+                if (detectedTransportFactory.getClass().equals(NIOTransportFactory.class)) {
+                    nioTransport = new AutoNIOTransport(format, socket,this.initBuffer);
+                } else {
+                    nioTransport = detectedTransportFactory.createTransport(
+                            format, socket, this.initBuffer);
+                }
+
+                if (format.getClass().toString().contains("MQTT")) {
+                    if (!allowLinkStealingSet) {
+                        this.setAllowLinkStealing(true);
+                    }
+                }
+
+                return nioTransport;
+            }
+        };
+
+    }
+
+    boolean allowLinkStealingSet = false;
+    private Set<String> enabledProtocols;
+
+    @Override
+    public TransportServer doBind(final URI location) throws IOException {
+        try {
+            Map<String, String> options = new HashMap<String, String>(URISupport.parseParameters(location));
+
+            Map<String, Object> autoProperties = IntrospectionSupport.extractProperties(options, "auto.");
+            this.enabledProtocols = AutoTransportUtils.parseProtocols((String) autoProperties.get("protocols"));
+
+            ServerSocketFactory serverSocketFactory = createServerSocketFactory();
+            AutoTcpTransportServer server = createTcpTransportServer(location, serverSocketFactory);
+            //server.setWireFormatFactory(createWireFormatFactory(options));
+            server.setWireFormatFactory(new OpenWireFormatFactory());
+            if (options.get("allowLinkStealing") != null){
+                allowLinkStealingSet = true;
+            }
+            IntrospectionSupport.setProperties(server, options);
+            server.setTransportOption(IntrospectionSupport.extractProperties(options, "transport."));
+            server.setWireFormatOptions(AutoTransportUtils.extractWireFormatOptions(options));
+            server.bind();
+
+            return server;
+        } catch (URISyntaxException e) {
+            throw IOExceptionSupport.create(e);
+        }
+    }
+
+
+}

http://git-wip-us.apache.org/repos/asf/activemq/blob/04ee70a1/activemq-broker/src/main/java/org/apache/activemq/broker/transport/protocol/AmqpProtocolVerifier.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/transport/protocol/AmqpProtocolVerifier.java b/activemq-broker/src/main/java/org/apache/activemq/broker/transport/protocol/AmqpProtocolVerifier.java
new file mode 100644
index 0000000..11f0574
--- /dev/null
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/transport/protocol/AmqpProtocolVerifier.java
@@ -0,0 +1,36 @@
+/**
+ * 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.activemq.broker.transport.protocol;
+
+
+/**
+ *
+ *
+ */
+public class AmqpProtocolVerifier implements ProtocolVerifier {
+
+    static final byte[] PREFIX = new byte[] { 'A', 'M', 'Q', 'P' };
+
+    @Override
+    public boolean isProtocol(byte[] value) {
+        for (int i = 0; i < PREFIX.length; i++) {
+            if (value[i] != PREFIX[i])
+                return false;
+        }
+        return true;
+    }
+}

http://git-wip-us.apache.org/repos/asf/activemq/blob/04ee70a1/activemq-broker/src/main/java/org/apache/activemq/broker/transport/protocol/MqttProtocolVerifier.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/transport/protocol/MqttProtocolVerifier.java b/activemq-broker/src/main/java/org/apache/activemq/broker/transport/protocol/MqttProtocolVerifier.java
new file mode 100644
index 0000000..4b4edd2
--- /dev/null
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/transport/protocol/MqttProtocolVerifier.java
@@ -0,0 +1,45 @@
+/**
+ * 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.activemq.broker.transport.protocol;
+
+/**
+ *
+ *
+ */
+public class MqttProtocolVerifier implements ProtocolVerifier {
+
+    /* (non-Javadoc)
+     * @see org.apache.activemq.broker.transport.protocol.ProtocolVerifier#isProtocol(byte[])
+     */
+    @Override
+    public boolean isProtocol(byte[] value) {
+        boolean mqtt311 = value[4] == 77 && // M
+                value[5] == 81 && // Q
+                value[6] == 84 && // T
+                value[7] == 84;   // T
+
+        boolean mqtt31  = value[4] == 77  && // M
+                        value[5] == 81  && // Q
+                        value[6] == 73  && // I
+                        value[7] == 115;   // s
+
+        return mqtt311 || mqtt31;
+    }
+
+
+
+}


Mime
View raw message