Sync ProtonJ2 with upstream

Author: acogoluegnes

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

@@ -620,8 +620,6 @@ void flush() {
     //----- Private implementation events handlers and utility methods
 
     private void handleLocalOpen(org.apache.qpid.protonj2.engine.Connection connection) {
-        connection.tickAuto(getScheduler());
-
         if (options.openTimeout() > 0) {
             executor.schedule(() -> {
                 if (!openFuture.isDone()) {
@@ -921,6 +919,7 @@ private void initializeProtonResources(ReconnectLocation location) throws Client
                         .localCloseHandler(this::handleLocalClose)
                         .openHandler(this::handleRemoteOpen)
                         .closeHandler(this::handleRemoteClose);
+        protonConnection.tickAuto(getScheduler());
 
         configureEngineSaslSupport();
     }

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

@@ -20,7 +20,6 @@
 
 import io.netty.channel.MultiThreadIoEventLoopGroup;
 import io.netty.channel.epoll.EpollIoHandler;
-import io.netty.channel.nio.NioIoHandler;
 import org.apache.qpid.protonj2.client.TransportOptions;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -37,19 +36,33 @@ public final class EpollSupport {
     public static final String NAME = "EPOLL";
 
     public static boolean isAvailable(TransportOptions transportOptions) {
+        return transportOptions.allowNativeIO() && isAvailable();
+    }
+
+    public static boolean isAvailable() {
         try {
-            return transportOptions.allowNativeIO() && Epoll.isAvailable();
+            return Epoll.isAvailable();
         } catch (NoClassDefFoundError ncdfe) {
             LOG.debug("Unable to check for Epoll support due to missing class definition", ncdfe);
             return false;
         }
     }
 
     public static EventLoopGroup createGroup(int nThreads, ThreadFactory ioThreadFactory) {
+        ensureAvailability();
         return new MultiThreadIoEventLoopGroup(nThreads, ioThreadFactory, EpollIoHandler.newFactory());
     }
 
     public static Class<? extends Channel> getChannelClass() {
+        ensureAvailability();
+
         return EpollSocketChannel.class;
     }
+
+    public static void ensureAvailability() {
+        if (!isAvailable()) {
+            throw new UnsupportedOperationException(
+                "Netty Epoll support is not enabled because the Netty library indicates it is not present or disabled");
+        }
+    }
 }

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

@@ -16,6 +16,7 @@
  */
 package org.apache.qpid.protonj2.client.transport.netty4;
 
+import java.lang.reflect.Method;
 import java.util.concurrent.ThreadFactory;
 
 import io.netty.channel.MultiThreadIoEventLoopGroup;
@@ -26,29 +27,73 @@
 
 import io.netty.channel.Channel;
 import io.netty.channel.EventLoopGroup;
-import io.netty.channel.uring.IoUring;
-import io.netty.channel.uring.IoUringSocketChannel;
 
+@SuppressWarnings("unchecked")
 public final class IOUringSupport {
 
     private static final Logger LOG = LoggerFactory.getLogger(IOUringSupport.class);
 
     public static final String NAME = "IO_URING";
 
-    public static boolean isAvailable(TransportOptions transportOptions) {
+    private static final boolean AVAILABLE;
+    private static final Class<? extends Channel> SOCKET_CHANNEL_CLASS;
+
+    static {
+        boolean available = false;
+        Class<? extends Channel> socketChannelClass = null;
+
+        // Try for new Netty built in IoUring before falling back to incubator checks
         try {
-            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;
+            final Class<?> ioUring = Class.forName("io.netty.channel.uring.IoUring");
+            final Method isAvailable = ioUring.getDeclaredMethod("isAvailable", (Class<?>[])null);
+            final Class<?> eventLoopGroup = Class.forName("io.netty.channel.MultiThreadIoEventLoopGroup");
+            final Class<?> ioUringHandler = Class.forName("io.netty.channel.uring.IoUringIoHandler");
+            final Class<?> ioUringHandlerFactory = Class.forName("io.netty.channel.IoHandlerFactory");
+
+            socketChannelClass = (Class<? extends Channel>) Class.forName("io.netty.channel.uring.IoUringSocketChannel");
+            available = (boolean) isAvailable.invoke(null);
+        } catch (Exception e) {
+            LOG.debug("Unable to enable netty io_uring support due to error", e);
         }
+
+        if (!available) {
+            try {
+                final Class<?> ioUring = Class.forName("io.netty.incubator.channel.uring.IOUring");
+                final Method isAvailable = ioUring.getDeclaredMethod("isAvailable");
+                final Class<?> eventLoopGroup = Class.forName("io.netty.incubator.channel.uring.IOUringEventLoopGroup");
+
+                socketChannelClass = (Class<? extends Channel>) Class.forName("io.netty.incubator.channel.uring.IOUringSocketChannel");
+                available = (boolean) isAvailable.invoke(null);
+            } catch (Exception e) {
+                LOG.debug("Unable to enable netty incubator io_uring support due to error", e);
+            }
+        }
+
+        AVAILABLE = available;
+        SOCKET_CHANNEL_CLASS = socketChannelClass;
+    }
+
+    public static boolean isAvailable(TransportOptions transportOptions) {
+        return transportOptions.allowNativeIO() && AVAILABLE;
+    }
+
+    public static boolean isAvailable() {
+        return AVAILABLE;
     }
 
     public static EventLoopGroup createGroup(int nThreads, ThreadFactory ioThreadFactory) {
         return new MultiThreadIoEventLoopGroup(nThreads, ioThreadFactory, IoUringIoHandler.newFactory());
     }
 
     public static Class<? extends Channel> getChannelClass() {
-        return IoUringSocketChannel.class;
+        ensureAvailability();
+        return SOCKET_CHANNEL_CLASS;
+    }
+
+    public static void ensureAvailability() {
+        if (!AVAILABLE) {
+            throw new UnsupportedOperationException(
+                "Netty io_ring support is not enabled because the Netty library indicates it is not present or disabled");
+        }
     }
 }

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

@@ -36,19 +36,33 @@ public final class KQueueSupport {
     public static final String NAME = "KQUEUE";
 
     public static boolean isAvailable(TransportOptions transportOptions) {
+        return transportOptions.allowNativeIO() && isAvailable();
+    }
+
+    public static boolean isAvailable() {
         try {
-            return transportOptions.allowNativeIO() && KQueue.isAvailable();
+            return KQueue.isAvailable();
         } catch (NoClassDefFoundError ncdfe) {
             LOG.debug("Unable to check for KQueue support due to missing class definition", ncdfe);
             return false;
         }
     }
 
     public static EventLoopGroup createGroup(int nThreads, ThreadFactory ioThreadFactory) {
-        return new MultiThreadIoEventLoopGroup(nThreads, KQueueIoHandler.newFactory());
+        ensureAvailability();
+        return new MultiThreadIoEventLoopGroup(nThreads, ioThreadFactory, KQueueIoHandler.newFactory());
     }
 
     public static Class<? extends Channel> getChannelClass() {
+        ensureAvailability();
+
         return KQueueSocketChannel.class;
     }
+
+    public static void ensureAvailability() {
+        if (!isAvailable()) {
+            throw new UnsupportedOperationException(
+                "Netty KQueue support is not enabled because the Netty library indicates it is not present or disabled");
+        }
+    }
 }

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

@@ -187,12 +187,7 @@ public static SSLEngine createJdkSslEngine(String host, int port, SSLContext con
         engine.setEnabledProtocols(buildEnabledProtocols(engine, options));
         engine.setEnabledCipherSuites(buildEnabledCipherSuites(engine, options));
         engine.setUseClientMode(true);
-
-        if (options.verifyHost()) {
-            SSLParameters sslParameters = engine.getSSLParameters();
-            sslParameters.setEndpointIdentificationAlgorithm("HTTPS");
-            engine.setSSLParameters(sslParameters);
-        }
+        engine.setSSLParameters(createSSLParameters(engine, options));
 
         return engine;
     }
@@ -274,12 +269,25 @@ public static SSLEngine createOpenSslEngine(ByteBufAllocator allocator, String h
         engine.setEnabledProtocols(buildEnabledProtocols(engine, options));
         engine.setEnabledCipherSuites(buildEnabledCipherSuites(engine, options));
         engine.setUseClientMode(true);
+        engine.setSSLParameters(createSSLParameters(engine, options));
 
         return engine;
     }
 
     //----- Internal support methods -----------------------------------------//
 
+    private static SSLParameters createSSLParameters(SSLEngine engine, SslOptions options) {
+        final SSLParameters sslParameters = engine.getSSLParameters();
+
+        if (options.verifyHost()) {
+            sslParameters.setEndpointIdentificationAlgorithm("HTTPS");
+        } else {
+            sslParameters.setEndpointIdentificationAlgorithm(null);
+        }
+
+        return sslParameters;
+    }
+
     private static String[] buildEnabledProtocols(SSLEngine engine, SslOptions options) {
         List<String> enabledProtocols = new ArrayList<String>();
 

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

@@ -41,6 +41,7 @@
 import io.netty.bootstrap.Bootstrap;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
@@ -155,7 +156,15 @@ public void initChannel(Channel transportChannel) throws Exception {
 
         configureNetty(bootstrap, options);
 
-        bootstrap.connect(getHost(), getPort()).addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE);
+        bootstrap.connect(getHost(), getPort()).addListener(new ChannelFutureListener() {
+
+            @Override
+            public void operationComplete(ChannelFuture future) throws Exception {
+                if (!future.isSuccess()) {
+                    handleTransportFailure(future.channel(), future.cause());
+                }
+            }
+        });
 
         return this;
     }

src/main/qpid/org/apache/qpid/protonj2/codec/Encoder.java

@@ -591,7 +591,7 @@ public interface Encoder {
     void writeBinary(ProtonBuffer buffer, EncoderState state, ProtonBuffer value) throws EncodeException;
 
     /**
-     * Writes the contents of the given {@link byte[]} value into the provided {@link ProtonBuffer}
+     * Writes the contents of the given <code>byte[]</code> value into the provided {@link ProtonBuffer}
      * instance as an AMQP Binary type.
      * <p>
      * If the provided value to write is null an AMQP null type is encoded into the target buffer.

src/main/qpid/org/apache/qpid/protonj2/engine/Engine.java

@@ -187,12 +187,16 @@ default void accept(ProtonBuffer input) throws EngineStateException {
      * Allows the engine to manage idle timeout processing by providing it the single threaded executor
      * context where all transport work is done which ensures singled threaded access while removing the
      * need for the client library or server application to manage calls to the {@link Engine#tick} methods.
+     * <p>
+     * The API should allow for configuring auto idle timeout handling before the connection has opened and
+     * should react to both local and remote open performatives passing through the engine to configure read
+     * and write checks under the constraints of the local and remote idle timeout configurations.
      *
      * @param executor
      *      The single threaded execution context where all engine work takes place.
      *
      * @throws IllegalStateException if the {@link Engine} is already performing auto tick handling.
-     * @throws EngineStateException if the Engine state precludes accepting new input.
+     * @throws EngineStateException if the Engine state precludes accepting new input (shutdown or failed).
      *
      * @return this {@link Engine}
      */
@@ -202,12 +206,16 @@ default void accept(ProtonBuffer input) throws EngineStateException {
      * Allows the engine to manage idle timeout processing by providing it the single threaded executor
      * context where all transport work is done which ensures singled threaded access while removing the
      * need for the client library or server application to manage calls to the {@link Engine#tick} methods.
+     * <p>
+     * The API should allow for configuring auto idle timeout handling before the connection has opened and
+     * should react to both local and remote open performatives passing through the engine to configure read
+     * and write checks under the constraints of the local and remote idle timeout configurations.
      *
      * @param scheduler
      *      The single threaded execution context where all engine work takes place.
      *
      * @throws IllegalStateException if the {@link Engine} is already performing auto tick handling.
-     * @throws EngineStateException if the Engine state precludes accepting new input.
+     * @throws EngineStateException if the Engine state precludes accepting new input (shutdown or failed).
      *
      * @return this {@link Engine}
      */

src/main/qpid/org/apache/qpid/protonj2/engine/impl/ProtonConnection.java

@@ -94,7 +94,7 @@ public class ProtonConnection extends ProtonEndpoint<Connection> implements Conn
      * Create a new unbound Connection instance.
      *
      * @param engine
-     * 		Parent engine that created and owns this {@link Connection} insatnce.
+     * 		Parent engine that created and owns this {@link Connection} instance.
      */
     ProtonConnection(ProtonEngine engine) {
         super(engine);
@@ -128,6 +128,7 @@ public ProtonConnection open() throws EngineStateException {
                 syncLocalStateWithRemote();
             } finally {
                 fireLocalOpen();
+                engine.handleLocalOpen(this);
             }
         }
 
@@ -446,6 +447,7 @@ public void handleOpen(Open open, ProtonBuffer payload, int channel, ProtonEngin
         remoteOpen = open;
 
         fireRemoteOpen();
+        engine.handleRemoteOpen(this);
     }
 
     @Override