Netty HttpClient changes (#45143)

Netty HttpClient changes

Author: alzimmermsft

sdk/core/azure-core-http-netty/src/main/java/com/azure/core/http/netty/NettyAsyncHttpClientBuilder.java

@@ -174,17 +174,34 @@ public NettyAsyncHttpClientBuilder(HttpClient nettyHttpClient) {
     public com.azure.core.http.HttpClient build() {
         HttpClient nettyHttpClient;
 
+        Configuration buildConfiguration
+            = (configuration == null) ? Configuration.getGlobalConfiguration() : configuration;
+
+        ProxyOptions buildProxyOptions
+            = proxyOptions == null ? ProxyOptions.fromConfiguration(buildConfiguration, true) : proxyOptions;
+
+        /*
+         * Only configure the custom authorization challenge handler and challenge holder when using an authenticated
+         * HTTP proxy. All other proxying such as SOCKS4, SOCKS5, and anonymous HTTP will use Netty's built-in handlers.
+         */
+        boolean useCustomProxyHandler = shouldUseCustomProxyHandler(buildProxyOptions);
+        AuthorizationChallengeHandler handler = useCustomProxyHandler
+            ? new AuthorizationChallengeHandler(buildProxyOptions.getUsername(), buildProxyOptions.getPassword())
+            : null;
+        AtomicReference<ChallengeHolder> proxyChallengeHolder = useCustomProxyHandler ? new AtomicReference<>() : null;
+
         // Used to track if the builder set the DefaultAddressResolverGroup. If it did, when proxying it allows the
         // no-op address resolver to be set.
-        boolean addressResolverWasSetByBuilder = false;
         if (this.baseHttpClient != null) {
             nettyHttpClient = baseHttpClient;
         } else if (this.connectionProvider != null) {
-            nettyHttpClient = HttpClient.create(this.connectionProvider).resolver(DefaultAddressResolverGroup.INSTANCE);
-            addressResolverWasSetByBuilder = true;
+            nettyHttpClient = HttpClient.create(this.connectionProvider);
         } else {
-            nettyHttpClient = HttpClient.create().resolver(DefaultAddressResolverGroup.INSTANCE);
-            addressResolverWasSetByBuilder = true;
+            nettyHttpClient = HttpClient.create();
+        }
+
+        if (proxyOptions == null) {
+            nettyHttpClient = nettyHttpClient.resolver(DefaultAddressResolverGroup.INSTANCE);
         }
 
         long writeTimeout = getTimeout(this.writeTimeout, getDefaultWriteTimeout()).toMillis();
@@ -213,22 +230,6 @@ public com.azure.core.http.HttpClient build() {
             nettyHttpClient.wiretap(enableWiretap);
         }
 
-        Configuration buildConfiguration
-            = (configuration == null) ? Configuration.getGlobalConfiguration() : configuration;
-
-        ProxyOptions buildProxyOptions
-            = proxyOptions == null ? ProxyOptions.fromConfiguration(buildConfiguration, true) : proxyOptions;
-
-        /*
-         * Only configure the custom authorization challenge handler and challenge holder when using an authenticated
-         * HTTP proxy. All other proxying such as SOCKS4, SOCKS5, and anonymous HTTP will use Netty's built-in handlers.
-         */
-        boolean useCustomProxyHandler = shouldUseCustomProxyHandler(buildProxyOptions);
-        AuthorizationChallengeHandler handler = useCustomProxyHandler
-            ? new AuthorizationChallengeHandler(buildProxyOptions.getUsername(), buildProxyOptions.getPassword())
-            : null;
-        AtomicReference<ChallengeHolder> proxyChallengeHolder = useCustomProxyHandler ? new AtomicReference<>() : null;
-
         boolean addProxyHandler = false;
 
         if (eventLoopGroup != null) {
@@ -257,6 +258,12 @@ public com.azure.core.http.HttpClient build() {
                                     handler, proxyChallengeHolder));
                     }
                 });
+
+                AddressResolverGroup<?> resolver = nettyHttpClient.configuration().resolver();
+                if (resolver == null) {
+                    // This mimics behaviors seen when Reactor Netty proxying is used.
+                    nettyHttpClient = nettyHttpClient.resolver(NoopAddressResolverGroup.INSTANCE);
+                }
             } else {
                 nettyHttpClient
                     = nettyHttpClient.proxy(proxy -> proxy.type(toReactorNettyProxyType(buildProxyOptions.getType()))
@@ -265,12 +272,6 @@ public com.azure.core.http.HttpClient build() {
                         .password(ignored -> buildProxyOptions.getPassword())
                         .nonProxyHosts(buildProxyOptions.getNonProxyHosts()));
             }
-
-            AddressResolverGroup<?> resolver = nettyHttpClient.configuration().resolver();
-            if (resolver == null || addressResolverWasSetByBuilder) {
-                // This mimics behaviors seen when Reactor Netty proxying is used.
-                nettyHttpClient = nettyHttpClient.resolver(NoopAddressResolverGroup.INSTANCE);
-            }
         }
 
         return new NettyAsyncHttpClient(nettyHttpClient, disableBufferCopy, addProxyHandler);

sdk/core/azure-core-http-netty/src/main/java/com/azure/core/http/netty/implementation/NettyUtility.java

@@ -76,10 +76,18 @@ public static void closeConnection(Connection reactorNettyConnection) {
             // From there the only thing that needs to be checked is whether the inbound has been disposed (completed),
             // and if not dispose it (aka drain it).
             if (!channelOperations.isInboundDisposed()) {
-                channelOperations.channel().eventLoop().execute(channelOperations::discard);
+                if (channelOperations.channel().isRegistered()) {
+                    channelOperations.channel().eventLoop().execute(channelOperations::discard);
+                } else {
+                    channelOperations.discard();
+                }
             }
         } else if (!reactorNettyConnection.isDisposed()) {
-            reactorNettyConnection.channel().eventLoop().execute(reactorNettyConnection::dispose);
+            if (reactorNettyConnection.channel().isRegistered()) {
+                reactorNettyConnection.channel().eventLoop().execute(reactorNettyConnection::dispose);
+            } else {
+                reactorNettyConnection.dispose();
+            }
         }
     }