Merge branch 'apache:main' into runAsNonRoot

Author: alxgomz

activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpNioSslTransport.java

@@ -54,6 +54,22 @@ public AmqpNioSslTransport(WireFormat wireFormat, Socket socket,
     @Override
     protected void initializeStreams() throws IOException {
         super.initializeStreams();
+        if (initBuffer != null) {
+            initBuffer.buffer.flip();
+            // If we are processing the initial buffer from the auto transport,
+            // then process first. This generally should only have 8 bytes from
+            // the initial read
+            if (initBuffer.buffer.hasRemaining()) {
+                receiveCounter.addAndGet(initBuffer.readSize);
+                try {
+                    // one call is all that is needed to consume all data
+                    processCommand(initBuffer.buffer);
+                } catch (Exception e) {
+                    throw new IOException(e);
+                }
+                initBuffer.buffer.clear();
+            }
+        }
         if (inputBuffer.position() != 0 && inputBuffer.hasRemaining()) {
             serviceRead();
         }
@@ -76,23 +92,4 @@ protected void doInit() throws Exception {
         super.doInit();
     }
 
-    @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);
-    }
-
-
-}
+}

activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpNioTransport.java

@@ -140,7 +140,7 @@ private void serviceRead() {
     }
 
     protected void processBuffer(ByteBuffer buffer, int readSize) throws Exception {
-        receiveCounter += readSize;
+        receiveCounter.addAndGet(readSize);
 
         buffer.flip();
         frameReader.parse(buffer);
@@ -164,4 +164,4 @@ protected void doStop(ServiceStopper stopper) throws Exception {
             super.doStop(stopper);
         }
     }
-}
+}

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

@@ -290,7 +290,7 @@ public void run() {
 
         try {
             //If this fails and throws an exception and the socket will be closed
-            waitForProtocolDetectionFinish(future, readBytes);
+            waitForProtocolDetectionFinish(future, readBytes.get());
         } finally {
             //call cancel in case task didn't complete
             future.cancel(true);
@@ -311,7 +311,7 @@ public void run() {
         return new TransportInfo(format, transport, protocolInfo.detectedTransportFactory);
     }
 
-    protected void waitForProtocolDetectionFinish(final Future<?> future, final AtomicInteger readBytes) throws Exception {
+    protected void waitForProtocolDetectionFinish(final Future<?> future, final int readBytes) throws Exception {
         try {
             //Wait for protocolDetectionTimeOut if defined
             if (protocolDetectionTimeOut > 0) {
@@ -321,7 +321,7 @@ protected void waitForProtocolDetectionFinish(final Future<?> future, final Atom
             }
         } 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.get() + " byte(s) were sent.");
+                    " 8 bytes are required to detect the protocol but only: " + readBytes + " byte(s) were sent.");
         }
     }
 

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

@@ -145,20 +145,20 @@ public void run() {
                     //to be told when bytes are ready
                     in.serviceRead();
                     attempts++;
-                } while(in.getReadSize().get() < 8 && !Thread.interrupted());
+                } while(in.getReceiveCounter() < 8 && !Thread.interrupted());
             }
         });
 
         try {
             //If this fails and throws an exception and the socket will be closed
-            waitForProtocolDetectionFinish(future, in.getReadSize());
+            waitForProtocolDetectionFinish(future, in.getReceiveCounter());
         } finally {
             //call cancel in case task didn't complete which will interrupt the task
             future.cancel(true);
         }
         in.stop();
 
-        InitBuffer initBuffer = new InitBuffer(in.getReadSize().get(), ByteBuffer.allocate(in.getReadData().length));
+        InitBuffer initBuffer = new InitBuffer(in.getReceiveCounter(), ByteBuffer.allocate(in.getReadData().length));
         initBuffer.buffer.put(in.getReadData());
 
         ProtocolInfo protocolInfo = detectProtocol(in.getReadData());

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

