Various code clean up including fanout Http2SettingsFrameHandler, adding unit tests, updating javadoc

Author: zoewangg

http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/NettyNioAsyncHttpClient.java

@@ -76,6 +76,7 @@
 @SdkPublicApi
 public final class NettyNioAsyncHttpClient implements SdkAsyncHttpClient {
     private static final Logger log = LoggerFactory.getLogger(NettyNioAsyncHttpClient.class);
+    private static final long MAX_STREAMS_ALLOWED = 4294967295L; // unsigned 32-bit, 2^32 -1
 
     private final SdkEventLoopGroup sdkEventLoopGroup;
     private final SdkChannelPoolMap<URI, ChannelPool> pools;
@@ -87,7 +88,7 @@ public final class NettyNioAsyncHttpClient implements SdkAsyncHttpClient {
     NettyNioAsyncHttpClient(DefaultBuilder builder, AttributeMap serviceDefaultsMap) {
         this.configuration = new NettyConfiguration(serviceDefaultsMap);
         this.protocol = serviceDefaultsMap.get(SdkHttpConfigurationOption.PROTOCOL);
-        this.maxStreams = builder.maxHttp2Streams == null ? Integer.MAX_VALUE : builder.maxHttp2Streams;
+        this.maxStreams = builder.maxHttp2Streams == null ? MAX_STREAMS_ALLOWED : builder.maxHttp2Streams;
         this.sdkEventLoopGroup = eventLoopGroup(builder);
         this.pools = createChannelPoolMap();
         this.sdkChannelOptions = channelOptions(builder);
@@ -106,10 +107,6 @@ public final class NettyNioAsyncHttpClient implements SdkAsyncHttpClient {
         this.maxStreams = maxStreams;
     }
 
-    private SdkChannelOptions channelOptions(DefaultBuilder builder) {
-        return builder.sdkChannelOptions;
-    }
-
     @Override
     public CompletableFuture<Void> execute(AsyncExecuteRequest request) {
         RequestContext ctx = createRequestContext(request);
@@ -120,6 +117,10 @@ public static Builder builder() {
         return new DefaultBuilder();
     }
 
+    private SdkChannelOptions channelOptions(DefaultBuilder builder) {
+        return builder.sdkChannelOptions;
+    }
+
     private RequestContext createRequestContext(AsyncExecuteRequest request) {
         ChannelPool pool = pools.get(poolKey(request.request()));
         return new RequestContext(pool, request, configuration);
@@ -304,7 +305,7 @@ public interface Builder extends SdkAsyncHttpClient.Builder<NettyNioAsyncHttpCli
         /**
          * Sets the {@link SdkEventLoopGroup} to use for the Netty HTTP client. This event loop group may be shared
          * across multiple HTTP clients for better resource and thread utilization. The preferred way to create
-         * an {@link EventLoopGroup} is by using the {@link SdkEventLoopGroup#builder()})} method which will choose the
+         * an {@link EventLoopGroup} is by using the {@link SdkEventLoopGroup#builder()} method which will choose the
          * optimal implementation per the platform.
          *
          * <p>The {@link EventLoopGroup} <b>MUST</b> be closed by the caller when it is ready to
@@ -348,12 +349,15 @@ public interface Builder extends SdkAsyncHttpClient.Builder<NettyNioAsyncHttpCli
         Builder protocol(Protocol protocol);
 
         /**
-         * Add new socket channel option which will be used to create Netty Http client. This allows custom configuration
-         * for Netty.
+         * Configures additional {@link ChannelOption} which will be used to create Netty Http client. This allows custom
+         * configuration for Netty.
+         *
+         * <p>
+         * If a {@link ChannelOption} was previously configured, the old value is replaced.
+         *
          * @param channelOption {@link ChannelOption} to set
          * @param value See {@link ChannelOption} to find the type of value for each option
          * @return This builder for method chaining.
-         * @see SdkEventLoopGroup.Builder
          */
         Builder putChannelOption(ChannelOption channelOption, Object value);
 
@@ -387,12 +391,6 @@ private static final class DefaultBuilder implements Builder {
         private DefaultBuilder() {
         }
 
-        /**
-         * Max allowed connections per endpoint allowed in the connection pool.
-         *
-         * @param maxConcurrency New value for max connections per endpoint.
-         * @return This builder for method chaining.
-         */
         @Override
         public Builder maxConcurrency(Integer maxConcurrency) {
             standardOptions.put(MAX_CONNECTIONS, maxConcurrency);
@@ -403,12 +401,6 @@ public void setMaxConcurrency(Integer maxConnectionsPerEndpoint) {
             maxConcurrency(maxConnectionsPerEndpoint);
         }
 
-        /**
-         * The maximum number of pending acquires allowed. Once this exceeds, acquire tries will be failed.
-         *
-         * @param maxPendingAcquires Max number of pending acquires
-         * @return This builder for method chaining.
-         */
         @Override
         public Builder maxPendingConnectionAcquires(Integer maxPendingAcquires) {
             standardOptions.put(MAX_PENDING_CONNECTION_ACQUIRES, maxPendingAcquires);
@@ -419,12 +411,6 @@ public void setMaxPendingConnectionAcquires(Integer maxPendingAcquires) {
             maxPendingConnectionAcquires(maxPendingAcquires);
         }
 
-        /**
-         * The amount of time to wait for a read on a socket before an exception is thrown.
-         *
-         * @param readTimeout timeout duration
-         * @return this builder for method chaining.
-         */
         @Override
         public Builder readTimeout(Duration readTimeout) {
             Validate.isPositive(readTimeout, "readTimeout");
@@ -436,12 +422,6 @@ public void setReadTimeout(Duration readTimeout) {
             readTimeout(readTimeout);
         }
 
-        /**
-         * The amount of time to wait for a write on a socket before an exception is thrown.
-         *
-         * @param writeTimeout timeout duration
-         * @return this builder for method chaining.
-         */
         @Override
         public Builder writeTimeout(Duration writeTimeout) {
             Validate.isPositive(writeTimeout, "writeTimeout");
@@ -453,12 +433,6 @@ public void setWriteTimeout(Duration writeTimeout) {
             writeTimeout(writeTimeout);
         }
 
-        /**
-         * The amount of time to wait when initially establishing a connection before giving up and timing out.
-         *
-         * @param timeout the timeout duration
-         * @return this builder for method chaining.
-         */
         @Override
         public Builder connectionTimeout(Duration timeout) {
             Validate.isPositive(timeout, "connectionTimeout");
@@ -470,11 +444,6 @@ public void setConnectionTimeout(Duration connectionTimeout) {
             connectionTimeout(connectionTimeout);
         }
 
-        /**
-         * The amount of time to wait when acquiring a connection from the pool before giving up and timing out.
-         * @param connectionAcquisitionTimeout the timeout duration
-         * @return this builder for method chaining.
-         */
         @Override
         public Builder connectionAcquisitionTimeout(Duration connectionAcquisitionTimeout) {
             Validate.isPositive(connectionAcquisitionTimeout, "connectionAcquisitionTimeout");

http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/SdkEventLoopGroup.java

@@ -30,21 +30,33 @@
 /**
  * Provides {@link EventLoopGroup} and {@link ChannelFactory} for {@link NettyNioAsyncHttpClient}.
  * <p>
- * There are two ways to create a new instance.
+ * There are three ways to create a new instance.
  *
  * <ul>
+ * <li>using {@link #builder()} to provide custom configuration of {@link EventLoopGroup}.
+ * This is the preferred configuration method when you just want to customize the {@link EventLoopGroup}</li>
+ *
+ *
+ * <li>Using {@link #create(EventLoopGroup)} to provide a custom {@link EventLoopGroup}. {@link ChannelFactory} will
+ * be resolved based on the type of {@link EventLoopGroup} provided via
+ * {@link SocketChannelResolver#resolveSocketChannelFactory(EventLoopGroup)}
+ * </li>
+ *
  * <li>Using {@link #create(EventLoopGroup, ChannelFactory)} to provide a custom {@link EventLoopGroup} and
  * {@link ChannelFactory}
+ * </ul>
+ *
  * <p>
- * The {@link EventLoopGroup} <b>MUST</b> be closed by the caller when it is ready to
- * be disposed. The SDK will not close the {@link EventLoopGroup} when the HTTP client is closed. See
- * {@link EventLoopGroup#shutdownGracefully()} to properly close the event loop group.
- * </li>
+ * When configuring the {@link EventLoopGroup} of {@link NettyNioAsyncHttpClient}, if {@link SdkEventLoopGroup.Builder} is
+ * passed to {@link NettyNioAsyncHttpClient.Builder#eventLoopGroupBuilder},
+ * the {@link EventLoopGroup} is managed by the SDK and will be shutdown when the HTTP client is closed. Otherwise,
+ * if an instance of {@link SdkEventLoopGroup} is passed to {@link NettyNioAsyncHttpClient.Builder#eventLoopGroup},
+ * the {@link EventLoopGroup} <b>MUST</b> be closed by the caller when it is ready to be disposed. The SDK will not
+ * close the {@link EventLoopGroup} when the HTTP client is closed. See {@link EventLoopGroup#shutdownGracefully()} to
+ * properly close the event loop group.
  *
- * <li>using {@link #builder()} to provide custom configuration of {@link EventLoopGroup}. The {@link EventLoopGroup} created by
- * the builder is managed by the SDK and will be shutdown when the HTTP client is closed.
- * </li>
- * </ul>
+ * @see NettyNioAsyncHttpClient.Builder#eventLoopGroupBuilder(Builder)
+ * @see NettyNioAsyncHttpClient.Builder#eventLoopGroup(SdkEventLoopGroup)
  */
 @SdkPublicApi
 public final class SdkEventLoopGroup {

http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/ChannelPipelineInitializer.java

@@ -15,38 +15,32 @@
 
 package software.amazon.awssdk.http.nio.netty.internal;
 
-import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.MAX_CONCURRENT_STREAMS;
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.PROTOCOL_FUTURE;
 import static software.amazon.awssdk.utils.StringUtils.lowerCase;
 
 import io.netty.channel.Channel;
-import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInitializer;
 import io.netty.channel.ChannelPipeline;
-import io.netty.channel.SimpleChannelInboundHandler;
 import io.netty.channel.pool.AbstractChannelPoolHandler;
 import io.netty.channel.pool.ChannelPool;
 import io.netty.handler.codec.http.HttpClientCodec;
 import io.netty.handler.codec.http2.ForkedHttp2MultiplexCodecBuilder;
 import io.netty.handler.codec.http2.Http2FrameLogger;
 import io.netty.handler.codec.http2.Http2Settings;
-import io.netty.handler.codec.http2.Http2SettingsFrame;
 import io.netty.handler.logging.LogLevel;
 import io.netty.handler.logging.LoggingHandler;
 import io.netty.handler.ssl.SslContext;
-import java.io.IOException;
-import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicReference;
 import software.amazon.awssdk.annotations.SdkInternalApi;
 import software.amazon.awssdk.http.Protocol;
-import software.amazon.awssdk.http.nio.netty.internal.http2.MultiplexedChannelRecord;
+import software.amazon.awssdk.http.nio.netty.internal.http2.Http2SettingsFrameHandler;
 
 /**
- * Configures the client pipeline to support HTTP/2 frames with multiplexed streams.
+ * ChannelPoolHandler to configure the client pipeline.
  */
 @SdkInternalApi
-public class ChannelPipelineInitializer extends AbstractChannelPoolHandler {
+public final class ChannelPipelineInitializer extends AbstractChannelPoolHandler {
     private final Protocol protocol;
     private final SslContext sslCtx;
     private final long clientMaxStreams;
@@ -103,46 +97,7 @@ private void configureHttp2(Channel ch, ChannelPipeline pipeline) {
 
         pipeline.addLast(codecBuilder.build());
 
-        pipeline.addLast(new SimpleChannelInboundHandler<Http2SettingsFrame>() {
-            @Override
-            protected void channelRead0(ChannelHandlerContext ctx, Http2SettingsFrame msg) {
-                Long serverMaxStreams = Optional.ofNullable(msg.settings().maxConcurrentStreams()).orElse(Long.MAX_VALUE);
-                ch.attr(MAX_CONCURRENT_STREAMS).set(Math.min(clientMaxStreams, serverMaxStreams));
-                ch.attr(PROTOCOL_FUTURE).get().complete(Protocol.HTTP2);
-            }
-
-            @Override
-            public void channelUnregistered(ChannelHandlerContext ctx) {
-                if (!ch.attr(PROTOCOL_FUTURE).get().isDone()) {
-                    channelError(new IOException("The channel was closed before the protocol could be determined."), ch);
-                }
-            }
-
-            @Override
-            public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
-                channelError(cause, ch);
-            }
-        });
-    }
-
-    private void channelError(Throwable cause, Channel ch) {
-        ch.attr(PROTOCOL_FUTURE).get().completeExceptionally(cause);
-        MultiplexedChannelRecord record = ch.attr(ChannelAttributeKey.CHANNEL_POOL_RECORD).get();
-        // Deliver the exception to any child channels registered to this connection.
-        if (record != null) {
-            record.shutdownChildChannels(cause);
-        }
-        // Channel status may still be active at this point even if it's not so queue up the close so that status is
-        // accurately updated
-        ch.eventLoop().submit(() -> {
-            try {
-                if (ch.isActive()) {
-                    ch.close();
-                }
-            } finally {
-                channelPoolRef.get().release(ch);
-            }
-        });
+        pipeline.addLast(new Http2SettingsFrameHandler(ch, clientMaxStreams, channelPoolRef));
     }
 
     private void configureHttp11(Channel ch, ChannelPipeline pipeline) {

http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/FutureCancelHandler.java

@@ -30,7 +30,7 @@
 @SdkInternalApi
 public class FutureCancelHandler extends SimpleChannelInboundHandler {
     @Override
-    protected void channelRead0(ChannelHandlerContext ctx, Object o) throws Exception {
+    protected void channelRead0(ChannelHandlerContext ctx, Object o) {
         ReferenceCountUtil.retain(o);
         ctx.fireChannelRead(o);
     }

http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/http2/Http2SettingsFrameHandler.java

@@ -0,0 +1,87 @@
+/*
+ * Copyright 2010-2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package software.amazon.awssdk.http.nio.netty.internal.http2;
+
+import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.MAX_CONCURRENT_STREAMS;
+import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.PROTOCOL_FUTURE;
+
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.channel.pool.ChannelPool;
+import io.netty.handler.codec.http2.Http2SettingsFrame;
+import java.io.IOException;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
+import software.amazon.awssdk.annotations.SdkInternalApi;
+import software.amazon.awssdk.http.Protocol;
+import software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey;
+
+/**
+ * Configure channel based on the {@link Http2SettingsFrame} received from server
+ */
+@SdkInternalApi
+public final class Http2SettingsFrameHandler extends SimpleChannelInboundHandler<Http2SettingsFrame> {
+
+    private Channel channel;
+    private final long clientMaxStreams;
+    private AtomicReference<ChannelPool> channelPoolRef;
+
+    public Http2SettingsFrameHandler(Channel channel, long clientMaxStreams, AtomicReference<ChannelPool> channelPoolRef) {
+        this.channel = channel;
+        this.clientMaxStreams = clientMaxStreams;
+        this.channelPoolRef = channelPoolRef;
+    }
+
+    @Override
+    protected void channelRead0(ChannelHandlerContext ctx, Http2SettingsFrame msg) {
+        Long serverMaxStreams = Optional.ofNullable(msg.settings().maxConcurrentStreams()).orElse(Long.MAX_VALUE);
+        channel.attr(MAX_CONCURRENT_STREAMS).set(Math.min(clientMaxStreams, serverMaxStreams));
+        channel.attr(PROTOCOL_FUTURE).get().complete(Protocol.HTTP2);
+    }
+
+    @Override
+    public void channelUnregistered(ChannelHandlerContext ctx) {
+        if (!channel.attr(PROTOCOL_FUTURE).get().isDone()) {
+            channelError(new IOException("The channel was closed before the protocol could be determined."), channel);
+        }
+    }
+
+    @Override
+    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+        channelError(cause, channel);
+    }
+
+    private void channelError(Throwable cause, Channel ch) {
+        ch.attr(PROTOCOL_FUTURE).get().completeExceptionally(cause);
+        MultiplexedChannelRecord record = ch.attr(ChannelAttributeKey.CHANNEL_POOL_RECORD).get();
+        // Deliver the exception to any child channels registered to this connection.
+        if (record != null) {
+            record.shutdownChildChannels(cause);
+        }
+        // Channel status may still be active at this point even if it's not so queue up the close so that status is
+        // accurately updated
+        ch.eventLoop().submit(() -> {
+            try {
+                if (ch.isActive()) {
+                    ch.close();
+                }
+            } finally {
+                channelPoolRef.get().release(ch);
+            }
+        });
+    }
+}

http-clients/netty-nio-client/src/test/java/software/amazon/awssdk/http/nio/netty/internal/MockChannel.java

@@ -17,7 +17,7 @@
 
 import io.netty.channel.embedded.EmbeddedChannel;
 
-class MockChannel extends EmbeddedChannel {
+public class MockChannel extends EmbeddedChannel {
     public MockChannel() throws Exception {
         super.doRegister();
     }