Sync ProtonJ2

Author: acogoluegnes

pom.xml

@@ -134,9 +134,9 @@
             <classifier>osx-x86_64</classifier>
         </dependency>
         <dependency>
-            <groupId>io.netty.incubator</groupId>
-            <artifactId>netty-incubator-transport-native-io_uring</artifactId>
-            <version>${netty4.iouring.version}</version>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-transport-native-io_uring</artifactId>
+            <version>${netty4.version}</version>
             <classifier>linux-x86_64</classifier>
         </dependency>
         <!-- End of QPid dependencies -->

src/main/qpid/org/apache/qpid/protonj2/buffer/ProtonBufferAccessors.java

@@ -24,6 +24,16 @@
  */
 public interface ProtonBufferAccessors {
 
+    /**
+     * Look ahead an return the next byte that would be read from a call to readByte or
+     * a call to getByte at the current read offset.
+     *
+     * @return the next readable byte without advancing the read offset.
+     *
+     * @throws IndexOutOfBoundsException if there is no readable bytes left in the buffer.
+     */
+    byte peekByte();
+
     /**
      * Reads a single byte at the given index and returns it without modification to the target
      * buffer read offset.

src/main/qpid/org/apache/qpid/protonj2/buffer/impl/ProtonByteArrayBuffer.java

@@ -351,6 +351,12 @@ public int hashCode() {
 
     //----- Indexed Get operations
 
+    @Override
+    public byte peekByte() {
+        checkPeek();
+        return ProtonBufferUtils.readByte(array, offset(readOffset));
+    }
+
     @Override
     public byte getByte(int index) {
         checkGet(index, Byte.BYTES);
@@ -965,6 +971,16 @@ private void checkWrite(int index, int size, boolean allowExpansion) {
         }
     }
 
+    private void checkPeek() {
+        if (readOffset == writeOffset) {
+            if (closed) {
+                throw ProtonBufferUtils.genericBufferIsClosed(this);
+            } else {
+                throw ProtonBufferUtils.genericOutOfBounds(this, readOffset);
+            }
+        }
+    }
+
     private void checkRead(int index, int size) {
         if (index < 0 || writeOffset < index + size || closed) {
             if (closed) {

src/main/qpid/org/apache/qpid/protonj2/buffer/impl/ProtonCompositeBufferImpl.java

@@ -758,6 +758,12 @@ public ProtonCompositeBuffer splitComponentsCeil(int splitOffset) {
 
     //----- Offset based get operations
 
+    @Override
+    public byte peekByte() {
+        checkPeek();
+        return findIndexedAccessor(readOffset, Byte.BYTES).getByte(readOffset);
+    }
+
     @Override
     public byte getByte(int index) {
         checkGetBounds(index, Byte.BYTES);
@@ -1426,6 +1432,12 @@ private static RuntimeException closeBuffers(ProtonBuffer[] source, ProtonBuffer
 
     //----- Internal API for composite buffer
 
+    private void checkPeek() {
+        if (readOffset == writeOffset) {
+            throw generateIndexOutOfBounds(readOffset, false);
+        }
+    }
+
     private void checkGetBounds(int index, int size) {
         if (index < 0 || capacity < index + size) {
             throw generateIndexOutOfBounds(index, false);
@@ -1701,6 +1713,11 @@ private int offset(int index) {
         // fall within the available portion of the buffer in this chunk or
         // an IOOBE will again be thrown.
 
+        @Override
+        public byte peekByte() {
+            return parent.buffers[chunkIndex].peekByte();
+        }
+
         @Override
         public byte getByte(int index) {
             return parent.buffers[chunkIndex].getByte(offset(index));
@@ -1817,6 +1834,11 @@ public ProtonBufferAccessors prepare(int chunk) {
             return this;
         }
 
+        @Override
+        public byte peekByte() {
+            throw new UnsupportedOperationException();
+        }
+
         @Override
         public byte getByte(int index) {
             throw new UnsupportedOperationException();

src/main/qpid/org/apache/qpid/protonj2/buffer/netty/Netty4ToProtonBufferAdapter.java

@@ -516,6 +516,12 @@ public int hashCode() {
 
     //----- Primitive Get Methods
 
+    @Override
+    public byte peekByte() {
+        checkPeek();
+        return resource.getByte(readOffset);
+    }
+
     @Override
     public byte getByte(int index) {
         checkGet(index, Byte.BYTES);
@@ -1035,6 +1041,16 @@ protected RuntimeException resourceIsClosedException() {
 
     //----- Internal utilities for mapping to netty
 
+    private void checkPeek() {
+        if (readOffset == writeOffset) {
+            if (closed) {
+                throw ProtonBufferUtils.genericBufferIsClosed(this);
+            } else {
+                throw ProtonBufferUtils.genericOutOfBounds(this, readOffset);
+            }
+        }
+    }
+
     private void checkRead(int index, int size) {
         if (index < 0 || writeOffset < index + size || closed) {
             if (closed) {

src/main/qpid/org/apache/qpid/protonj2/client/SaslOptions.java

@@ -24,11 +24,11 @@
 /**
  * Connection options that are applied to the SASL layer.
  */
-public class SaslOptions {
+public class SaslOptions implements Cloneable {
 
-	/**
-	 * The client default configuration value for SASL enabled state (default is true)
-	 */
+    /**
+     * The client default configuration value for SASL enabled state (default is true)
+     */
     public static final boolean DEFAULT_SASL_ENABLED = true;
 
     private boolean saslEnabled = DEFAULT_SASL_ENABLED;

src/main/qpid/org/apache/qpid/protonj2/client/impl/ClientDeliveryState.java

@@ -128,6 +128,10 @@ public static class ClientAccepted extends ClientDeliveryState {
 
         private static final ClientAccepted INSTANCE = new ClientAccepted();
 
+        private ClientAccepted() {
+            // Singleton
+        }
+
         @Override
         public Type getType() {
             return Type.ACCEPTED;
@@ -153,6 +157,10 @@ public static class ClientReleased extends ClientDeliveryState {
 
         private static final ClientReleased INSTANCE = new ClientReleased();
 
+        private ClientReleased() {
+            // Singleton
+        }
+
         @Override
         public Type getType() {
             return Type.RELEASED;

src/main/qpid/org/apache/qpid/protonj2/client/impl/ClientNextReceiverSelector.java

@@ -162,7 +162,6 @@ private ClientReceiver selectNextAvailable() {
         return result != null ? result : selectFirstAvailable();
     }
 
-    @SuppressWarnings("resource")
     private ClientReceiver selectFirstAvailable() {
         return session.getProtonSession().receivers().stream()
                 .filter((r) -> r.getLinkedResource() instanceof ClientReceiver &&
@@ -172,7 +171,6 @@ private ClientReceiver selectFirstAvailable() {
                 .orElse(null);
     }
 
-    @SuppressWarnings("resource")
     private ClientReceiver selectLargestBacklog() {
         return session.getProtonSession().receivers().stream()
                 .filter((r) -> r.getLinkedResource() instanceof ClientReceiver &&
@@ -182,7 +180,6 @@ private ClientReceiver selectLargestBacklog() {
                 .orElse(null);
     }
 
-    @SuppressWarnings("resource")
     private ClientReceiver selectSmallestBacklog() {
         return session.getProtonSession().receivers().stream()
                 .filter((r) -> r.getLinkedResource() instanceof ClientReceiver &&

src/main/qpid/org/apache/qpid/protonj2/client/transport/netty4/IOUringSupport.java

@@ -18,15 +18,16 @@
 
 import java.util.concurrent.ThreadFactory;
 
+import io.netty.channel.MultiThreadIoEventLoopGroup;
+import io.netty.channel.uring.IoUringIoHandler;
 import org.apache.qpid.protonj2.client.TransportOptions;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import io.netty.channel.Channel;
 import io.netty.channel.EventLoopGroup;
-import io.netty.incubator.channel.uring.IOUring;
-import io.netty.incubator.channel.uring.IOUringEventLoopGroup;
-import io.netty.incubator.channel.uring.IOUringSocketChannel;
+import io.netty.channel.uring.IoUring;
+import io.netty.channel.uring.IoUringSocketChannel;
 
 public final class IOUringSupport {
 
@@ -36,18 +37,18 @@ public final class IOUringSupport {
 
     public static boolean isAvailable(TransportOptions transportOptions) {
         try {
-            return transportOptions.allowNativeIO() && IOUring.isAvailable();
+            return transportOptions.allowNativeIO() && IoUring.isAvailable();
         } catch (NoClassDefFoundError ncdfe) {
             LOG.debug("Unable to check for IO_Uring support due to missing class definition", ncdfe);
             return false;
         }
     }
 
     public static EventLoopGroup createGroup(int nThreads, ThreadFactory ioThreadFactory) {
-        return new IOUringEventLoopGroup(nThreads, ioThreadFactory);
+        return new MultiThreadIoEventLoopGroup(nThreads, ioThreadFactory, IoUringIoHandler.newFactory());
     }
 
     public static Class<? extends Channel> getChannelClass() {
-        return IOUringSocketChannel.class;
+        return IoUringSocketChannel.class;
     }
 }

src/main/qpid/org/apache/qpid/protonj2/client/transport/netty4/WebSocketTransport.java

@@ -153,7 +153,7 @@ protected void addAdditionalHandlers(ChannelPipeline pipeline) {
         pipeline.addLast(new HttpClientCodec());
         pipeline.addLast(new HttpObjectAggregator(8192));
         if (options.webSocketCompression()) {
-            pipeline.addLast(WebSocketClientCompressionHandler.INSTANCE);
+            pipeline.addLast(new WebSocketClientCompressionHandler(0));
         }
     }