https://issues.apache.org/jira/browse/AMQ-5889

-Cleaned up missing license headers and refactored packages.
-Added configuration options for a protocol detection timeout and
for the max number of connections accepted at the same time.
-Fixed a regression with connection counts
-Also added some more tests

activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/auto/JMSClientAutoNioPlusSslTest.java

@@ -18,7 +18,6 @@ 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;

activemq-broker/src/main/java/org/apache/activemq/broker/transport/protocol/ProtocolVerifier.java

@@ -1,9 +0,0 @@
-package org.apache.activemq.broker.transport.protocol;
-
-
-public interface ProtocolVerifier {
-
-    public boolean isProtocol(byte[] value);
-
-
-}

activemq-broker/src/main/java/org/apache/activemq/transport/auto/AutoSslTransportFactory.java

@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.activemq.broker.transport.auto;
+package org.apache.activemq.transport.auto;
 
 import java.io.IOException;
 import java.net.Socket;

activemq-broker/src/main/java/org/apache/activemq/transport/auto/AutoSslTransportServer.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.activemq.broker.transport.auto;
+package org.apache.activemq.transport.auto;
 
 import java.io.IOException;
 import java.net.Socket;

activemq-broker/src/main/java/org/apache/activemq/transport/auto/AutoTcpTransportFactory.java

@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.activemq.broker.transport.auto;
+package org.apache.activemq.transport.auto;
 
 import java.io.IOException;
 import java.net.Socket;