@@ -148,16 +148,10 @@ public SSLEngine getSslSession() {
 
     private volatile byte[] readData;
 
-    private final AtomicInteger readSize = new AtomicInteger();
-
     public byte[] getReadData() {
         return readData != null ? readData : new byte[0];
     }
 
-    public AtomicInteger getReadSize() {
-        return readSize;
-    }
-
     @Override
     public void serviceRead() {
         try {
@@ -187,14 +181,13 @@ public void serviceRead() {
                         break;
                     }
 
-                    receiveCounter += readCount;
-                    readSize.addAndGet(readCount);
+                    receiveCounter.addAndGet(readCount);
                 }
 
                 if (status == SSLEngineResult.Status.OK && handshakeStatus != SSLEngineResult.HandshakeStatus.NEED_UNWRAP) {
                     processCommand(plain);
                     //we have received enough bytes to detect the protocol
-                    if (receiveCounter >= 8) {
+                    if (receiveCounter.get() >= 8) {
                         break;
                     }
                 }
@@ -208,7 +201,7 @@ public void serviceRead() {
 
     @Override
     protected void processCommand(ByteBuffer plain) throws Exception {
-        ByteBuffer newBuffer = ByteBuffer.allocate(receiveCounter);
+        ByteBuffer newBuffer = ByteBuffer.allocate(receiveCounter.get());
         if (readData != null) {
             newBuffer.put(readData);
         }

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

@@ -215,19 +215,27 @@ public void run() {
     }
 
     //Only used for the auto transport to abort the openwire init method early if already initialized
-    boolean openWireInititialized = false;
+    boolean openWireInitialized = false;
 
     protected void doOpenWireInit() throws Exception {
         //Do this later to let wire format negotiation happen
-        if (initBuffer != null && !openWireInititialized && this.wireFormat instanceof OpenWireFormat) {
+        if (initBuffer != null && !openWireInitialized && this.wireFormat instanceof OpenWireFormat) {
             initBuffer.buffer.flip();
             if (initBuffer.buffer.hasRemaining()) {
                 nextFrameSize = -1;
-                receiveCounter += initBuffer.readSize;
-                processCommand(initBuffer.buffer);
-                processCommand(initBuffer.buffer);
+                receiveCounter.addAndGet(initBuffer.readSize);
+                do {
+                    // This should almost always just be called 2 times, the first call reads
+                    // the size and allocates space for the frame. The second call reads
+                    // in the frame to process. This is enough to read in the initial WireFormatInfo
+                    // frame that will be sent. However, it's technically possible for
+                    // there to be extra data after that if more bytes came in during the initial
+                    // socket read if a client sends more, so keep calling until we process the
+                    // entire initial buffer before we continue so we do not miss any bytes.
+                    processCommand(initBuffer.buffer);
+                } while (initBuffer.buffer.hasRemaining());
                 initBuffer.buffer.clear();
-                openWireInititialized = true;
+                openWireInitialized = true;
             }
         }
     }
@@ -277,7 +285,7 @@ public void serviceRead() {
                         break;
                     }
 
-                    receiveCounter += readCount;
+                    receiveCounter.addAndGet(readCount);
                 }
 
                 if (status == SSLEngineResult.Status.OK && handshakeStatus != SSLEngineResult.HandshakeStatus.NEED_UNWRAP) {

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

@@ -124,7 +124,7 @@ protected void serviceRead() {
                     break;
                 }
 
-                this.receiveCounter += readSize;
+                this.receiveCounter.addAndGet(readSize);
                 if (currentBuffer.hasRemaining()) {
                     continue;
                 }

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

@@ -34,6 +34,7 @@
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
 import javax.net.SocketFactory;
@@ -130,8 +131,8 @@ public class TcpTransport extends TransportThreadSupport implements Transport, S
     protected boolean useLocalHost = false;
     protected int minmumWireFormatVersion;
     protected SocketFactory socketFactory;
-    protected final AtomicReference<CountDownLatch> stoppedLatch = new AtomicReference<CountDownLatch>();
-    protected volatile int receiveCounter;
+    protected final AtomicReference<CountDownLatch> stoppedLatch = new AtomicReference<>();
+    protected final AtomicInteger receiveCounter = new AtomicInteger();
 
     protected Map<String, Object> socketOptions;
     private int soLinger = Integer.MIN_VALUE;
@@ -615,22 +616,22 @@ protected void initializeStreams() throws Exception {
         TcpBufferedInputStream buffIn = new TcpBufferedInputStream(socket.getInputStream(), ioBufferSize) {
             @Override
             public int read() throws IOException {
-                receiveCounter++;
+                receiveCounter.incrementAndGet();
                 return super.read();
             }
             @Override
             public int read(byte[] b, int off, int len) throws IOException {
-                receiveCounter++;
+                receiveCounter.incrementAndGet();
                 return super.read(b, off, len);
             }
             @Override
             public long skip(long n) throws IOException {
-                receiveCounter++;
+                receiveCounter.incrementAndGet();
                 return super.skip(n);
             }
             @Override
             protected void fill() throws IOException {
-                receiveCounter++;
+                receiveCounter.incrementAndGet();
                 super.fill();
             }
         };
@@ -684,7 +685,7 @@ public <T> T narrow(Class<T> target) {
 
     @Override
     public int getReceiveCounter() {
-        return receiveCounter;
+        return receiveCounter.get();
     }
 
     public static class InitBuffer {

activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTNIOSSLTransport.java

@@ -70,12 +70,12 @@ protected void processCommand(ByteBuffer plain) throws Exception {
     protected void doInit() throws Exception {
         if (initBuffer != null) {
             nextFrameSize = -1;
-            receiveCounter += initBuffer.readSize;
+            receiveCounter.addAndGet(initBuffer.readSize);
             initBuffer.buffer.flip();
             processCommand(initBuffer.buffer);
         }
         super.doInit();
     }
 
 
-}
+}

activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTNIOTransport.java

@@ -131,7 +131,7 @@ protected void processBuffer(ByteBuffer buffer, int readSize) throws Exception {
         DataByteArrayInputStream dis = new DataByteArrayInputStream(buffer.array());
         codec.parse(dis, readSize);
 
-        receiveCounter += readSize;
+        receiveCounter.addAndGet(readSize);
 
         // clear the buffer
         buffer.clear();
@@ -154,4 +154,4 @@ protected void doStop(ServiceStopper stopper) throws Exception {
             super.doStop(stopper);
         }
     }
-}
+}