Sync ProtonJ2 with upstream

Author: acogoluegnes

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

@@ -40,6 +40,7 @@ public class TransportOptions implements Cloneable {
     public static final boolean DEFAULT_TRACE_BYTES = false;
     public static final int DEFAULT_LOCAL_PORT = 0;
     public static final boolean DEFAULT_USE_WEBSOCKETS = false;
+    public static final boolean DEFAULT_WEBSOCKET_COMPRESSION = false;
     public static final int DEFAULT_WEBSOCKET_MAX_FRAME_SIZE = 65535;
     private static final String[] DEFAULT_NATIVEIO_PREFERENCES_ARRAY = { "EPOLL", "KQUEUE" };
     public static final List<String> DEFAULT_NATIVEIO_PREFERENCES =
@@ -62,6 +63,7 @@ public class TransportOptions implements Cloneable {
     private boolean useWebSockets = DEFAULT_USE_WEBSOCKETS;
     private String webSocketPath;
     private int webSocketMaxFrameSize = DEFAULT_WEBSOCKET_MAX_FRAME_SIZE;
+    private boolean webSocketCompression = DEFAULT_WEBSOCKET_COMPRESSION;
 
     private final Map<String, String> webSocketHeaders = new HashMap<>();
 
@@ -454,6 +456,31 @@ public TransportOptions webSocketMaxFrameSize(int maxFrameSize) {
         return this;
     }
 
+    /**
+     * @return the configured value for the WebSocket compression support enabled flag.
+     */
+    public boolean webSocketCompression() {
+        return webSocketCompression;
+    }
+
+    /**
+     * Set to true to configure the transport layer as a WebSocket based connection that
+     * support compression of the WebSocket packets. This option simply allows the client
+     * to support compression if the server offers support but does not influence the server
+     * side, if the server does not offer support for compression of WS packets then this
+     * value has no affect on the WS packets and they remain uncompressed as if not enabled.
+     * (default is disabled).
+     *
+     * @param enabled
+     * 		should the transport support WebSocket compression if server offers it.
+     *
+     * @return this {@link TransportOptions} instance.
+     */
+    public TransportOptions webSocketCompression(boolean enabled) {
+        this.webSocketCompression = enabled;
+        return this;
+    }
+
     /**
      * Copy all configuration into the given {@link TransportOptions} from this instance.
      *
@@ -481,6 +508,7 @@ public TransportOptions copyInto(TransportOptions other) {
         other.webSocketPath(webSocketPath());
         other.webSocketHeaders().putAll(webSocketHeaders);
         other.webSocketMaxFrameSize(webSocketMaxFrameSize());
+        other.webSocketCompression(webSocketCompression());
 
         return other;
     }

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

@@ -559,6 +559,7 @@ ClientConnection connect() throws ClientException {
             openFuture.failed(failureCause);
             closeFuture.complete(this);
             ioContext.shutdown();
+            notifications.shutdown();
 
             throw failureCause;
         }
@@ -754,7 +755,6 @@ private void handleEngineFailure(Engine engine) {
         } else {
             FAILURE_CAUSE_UPDATER.compareAndSet(this, null, failureCause);
             engine.shutdown();
-            notifications.shutdownNow();
         }
     }
 
@@ -778,23 +778,24 @@ private void handleEngineShutdown(Engine engine) {
 
         ioContext.shutdownAsync();
 
-        if (failureCause != null)
-        {
-            openFuture.failed(failureCause);
-            closeFuture.complete(this);
+        try {
+            if (failureCause != null) {
+                openFuture.failed(failureCause);
+                closeFuture.complete(this);
 
-            LOG.warn("Connection {} has failed due to: {}", getId(), failureCause != null ?
-                    failureCause.getClass().getSimpleName() + " -> " + failureCause.getMessage() : "No failure details provided.");
+                LOG.warn("Connection {} has failed due to: {}", getId(), failureCause != null ?
+                         failureCause.getClass().getSimpleName() + " -> " + failureCause.getMessage() : "No failure details provided.");
 
-           submitDisconnectionEvent(options.disconnectedHandler(), transport.getHost(), transport.getPort(), failureCause);
-        }
-        else
-        {
-            openFuture.complete(this);
-            closeFuture.complete(this);
-        }
+               submitDisconnectionEvent(options.disconnectedHandler(), transport.getHost(), transport.getPort(), failureCause);
+            } else {
+                openFuture.complete(this);
+                closeFuture.complete(this);
+            }
 
-        client.unregisterConnection(this);
+            client.unregisterConnection(this);
+        } finally {
+            submitNotificationShutdownTask();
+        }
     }
 
     private void submitConnectionEvent(BiConsumer<Connection, ConnectionEvent> handler, String host, int port, ClientIOException cause) {
@@ -829,6 +830,22 @@ private void submitDisconnectionEvent(BiConsumer<Connection, DisconnectionEvent>
         }
     }
 
+    private void submitNotificationShutdownTask() {
+        try {
+            if (!notifications.isShutdown()) {
+                notifications.submit(() -> {
+                    try {
+                        notifications.shutdown();
+                    } catch (Exception ex) {
+                        LOG.trace("Shutdown of notification event handler threw an error: ", ex);
+                    }
+                });
+            }
+        } catch (Exception ex) {
+            LOG.trace("Error thrown while attempting to submit shutdown of notification executor: ", ex);
+        }
+    }
+
     private Engine configureEngineSaslSupport() {
         if (options.saslOptions().saslEnabled()) {
             SaslMechanismSelector mechSelector =

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

@@ -47,6 +47,7 @@
 import io.netty.handler.codec.http.websocketx.WebSocketClientHandshakerFactory;
 import io.netty.handler.codec.http.websocketx.WebSocketFrame;
 import io.netty.handler.codec.http.websocketx.WebSocketVersion;
+import io.netty.handler.codec.http.websocketx.extensions.compression.WebSocketClientCompressionHandler;
 import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.GenericFutureListener;
 import io.netty.util.concurrent.ScheduledFuture;
@@ -151,6 +152,9 @@ protected ChannelInboundHandlerAdapter createChannelHandler() {
     protected void addAdditionalHandlers(ChannelPipeline pipeline) {
         pipeline.addLast(new HttpClientCodec());
         pipeline.addLast(new HttpObjectAggregator(8192));
+        if (options.webSocketCompression()) {
+            pipeline.addLast(WebSocketClientCompressionHandler.INSTANCE);
+        }
     }
 
     @Override

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

@@ -49,6 +49,7 @@
 import io.netty5.handler.codec.http.websocketx.WebSocketClientHandshakerFactory;
 import io.netty5.handler.codec.http.websocketx.WebSocketFrame;
 import io.netty5.handler.codec.http.websocketx.WebSocketVersion;
+import io.netty5.handler.codec.http.websocketx.extensions.compression.WebSocketClientCompressionHandler;
 import io.netty5.util.concurrent.Future;
 import io.netty5.util.concurrent.FutureListener;
 
@@ -98,6 +99,9 @@ protected ChannelHandler createChannelHandler() {
     protected void addAdditionalHandlers(ChannelPipeline pipeline) {
         pipeline.addLast(new HttpClientCodec());
         pipeline.addLast(new HttpObjectAggregator<DefaultHttpContent>(8192));
+        if (options.webSocketCompression()) {
+            pipeline.addLast(WebSocketClientCompressionHandler.INSTANCE);
+        }
     }
 
     @Override
@@ -169,7 +173,6 @@ public void operationComplete(Future<? extends Void> future) throws Exception {
             super.channelActive(context);
         }
 
-        @SuppressWarnings("resource")
         @Override
         protected void messageReceived(ChannelHandlerContext ctx, Object message) throws Exception {
             LOG.trace("New data read: incoming: {}", message);