@@ -23,27 +23,19 @@ 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;
 
 /**
  *

activemq-broker/src/main/java/org/apache/activemq/transport/auto/AutoTcpTransportServer.java

@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.activemq.broker.transport.auto;
+package org.apache.activemq.transport.auto;
 
 import java.io.IOException;
 import java.io.InputStream;
@@ -28,20 +28,27 @@ import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
 
 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.InactivityIOException;
 import org.apache.activemq.transport.Transport;
 import org.apache.activemq.transport.TransportFactory;
 import org.apache.activemq.transport.TransportServer;
+import org.apache.activemq.transport.protocol.AmqpProtocolVerifier;
+import org.apache.activemq.transport.protocol.MqttProtocolVerifier;
+import org.apache.activemq.transport.protocol.OpenWireProtocolVerifier;
+import org.apache.activemq.transport.protocol.ProtocolVerifier;
+import org.apache.activemq.transport.protocol.StompProtocolVerifier;
 import org.apache.activemq.transport.tcp.TcpTransport;
 import org.apache.activemq.transport.tcp.TcpTransport.InitBuffer;
 import org.apache.activemq.transport.tcp.TcpTransportFactory;
@@ -49,9 +56,9 @@ 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.util.ServiceStopper;
 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;
 
@@ -69,6 +76,9 @@ public class AutoTcpTransportServer extends TcpTransportServer {
 
     protected BrokerService brokerService;
 
+    protected int maxConnectionThreadPoolSize = Integer.MAX_VALUE;
+    protected int protocolDetectionTimeOut = 30000;
+
     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>();
 
@@ -104,8 +114,9 @@ public class AutoTcpTransportServer extends TcpTransportServer {
             // Try to load if from a META-INF property.
             try {
                 tf = (TransportFactory)TRANSPORT_FACTORY_FINDER.newInstance(scheme);
-                if (options != null)
+                if (options != null) {
                     IntrospectionSupport.setProperties(tf, options);
+                }
                 transportFactories.put(scheme, tf);
             } catch (Throwable e) {
                 throw IOExceptionSupport.create("Transport scheme NOT recognized: [" + scheme + "]", e);
@@ -136,12 +147,35 @@ public class AutoTcpTransportServer extends TcpTransportServer {
             Set<String> enabledProtocols)
             throws IOException, URISyntaxException {
         super(transportFactory, location, serverSocketFactory);
-        service = Executors.newCachedThreadPool();
+
+        //Use an executor service here to handle new connections.  Setting the max number
+        //of threads to the maximum number of connections the thread count isn't unbounded
+        service = new ThreadPoolExecutor(maxConnectionThreadPoolSize,
+                maxConnectionThreadPoolSize,
+                30L, TimeUnit.SECONDS,
+                new LinkedBlockingQueue<Runnable>());
+        //allow the thread pool to shrink if the max number of threads isn't needed
+        service.allowCoreThreadTimeOut(true);
+
         this.brokerService = brokerService;
         this.enabledProtocols = enabledProtocols;
         initProtocolVerifiers();
     }
 
+    public int getMaxConnectionThreadPoolSize() {
+        return maxConnectionThreadPoolSize;
+    }
+
+    public void setMaxConnectionThreadPoolSize(int maxConnectionThreadPoolSize) {
+        this.maxConnectionThreadPoolSize = maxConnectionThreadPoolSize;
+        service.setCorePoolSize(maxConnectionThreadPoolSize);
+        service.setMaximumPoolSize(maxConnectionThreadPoolSize);
+    }
+
+    public void setProtocolDetectionTimeOut(int protocolDetectionTimeOut) {
+        this.protocolDetectionTimeOut = protocolDetectionTimeOut;
+    }
+
     @Override
     public void setWireFormatFactory(WireFormatFactory factory) {
         super.setWireFormatFactory(factory);
@@ -179,7 +213,7 @@ public class AutoTcpTransportServer extends TcpTransportServer {
     }
 
 
-    protected final ExecutorService service;
+    protected final ThreadPoolExecutor service;
 
 
     /**
@@ -190,7 +224,6 @@ public class AutoTcpTransportServer extends TcpTransportServer {
     @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
@@ -204,23 +237,54 @@ public class AutoTcpTransportServer extends TcpTransportServer {
 
     @Override
     protected TransportInfo configureTransport(final TcpTransportServer server, final Socket socket) throws Exception {
-         InputStream is = socket.getInputStream();
+        final InputStream is = socket.getInputStream();
+        ExecutorService executor = Executors.newSingleThreadExecutor();
 
-         //We need to peak at the first 8 bytes of the buffer to detect the protocol
+        final AtomicInteger readBytes = new AtomicInteger(0);
-         Buffer magic = new Buffer(8);
+        final ByteBuffer data = ByteBuffer.allocate(8);
-         magic.readFrom(is);
+        // We need to peak at the first 8 bytes of the buffer to detect the protocol
+        Future<?> future = executor.submit(new Runnable() {
+            @Override
+            public void run() {
+                try {
+                    do {
+                        int read = is.read();
+                        if (read == -1) {
+                            throw new IOException("Connection faild, stream is closed.");
+                        }
+                        data.put((byte) read);
+                        readBytes.incrementAndGet();
+                    } while (readBytes.get() < 8);
+                } catch (Exception e) {
+                    throw new IllegalStateException(e);
+                }
+            }
+        });
 
-         ProtocolInfo protocolInfo = detectProtocol(magic.getData());
+        try {
+            //Wait for protocolDetectionTimeOut if defined
+            if (protocolDetectionTimeOut > 0) {
+                future.get(protocolDetectionTimeOut, TimeUnit.MILLISECONDS);
+            } else {
+                future.get();
+            }
+            data.flip();
+        } catch (TimeoutException e) {
+            throw new InactivityIOException("Client timed out before wire format could be detected. " +
+                    " 8 bytes are required to detect the protocol but only: " + readBytes + " were sent.");
+        }
 
-         initBuffer = new InitBuffer(8, ByteBuffer.allocate(8));
+        ProtocolInfo protocolInfo = detectProtocol(data.array());
-         initBuffer.buffer.put(magic.getData());
 
-         if (protocolInfo.detectedTransportFactory instanceof BrokerServiceAware) {
+        initBuffer = new InitBuffer(readBytes.get(), ByteBuffer.allocate(readBytes.get()));
-             ((BrokerServiceAware) protocolInfo.detectedTransportFactory).setBrokerService(brokerService);
+        initBuffer.buffer.put(data.array());
-         }
+
+        if (protocolInfo.detectedTransportFactory instanceof BrokerServiceAware) {
+            ((BrokerServiceAware) protocolInfo.detectedTransportFactory).setBrokerService(brokerService);
+        }
 
         WireFormat format = protocolInfo.detectedWireFormatFactory.createWireFormat();
-        Transport transport = createTransport(socket, format, protocolInfo.detectedTransportFactory);
+        Transport transport = createTransport(socket, format,protocolInfo.detectedTransportFactory);
 
         return new TransportInfo(format, transport, protocolInfo.detectedTransportFactory);
     }
@@ -251,8 +315,16 @@ public class AutoTcpTransportServer extends TcpTransportServer {
 
     public void setAutoTransportOptions(Map<String, Object> autoTransportOptions) {
         this.autoTransportOptions = autoTransportOptions;
-        if (autoTransportOptions.get("protocols") != null)
+        if (autoTransportOptions.get("protocols") != null) {
             this.enabledProtocols = AutoTransportUtils.parseProtocols((String) autoTransportOptions.get("protocols"));
+        }
+    }
+    @Override
+    protected void doStop(ServiceStopper stopper) throws Exception {
+        if (service != null) {
+            service.shutdown();
+        }
+        super.doStop(stopper);
     }
 
     protected ProtocolInfo detectProtocol(byte[] buffer) throws IOException {

activemq-broker/src/main/java/org/apache/activemq/transport/auto/AutoTransportUtils.java

@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.activemq.broker.transport.auto;
+package org.apache.activemq.transport.auto;
 
 import java.util.Arrays;
 import java.util.HashMap;

activemq-broker/src/main/java/org/apache/activemq/transport/auto/nio/AutoNIOSSLTransportServer.java

@@ -1,4 +1,4 @@
-package org.apache.activemq.broker.transport.auto.nio;
+package org.apache.activemq.transport.auto.nio;
 
 import java.io.IOException;
 import java.net.Socket;
@@ -6,6 +6,12 @@ import java.net.URI;
 import java.net.URISyntaxException;
 import java.nio.ByteBuffer;
 import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 import javax.net.ServerSocketFactory;
 import javax.net.ssl.SSLContext;
@@ -13,8 +19,9 @@ 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.InactivityIOException;
 import org.apache.activemq.transport.Transport;
+import org.apache.activemq.transport.auto.AutoTcpTransportServer;
 import org.apache.activemq.transport.nio.AutoInitNioSSLTransport;
 import org.apache.activemq.transport.nio.NIOSSLTransport;
 import org.apache.activemq.transport.tcp.TcpTransport;
@@ -25,6 +32,22 @@ import org.apache.activemq.wireformat.WireFormat;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+/**
+ * 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.
+ */
 public class AutoNIOSSLTransportServer extends AutoTcpTransportServer {
 
     private static final Logger LOG = LoggerFactory.getLogger(AutoNIOSSLTransportServer.class);
@@ -84,27 +107,41 @@ public class AutoNIOSSLTransportServer extends AutoTcpTransportServer {
 
     @Override
     protected TransportInfo configureTransport(final TcpTransportServer server, final Socket socket) throws Exception {
+        ExecutorService executor = Executors.newSingleThreadExecutor();
 
         //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);
+        final 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
+        Future<Integer> future = executor.submit(new Callable<Integer>() {
-        byte[] read = null;
+            @Override
-        do {
+            public Integer call() throws Exception {
-            in.serviceRead();
+                //Wait for handshake to finish initializing
-        } while((read = in.read) == null);
+                do {
+                    in.serviceRead();
+                } while(in.readSize < 8);
+
+                return in.readSize;
+            }
+        });
+
+        try {
+            future.get(protocolDetectionTimeOut, TimeUnit.MILLISECONDS);
+        } catch (TimeoutException e) {
+            throw new InactivityIOException("Client timed out before wire format could be detected. " +
+                    " 8 bytes are required to detect the protocol but only: " + in.readSize + " were sent.");
+        }
 
         in.stop();
 
-        initBuffer = new InitBuffer(in.readSize, ByteBuffer.allocate(read.length));
+        initBuffer = new InitBuffer(in.readSize, ByteBuffer.allocate(in.read.length));
-        initBuffer.buffer.put(read);
+        initBuffer.buffer.put(in.read);
 
-        ProtocolInfo protocolInfo = detectProtocol(read);
+        ProtocolInfo protocolInfo = detectProtocol(in.read);
 
         if (protocolInfo.detectedTransportFactory instanceof BrokerServiceAware) {
             ((BrokerServiceAware) protocolInfo.detectedTransportFactory).setBrokerService(brokerService);

activemq-broker/src/main/java/org/apache/activemq/transport/auto/nio/AutoNIOTransport.java

@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.activemq.broker.transport.auto.nio;
+package org.apache.activemq.transport.auto.nio;
 
 import java.io.IOException;
 import java.net.Socket;
@@ -58,8 +58,8 @@ public class AutoNIOTransport extends NIOTransport {
     protected int readFromBuffer() throws IOException {
         int readSize = 0;
         if (!doneInitBuffer) {
-            if (initBuffer == null) {
+            if (initBuffer == null || initBuffer.readSize < 8) {
-                throw new IOException("Null initBuffer");
+                throw new IOException("Protocol type could not be determined.");
             }
             if (nextFrameSize == -1) {
                 readSize = 4;

activemq-broker/src/main/java/org/apache/activemq/transport/auto/nio/AutoNioSslTransportFactory.java

@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.activemq.broker.transport.auto.nio;
+package org.apache.activemq.transport.auto.nio;
 
 import java.io.IOException;
 import java.net.Socket;
@@ -31,11 +31,11 @@ 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.auto.AutoTcpTransportServer;
+import org.apache.activemq.transport.auto.AutoTransportUtils;
 import org.apache.activemq.transport.nio.NIOSSLTransport;
 import org.apache.activemq.transport.nio.NIOSSLTransportFactory;
 import org.apache.activemq.transport.tcp.TcpTransport.InitBuffer;

activemq-broker/src/main/java/org/apache/activemq/transport/auto/nio/AutoNioTransportFactory.java

@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.activemq.broker.transport.auto.nio;
+package org.apache.activemq.transport.auto.nio;
 
 import java.io.IOException;
 import java.net.Socket;
@@ -28,11 +28,11 @@ 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.auto.AutoTcpTransportServer;
+import org.apache.activemq.transport.auto.AutoTransportUtils;
 import org.apache.activemq.transport.nio.NIOTransport;
 import org.apache.activemq.transport.nio.NIOTransportFactory;
 import org.apache.activemq.transport.tcp.TcpTransport;

activemq-broker/src/main/java/org/apache/activemq/transport/nio/AutoInitNioSSLTransport.java

@@ -172,17 +172,11 @@ public class AutoInitNioSSLTransport extends NIOSSLTransport {
 
             while (true) {
                 if (!plain.hasRemaining()) {
-
                     int readCount = secureRead(plain);
 
-                    if (readCount == 0) {
-                        break;
-                    }
-
                     // channel is closed, cleanup
                     if (readCount == -1) {
                         onException(new EOFException());
-                        selection.close();
                         break;
                     }
 
@@ -191,8 +185,11 @@ public class AutoInitNioSSLTransport extends NIOSSLTransport {
 
                 if (status == SSLEngineResult.Status.OK && handshakeStatus != SSLEngineResult.HandshakeStatus.NEED_UNWRAP) {
                     processCommand(plain);
-                    //Break when command is found
+                    //we have received enough bytes to detect the protocol
-                    break;
+                    if (receiveCounter >= 8) {
+                        readSize = receiveCounter;
+                        break;
+                    }
                 }
             }
         } catch (IOException e) {
@@ -204,8 +201,13 @@ public class AutoInitNioSSLTransport extends NIOSSLTransport {
 
     @Override
     protected void processCommand(ByteBuffer plain) throws Exception {
-        read = plain.array();
+        ByteBuffer newBuffer = ByteBuffer.allocate(receiveCounter);
-        readSize = receiveCounter;
+        if (read != null) {
+            newBuffer.put(read);
+        }
+        newBuffer.put(plain);
+        newBuffer.flip();
+        read = newBuffer.array();
     }
 
 
@@ -214,7 +216,6 @@ public class AutoInitNioSSLTransport extends NIOSSLTransport {
         taskRunnerFactory = new TaskRunnerFactory("ActiveMQ NIOSSLTransport Task");
         // no need to init as we can delay that until demand (eg in doHandshake)
         connect();
-        //super.doStart();
     }
 
 
@@ -224,10 +225,6 @@ public class AutoInitNioSSLTransport extends NIOSSLTransport {
             taskRunnerFactory.shutdownNow();
             taskRunnerFactory = null;
         }
-//        if (selection != null) {
-//            selection.close();
-//            selection = null;
-//        }
     }
 
 

activemq-broker/src/main/java/org/apache/activemq/transport/protocol/AmqpProtocolVerifier.java

@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.activemq.broker.transport.protocol;
+package org.apache.activemq.transport.protocol;
 
 
 /**
@@ -28,8 +28,9 @@ public class AmqpProtocolVerifier implements ProtocolVerifier {
     @Override
     public boolean isProtocol(byte[] value) {
         for (int i = 0; i < PREFIX.length; i++) {
-            if (value[i] != PREFIX[i])
+            if (value[i] != PREFIX[i]) {
                 return false;
+            }
         }
         return true;
     }

activemq-broker/src/main/java/org/apache/activemq/transport/protocol/MqttProtocolVerifier.java

@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.activemq.broker.transport.protocol;
+package org.apache.activemq.transport.protocol;
 
 /**
  *

activemq-broker/src/main/java/org/apache/activemq/transport/protocol/OpenWireProtocolVerifier.java

@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.activemq.broker.transport.protocol;
+package org.apache.activemq.transport.protocol;
 
 import org.apache.activemq.command.WireFormatInfo;
 import org.apache.activemq.openwire.OpenWireFormat;

activemq-broker/src/main/java/org/apache/activemq/transport/protocol/ProtocolVerifier.java

@@ -0,0 +1,24 @@
+/**
+ * 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.protocol;
+
+
+public interface ProtocolVerifier {
+
+    public boolean isProtocol(byte[] value);
+
+}

activemq-broker/src/main/java/org/apache/activemq/transport/protocol/StompProtocolVerifier.java

@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.activemq.broker.transport.protocol;
+package org.apache.activemq.transport.protocol;
 
 import java.nio.charset.StandardCharsets;
 

activemq-broker/src/main/resources/META-INF/services/org/apache/activemq/transport/auto

@@ -14,4 +14,4 @@
 ## See the License for the specific language governing permissions and
 ## limitations under the License.
 ## ---------------------------------------------------------------------------
-class=org.apache.activemq.broker.transport.auto.AutoTcpTransportFactory
+class=org.apache.activemq.transport.auto.AutoTcpTransportFactory

activemq-broker/src/main/resources/META-INF/services/org/apache/activemq/transport/auto+nio

@@ -14,4 +14,4 @@
 ## See the License for the specific language governing permissions and
 ## limitations under the License.
 ## ---------------------------------------------------------------------------
-class=org.apache.activemq.broker.transport.auto.nio.AutoNioTransportFactory
+class=org.apache.activemq.transport.auto.nio.AutoNioTransportFactory

activemq-broker/src/main/resources/META-INF/services/org/apache/activemq/transport/auto+nio+ssl

@@ -14,4 +14,4 @@
 ## See the License for the specific language governing permissions and
 ## limitations under the License.
 ## ---------------------------------------------------------------------------
-class=org.apache.activemq.broker.transport.auto.nio.AutoNioSslTransportFactory
+class=org.apache.activemq.transport.auto.nio.AutoNioSslTransportFactory

activemq-broker/src/main/resources/META-INF/services/org/apache/activemq/transport/auto+ssl

@@ -14,4 +14,4 @@
 ## See the License for the specific language governing permissions and
 ## limitations under the License.
 ## ---------------------------------------------------------------------------
-class=org.apache.activemq.broker.transport.auto.AutoSslTransportFactory
+class=org.apache.activemq.transport.auto.AutoSslTransportFactory

activemq-client/src/main/java/org/apache/activemq/transport/nio/NIOSSLTransport.java

@@ -73,8 +73,9 @@ public class NIOSSLTransport extends NIOTransport {
             ByteBuffer inputBuffer) throws IOException {
         super(wireFormat, socket, initBuffer);
         this.sslEngine = engine;
-        if (engine != null)
+        if (engine != null) {
             this.sslSession = engine.getSession();
+        }
         this.inputBuffer = inputBuffer;
     }
 
@@ -146,11 +147,13 @@ public class NIOSSLTransport extends NIOTransport {
             this.buffOut = outputStream;
 
             //If the sslEngine was not passed in, then handshake
-            if (!hasSslEngine)
+            if (!hasSslEngine) {
                 sslEngine.beginHandshake();
+            }
             handshakeStatus = sslEngine.getHandshakeStatus();
-            if (!hasSslEngine)
+            if (!hasSslEngine) {
                 doHandshake();
+            }
 
            // if (hasSslEngine) {
             selection = SelectorManager.getInstance().register(channel, new SelectorManager.Listener() {
@@ -328,27 +331,28 @@ public class NIOSSLTransport extends NIOTransport {
             currentBuffer.putInt(nextFrameSize);
 
         } else {
-
             // If its all in one read then we can just take it all, otherwise take only
             // the current frame size and the next iteration starts a new command.
-            if (currentBuffer.remaining() >= plain.remaining()) {
+            if (currentBuffer != null) {
-                currentBuffer.put(plain);
+                if (currentBuffer.remaining() >= plain.remaining()) {
-            } else {
+                    currentBuffer.put(plain);
-                byte[] fill = new byte[currentBuffer.remaining()];
+                } else {
-                plain.get(fill);
+                    byte[] fill = new byte[currentBuffer.remaining()];
-                currentBuffer.put(fill);
+                    plain.get(fill);
-            }
+                    currentBuffer.put(fill);
+                }
 
-            // Either we have enough data for a new command or we have to wait for some more.
+                // Either we have enough data for a new command or we have to wait for some more.
-            if (currentBuffer.hasRemaining()) {
+                if (currentBuffer.hasRemaining()) {
-                return;
+                    return;
-            } else {
+                } else {
-                currentBuffer.flip();
+                    currentBuffer.flip();
-                Object command = wireFormat.unmarshal(new DataInputStream(new NIOInputStream(currentBuffer)));
+                    Object command = wireFormat.unmarshal(new DataInputStream(new NIOInputStream(currentBuffer)));
-                doConsume(command);
+                    doConsume(command);
-                nextFrameSize = -1;
+                    nextFrameSize = -1;
-                currentBuffer = null;
+                    currentBuffer = null;
-           }
+               }
+            }
         }
     }
 

activemq-client/src/main/java/org/apache/activemq/transport/tcp/TcpTransportServer.java

@@ -481,47 +481,60 @@ public class TcpTransportServer extends TransportServerThreadSupport implements
 
     final protected void doHandleSocket(Socket socket) {
         boolean closeSocket = true;
+        boolean countIncremented = false;
         try {
-            if (this.currentTransportCount.get() >= this.maximumConnections) {
+            int currentCount;
-                throw new ExceededMaximumConnectionsException(
+            do {
-                    "Exceeded the maximum number of allowed client connections. See the '" +
+                currentCount = currentTransportCount.get();
-                    "maximumConnections' property on the TCP transport configuration URI " +
+                if (currentCount >= this.maximumConnections) {
-                    "in the ActiveMQ configuration file (e.g., activemq.xml)");
+                     throw new ExceededMaximumConnectionsException(
-            } else {
+                         "Exceeded the maximum number of allowed client connections. See the '" +
-                currentTransportCount.incrementAndGet();
+                         "maximumConnections' property on the TCP transport configuration URI " +
+                         "in the ActiveMQ configuration file (e.g., activemq.xml)");
+                 }
 
-                HashMap<String, Object> options = new HashMap<String, Object>();
+            //Increment this value before configuring the transport
-                options.put("maxInactivityDuration", Long.valueOf(maxInactivityDuration));
+            //This is necessary because some of the transport servers must read from the
-                options.put("maxInactivityDurationInitalDelay", Long.valueOf(maxInactivityDurationInitalDelay));
+            //socket during configureTransport() so we want to make sure this value is
-                options.put("minmumWireFormatVersion", Integer.valueOf(minmumWireFormatVersion));
+            //accurate as the transport server could pause here waiting for data to be sent from a client
-                options.put("trace", Boolean.valueOf(trace));
+            } while(!currentTransportCount.compareAndSet(currentCount, currentCount + 1));
-                options.put("soTimeout", Integer.valueOf(soTimeout));
+            countIncremented = true;
-                options.put("socketBufferSize", Integer.valueOf(socketBufferSize));
-                options.put("connectionTimeout", Integer.valueOf(connectionTimeout));
-                options.put("logWriterName", logWriterName);
-                options.put("dynamicManagement", Boolean.valueOf(dynamicManagement));
-                options.put("startLogging", Boolean.valueOf(startLogging));
-                options.putAll(transportOptions);
 
-                TransportInfo transportInfo = configureTransport(this, socket);
+            HashMap<String, Object> options = new HashMap<String, Object>();
-                closeSocket = false;
+            options.put("maxInactivityDuration", Long.valueOf(maxInactivityDuration));
+            options.put("maxInactivityDurationInitalDelay", Long.valueOf(maxInactivityDurationInitalDelay));
+            options.put("minmumWireFormatVersion", Integer.valueOf(minmumWireFormatVersion));
+            options.put("trace", Boolean.valueOf(trace));
+            options.put("soTimeout", Integer.valueOf(soTimeout));
+            options.put("socketBufferSize", Integer.valueOf(socketBufferSize));
+            options.put("connectionTimeout", Integer.valueOf(connectionTimeout));
+            options.put("logWriterName", logWriterName);
+            options.put("dynamicManagement", Boolean.valueOf(dynamicManagement));
+            options.put("startLogging", Boolean.valueOf(startLogging));
+            options.putAll(transportOptions);
 
-                if (transportInfo.transport instanceof ServiceSupport) {
+            TransportInfo transportInfo = configureTransport(this, socket);
-                    ((ServiceSupport) transportInfo.transport).addServiceListener(this);
+            closeSocket = false;
-                }
 
-                Transport configuredTransport = transportInfo.transportFactory.serverConfigure(
+            if (transportInfo.transport instanceof ServiceSupport) {
-                        transportInfo.transport, transportInfo.format, options);
+                ((ServiceSupport) transportInfo.transport).addServiceListener(this);
-
-                getAcceptListener().onAccept(configuredTransport);
             }
+
+            Transport configuredTransport = transportInfo.transportFactory.serverConfigure(
+                    transportInfo.transport, transportInfo.format, options);
+
+            getAcceptListener().onAccept(configuredTransport);
+
         } catch (SocketTimeoutException ste) {
             // expect this to happen
-            currentTransportCount.decrementAndGet();
         } catch (Exception e) {
-            currentTransportCount.decrementAndGet();
             if (closeSocket) {
                 try {
+                    //if closing the socket, only decrement the count it was actually incremented
+                    //where it was incremented
+                    if (countIncremented) {
+                        currentTransportCount.decrementAndGet();
+                    }
                     socket.close();
                 } catch (Exception ignore) {
                 }

activemq-stomp/src/test/java/org/apache/activemq/transport/stomp/auto/AutoStompConnectTimeoutTest.java

@@ -0,0 +1,180 @@
+/**
+ * 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.stomp.auto;
+
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.net.Socket;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.concurrent.TimeUnit;
+
+import javax.net.ssl.SSLSocketFactory;
+
+import org.apache.activemq.transport.auto.AutoTcpTransportServer;
+import org.apache.activemq.transport.stomp.StompTestSupport;
+import org.apache.activemq.util.Wait;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Test that connection attempts that don't send the connect get cleaned by
+ * by the protocolDetectionTimeOut property
+ */
+@RunWith(Parameterized.class)
+public class AutoStompConnectTimeoutTest extends StompTestSupport {
+
+    private static final Logger LOG = LoggerFactory.getLogger(AutoStompConnectTimeoutTest.class);
+
+    private Socket connection;
+    protected String connectorScheme;
+
+    @Parameters(name="{0}")
+    public static Collection<Object[]> data() {
+        return Arrays.asList(new Object[][] {
+                {"auto"},
+                {"auto+ssl"},
+                {"auto+nio"},
+                {"auto+nio+ssl"}
+            });
+    }
+
+    public AutoStompConnectTimeoutTest(String connectorScheme) {
+        this.connectorScheme = connectorScheme;
+    }
+
+    protected String getConnectorScheme() {
+        return connectorScheme;
+    }
+
+    @Override
+    public void tearDown() throws Exception {
+        if (connection != null) {
+            try {
+                connection.close();
+            } catch (Throwable e) {}
+            connection = null;
+        }
+        super.tearDown();
+    }
+
+    @Override
+    public String getAdditionalConfig() {
+        return "?protocolDetectionTimeOut=1500";
+    }
+
+    @Test(timeout = 15000)
+    public void testInactivityMonitor() throws Exception {
+
+        Thread t1 = new Thread() {
+
+            @Override
+            public void run() {
+                try {
+                    connection = createSocket();
+                    connection.getOutputStream().write('C');
+                    connection.getOutputStream().flush();
+                } catch (Exception ex) {
+                    LOG.error("unexpected exception on connect/disconnect", ex);
+                    exceptions.add(ex);
+                }
+            }
+        };
+
+        t1.start();
+
+        assertTrue("one connection", Wait.waitFor(new Wait.Condition() {
+             @Override
+             public boolean isSatisified() throws Exception {
+                 AutoTcpTransportServer server = (AutoTcpTransportServer) brokerService.getTransportConnectorByScheme(getConnectorScheme()).getServer();
+                 return 1 == server.getCurrentTransportCount().get();
+             }
+        }, TimeUnit.SECONDS.toMillis(15), TimeUnit.MILLISECONDS.toMillis(250)));
+
+        // and it should be closed due to inactivity
+        assertTrue("no dangling connections", Wait.waitFor(new Wait.Condition() {
+            @Override
+            public boolean isSatisified() throws Exception {
+                AutoTcpTransportServer server = (AutoTcpTransportServer) brokerService.getTransportConnectorByScheme(getConnectorScheme()).getServer();
+                return 0 == server.getCurrentTransportCount().get();
+            }
+        }, TimeUnit.SECONDS.toMillis(15), TimeUnit.MILLISECONDS.toMillis(500)));
+
+        assertTrue("no exceptions", exceptions.isEmpty());
+    }
+
+    @Override
+    protected boolean isUseTcpConnector() {
+        return false;
+    }
+    @Override
+    protected boolean isUseAutoConnector() {
+        return connectorScheme.equalsIgnoreCase("auto");
+    }
+
+    @Override
+    protected boolean isUseAutoSslConnector() {
+        return connectorScheme.equalsIgnoreCase("auto+ssl");
+    }
+
+    @Override
+    protected boolean isUseAutoNioConnector() {
+        return connectorScheme.equalsIgnoreCase("auto+nio");
+    }
+
+    @Override
+    protected boolean isUseAutoNioPlusSslConnector() {
+        return connectorScheme.equalsIgnoreCase("auto+nio+ssl");
+    }
+
+    @Override
+    protected Socket createSocket() throws IOException {
+
+        boolean useSSL = false;
+        int port = 0;
+
+        switch (connectorScheme) {
+            case "auto":
+                port = this.autoPort;
+                break;
+            case "auto+ssl":
+                useSSL = true;
+                port = this.autoSslPort;
+                break;
+            case "auto+nio":
+                port = this.autoNioPort;
+                break;
+            case "auto+nio+ssl":
+                useSSL = true;
+                port = this.autoNioSslPort;
+                break;
+            default:
+                throw new IOException("Invalid STOMP connector scheme passed to test.");
+        }
+
+        if (useSSL) {
+            return SSLSocketFactory.getDefault().createSocket("localhost", port);
+        } else {
+            return new Socket("localhost", port);
+        }
+    }
+}

activemq-unit-tests/src/test/java/org/apache/activemq/transport/auto/AutoNIOSslTransportBrokerTest.java

@@ -49,7 +49,6 @@ public class AutoNIOSslTransportBrokerTest extends TransportBrokerTestSupport {
         System.setProperty("javax.net.ssl.keyStore", SERVER_KEYSTORE);
         System.setProperty("javax.net.ssl.keyStorePassword", PASSWORD);
         System.setProperty("javax.net.ssl.keyStoreType", KEYSTORE_TYPE);
-        //System.setProperty("javax.net.debug", "ssl,handshake,data,trustmanager");
 
         maxWait = 10000;
         super.setUp();

activemq-unit-tests/src/test/java/org/apache/activemq/transport/auto/AutoTransportMaxConnectionsTest.java

@@ -0,0 +1,148 @@
+/**
+ * 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.auto;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.TransportConnector;
+import org.apache.activemq.transport.tcp.TcpTransportServer;
+import org.apache.activemq.util.Wait;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
+import org.springframework.jms.support.JmsUtils;
+
+@RunWith(Parameterized.class)
+public class AutoTransportMaxConnectionsTest {
+
+    public static final String KEYSTORE_TYPE = "jks";
+    public static final String PASSWORD = "password";
+    public static final String SERVER_KEYSTORE = "src/test/resources/server.keystore";
+    public static final String TRUST_KEYSTORE = "src/test/resources/client.keystore";
+    private static final int maxConnections = 20;
+
+    private final ExecutorService executor = Executors.newCachedThreadPool();
+    private String connectionUri;
+    private BrokerService service;
+    private TransportConnector connector;
+    private final String transportType;
+
+    @Parameters
+    public static Collection<Object[]> data() {
+        return Arrays.asList(new Object[][] {
+                {"auto"},
+                {"auto+nio"},
+                {"auto+ssl"},
+                {"auto+nio+ssl"},
+            });
+    }
+
+
+    public AutoTransportMaxConnectionsTest(String transportType) {
+        super();
+        this.transportType = transportType;
+    }
+
+
+    @Before
+    public void setUp() throws Exception {
+        System.setProperty("javax.net.ssl.trustStore", TRUST_KEYSTORE);
+        System.setProperty("javax.net.ssl.trustStorePassword", PASSWORD);
+        System.setProperty("javax.net.ssl.trustStoreType", KEYSTORE_TYPE);
+        System.setProperty("javax.net.ssl.keyStore", SERVER_KEYSTORE);
+        System.setProperty("javax.net.ssl.keyStorePassword", PASSWORD);
+        System.setProperty("javax.net.ssl.keyStoreType", KEYSTORE_TYPE);
+
+        service = new BrokerService();
+        service.setPersistent(false);
+        service.setUseJmx(false);
+        connector = service.addConnector(transportType + "://0.0.0.0:0?maxConnectionThreadPoolSize=10&maximumConnections="+maxConnections);
+        connectionUri = connector.getPublishableConnectString();
+        service.start();
+        service.waitUntilStarted();
+    }
+
+    protected ConnectionFactory createConnectionFactory() throws Exception {
+        return new ActiveMQConnectionFactory(connectionUri);
+    }
+
+    @Test
+    public void testMaxConnectionControl() throws Exception {
+        final ConnectionFactory cf = createConnectionFactory();
+        final CountDownLatch startupLatch = new CountDownLatch(1);
+
+        for(int i = 0; i < maxConnections + 20; i++) {
+            executor.submit(new Runnable() {
+                @Override
+                public void run() {
+                    Connection conn = null;
+                    try {
+                        startupLatch.await();
+                        conn = cf.createConnection();
+                        conn.start();
+                    } catch (Exception e) {
+                        //JmsUtils.closeConnection(conn);
+                    }
+                }
+            });
+        }
+
+        TcpTransportServer transportServer = (TcpTransportServer)connector.getServer();
+        // ensure the max connections is in effect
+        assertEquals(maxConnections, transportServer.getMaximumConnections());
+        // No connections at first
+        assertEquals(0, connector.getConnections().size());
+        // Release the latch to set up connections in parallel
+        startupLatch.countDown();
+
+        final TransportConnector connector = this.connector;
+
+        // Expect the max connections is created
+        assertTrue("Expected: " + maxConnections + " found: " + connector.getConnections().size(),
+            Wait.waitFor(new Wait.Condition() {
+                @Override
+                public boolean isSatisified() throws Exception {
+                    return connector.getConnections().size() == maxConnections;
+                }
+            })
+        );
+
+    }
+
+    @After
+    public void tearDown() throws Exception {
+        executor.shutdown();
+
+        service.stop();
+        service.waitUntilStopped();
+    }
+}