Add TransportConfig to control low level transport settings (#3041)

Motivation:

Users need a way to change low-level Netty settings, like
`AdaptiveRecvByteBufAllocator` parameters. In the future, we can use
this interface for more similar settings without the need to change
client/server builder.

Modifications:

- Add `TransportConfig` and `TransportConfigBuilder` in `transport-api`
module;
- Add methods on `SingleAddressHttpClientBuilder` and
`HttpServerBuilder` to let users pass `TransportConfig`;
- Wire it up to propagate to `Tcp[Client|Server]ChannelInitializer` as
`TransportConfigInitializer`;

Result:

Users can control `AdaptiveRecvByteBufAllocator` settings.

servicetalk-http-api/src/main/java/io/servicetalk/http/api/DelegatingHttpServerBuilder.java

@@ -24,6 +24,7 @@
 import io.servicetalk.transport.api.IoExecutor;
 import io.servicetalk.transport.api.LateConnectionAcceptor;
 import io.servicetalk.transport.api.ServerSslConfig;
+import io.servicetalk.transport.api.TransportConfig;
 import io.servicetalk.transport.api.TransportObserver;
 
 import java.net.SocketOption;
@@ -95,6 +96,12 @@ public HttpServerBuilder sslConfig(final ServerSslConfig config, final boolean a
         return this;
     }
 
+    @Override
+    public HttpServerBuilder transportConfig(final TransportConfig transportConfig) {
+        delegate = delegate.transportConfig(transportConfig);
+        return this;
+    }
+
     @Override
     public <T> HttpServerBuilder socketOption(final SocketOption<T> option, final T value) {
         delegate = delegate.socketOption(option, value);

servicetalk-http-api/src/main/java/io/servicetalk/http/api/DelegatingSingleAddressHttpClientBuilder.java

@@ -25,6 +25,7 @@
 import io.servicetalk.logging.api.LogLevel;
 import io.servicetalk.transport.api.ClientSslConfig;
 import io.servicetalk.transport.api.IoExecutor;
+import io.servicetalk.transport.api.TransportConfig;
 
 import java.net.SocketOption;
 import java.util.function.BooleanSupplier;
@@ -221,6 +222,12 @@ public SingleAddressHttpClientBuilder<U, R> inferSniHostname(final boolean shoul
         return this;
     }
 
+    @Override
+    public SingleAddressHttpClientBuilder<U, R> transportConfig(final TransportConfig transportConfig) {
+        delegate = delegate.transportConfig(transportConfig);
+        return this;
+    }
+
     @Override
     public HttpClient build() {
         return delegate.build();

servicetalk-http-api/src/main/java/io/servicetalk/http/api/HttpServerBuilder.java

@@ -31,6 +31,8 @@
 import io.servicetalk.transport.api.LateConnectionAcceptor;
 import io.servicetalk.transport.api.ServerSslConfig;
 import io.servicetalk.transport.api.ServiceTalkSocketOptions;
+import io.servicetalk.transport.api.TransportConfig;
+import io.servicetalk.transport.api.TransportConfigBuilder;
 import io.servicetalk.transport.api.TransportObserver;
 
 import java.net.SocketOption;
@@ -112,13 +114,26 @@ default HttpServerBuilder sslConfig(ServerSslConfig defaultConfig, Map<String, S
                 "sslConfig(ServerSslConfig, Map, int, Duration) is not supported by " + getClass());
     }
 
+    /**
+     * Set the transport configuration.
+     *
+     * @param transportConfig {@link TransportConfig} to use.
+     * @return {@code this}.
+     * @see TransportConfigBuilder
+     */
+    // FIXME: 0.43 - consider removing default impl
+    default HttpServerBuilder transportConfig(TransportConfig transportConfig) {
+        throw new UnsupportedOperationException("Setting transport config is not yet supported by " +
+                getClass().getName());
+    }
+
     /**
      * Adds a {@link SocketOption} that is applied to connected/accepted socket channels.
      *
      * @param <T> the type of the value.
      * @param option the option to apply.
      * @param value the value.
-     * @return this.
+     * @return {@code this}.
      * @see StandardSocketOptions
      * @see ServiceTalkSocketOptions
      */
@@ -129,7 +144,7 @@ default HttpServerBuilder sslConfig(ServerSslConfig defaultConfig, Map<String, S
      * @param <T> the type of the value.
      * @param option the option to apply.
      * @param value the value.
-     * @return this.
+     * @return {@code this}.
      * @see StandardSocketOptions
      * @see ServiceTalkSocketOptions
      */

servicetalk-http-api/src/main/java/io/servicetalk/http/api/ProxyConfig.java

@@ -23,6 +23,7 @@
  * Configuration for a proxy.
  *
  * @param <A> the type of address
+ * @see ProxyConfigBuilder
  */
 public interface ProxyConfig<A> {
 

servicetalk-http-api/src/main/java/io/servicetalk/http/api/RedirectConfig.java

@@ -22,6 +22,8 @@
 
 /**
  * Configuration options for <a href="https://datatracker.ietf.org/doc/html/rfc7231#section-6.4">redirection</a>.
+ *
+ * @see RedirectConfigBuilder
  */
 public interface RedirectConfig {
 

servicetalk-http-api/src/main/java/io/servicetalk/http/api/SingleAddressHttpClientBuilder.java

@@ -29,6 +29,8 @@
 import io.servicetalk.transport.api.ClientSslConfig;
 import io.servicetalk.transport.api.IoExecutor;
 import io.servicetalk.transport.api.ServiceTalkSocketOptions;
+import io.servicetalk.transport.api.TransportConfig;
+import io.servicetalk.transport.api.TransportConfigBuilder;
 import io.servicetalk.transport.api.TransportObserver;
 
 import java.net.SocketOption;
@@ -401,4 +403,17 @@ SingleAddressHttpClientBuilder<U, R> retryServiceDiscoveryErrors(
      * @return {@code this}
      */
     SingleAddressHttpClientBuilder<U, R> inferSniHostname(boolean shouldInfer);
+
+    /**
+     * Set the transport configuration.
+     *
+     * @param transportConfig {@link TransportConfig} to use
+     * @return {@code this}
+     * @see TransportConfigBuilder
+     */
+    // FIXME: 0.43 - consider removing default impl
+    default SingleAddressHttpClientBuilder<U, R> transportConfig(TransportConfig transportConfig) {
+        throw new UnsupportedOperationException("Setting transport config is not yet supported by " +
+                getClass().getName());
+    }
 }

servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/DefaultHttpServerBuilder.java

@@ -49,6 +49,7 @@
 import io.servicetalk.transport.api.IoExecutor;
 import io.servicetalk.transport.api.LateConnectionAcceptor;
 import io.servicetalk.transport.api.ServerSslConfig;
+import io.servicetalk.transport.api.TransportConfig;
 import io.servicetalk.transport.api.TransportObserver;
 import io.servicetalk.transport.netty.internal.InfluencerConnectionAcceptor;
 
@@ -230,6 +231,12 @@ public HttpServerBuilder sslConfig(final ServerSslConfig config, final boolean a
         return this;
     }
 
+    @Override
+    public HttpServerBuilder transportConfig(final TransportConfig transportConfig) {
+        config.tcpConfig().transportConfig(transportConfig);
+        return this;
+    }
+
     @Override
     public <T> HttpServerBuilder socketOption(final SocketOption<T> option, final T value) {
         config.tcpConfig().socketOption(option, value);

servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/DefaultSingleAddressHttpClientBuilder.java

@@ -60,6 +60,7 @@
 import io.servicetalk.transport.api.ExecutionStrategy;
 import io.servicetalk.transport.api.HostAndPort;
 import io.servicetalk.transport.api.IoExecutor;
+import io.servicetalk.transport.api.TransportConfig;
 
 import io.netty.handler.ssl.SslContext;
 import org.slf4j.Logger;
@@ -647,6 +648,12 @@ public DefaultSingleAddressHttpClientBuilder<U, R> inferSniHostname(boolean shou
         return this;
     }
 
+    @Override
+    public DefaultSingleAddressHttpClientBuilder<U, R> transportConfig(final TransportConfig transportConfig) {
+        config.tcpConfig().transportConfig(transportConfig);
+        return this;
+    }
+
     private static <U> CharSequence toAuthorityForm(final U address) {
         if (address instanceof CharSequence) {
             return (CharSequence) address;

servicetalk-tcp-netty-internal/src/main/java/io/servicetalk/tcp/netty/internal/AbstractReadOnlyTcpConfig.java

@@ -18,6 +18,7 @@
 import io.servicetalk.logging.api.UserDataLoggerConfig;
 import io.servicetalk.transport.api.ServiceTalkSocketOptions;
 import io.servicetalk.transport.api.SslConfig;
+import io.servicetalk.transport.api.TransportConfig;
 import io.servicetalk.transport.netty.internal.FlushStrategy;
 
 import io.netty.channel.ChannelOption;
@@ -40,19 +41,22 @@ abstract class AbstractReadOnlyTcpConfig {
     private final FlushStrategy flushStrategy;
     @Nullable
     private final UserDataLoggerConfig wireLoggerConfig;
+    private final TransportConfig transportConfig;
 
     protected AbstractReadOnlyTcpConfig(final AbstractTcpConfig from) {
         options = nonNullOptions(from.options());
         idleTimeoutMs = from.idleTimeoutMs();
         flushStrategy = from.flushStrategy();
         wireLoggerConfig = from.wireLoggerConfig();
+        transportConfig = from.transportConfig();
     }
 
     AbstractReadOnlyTcpConfig(final AbstractReadOnlyTcpConfig from) {
         options = from.options();
         idleTimeoutMs = from.idleTimeoutMs();
         flushStrategy = from.flushStrategy();
         wireLoggerConfig = from.wireLoggerConfig();
+        transportConfig = from.transportConfig();
     }
 
     @SuppressWarnings("rawtypes")
@@ -113,4 +117,13 @@ public final UserDataLoggerConfig wireLoggerConfig() {
      */
     @Nullable
     public abstract SslConfig sslConfig();
+
+    /**
+     * Get the {@link TransportConfig}.
+     *
+     * @return {@link TransportConfig} to use
+     */
+    public final TransportConfig transportConfig() {
+        return transportConfig;
+    }
 }

servicetalk-tcp-netty-internal/src/main/java/io/servicetalk/tcp/netty/internal/AbstractTcpConfig.java

@@ -19,6 +19,8 @@
 import io.servicetalk.logging.api.UserDataLoggerConfig;
 import io.servicetalk.logging.slf4j.internal.DefaultUserDataLoggerConfig;
 import io.servicetalk.transport.api.ServiceTalkSocketOptions;
+import io.servicetalk.transport.api.TransportConfig;
+import io.servicetalk.transport.api.TransportConfigBuilder;
 import io.servicetalk.transport.netty.internal.FlushStrategy;
 
 import io.netty.channel.ChannelOption;
@@ -41,13 +43,16 @@
  */
 abstract class AbstractTcpConfig {
 
+    private static final TransportConfig DEFAULT_TRANSPORT_CONFIG = new TransportConfigBuilder().build();
+
     @Nullable
     @SuppressWarnings("rawtypes")
     private Map<ChannelOption, Object> options;
     private long idleTimeoutMs;
     private FlushStrategy flushStrategy = defaultFlushStrategy();
     @Nullable
     private UserDataLoggerConfig wireLoggerConfig;
+    private TransportConfig transportConfig = DEFAULT_TRANSPORT_CONFIG;
 
     protected AbstractTcpConfig() {
         socketOption(SO_KEEPALIVE, true);
@@ -58,6 +63,7 @@ protected AbstractTcpConfig(final AbstractTcpConfig from) {
         idleTimeoutMs = from.idleTimeoutMs;
         flushStrategy = from.flushStrategy;
         wireLoggerConfig = from.wireLoggerConfig;
+        transportConfig = from.transportConfig;
     }
 
     @Nullable
@@ -79,6 +85,10 @@ final UserDataLoggerConfig wireLoggerConfig() {
         return wireLoggerConfig;
     }
 
+    final TransportConfig transportConfig() {
+        return transportConfig;
+    }
+
     /**
      * Add a {@link SocketOption} that is applied.
      *
@@ -125,4 +135,13 @@ public final void enableWireLogging(final String loggerName,
                                         final BooleanSupplier logUserData) {
         wireLoggerConfig = new DefaultUserDataLoggerConfig(loggerName, logLevel, logUserData);
     }
+
+    /**
+     * Sets the transport configuration.
+     *
+     * @param transportConfig {@link TransportConfig} to use
+     */
+    public final void transportConfig(final TransportConfig transportConfig) {
+        this.transportConfig = requireNonNull(transportConfig);
+    }
 }

servicetalk-tcp-netty-internal/src/main/java/io/servicetalk/tcp/netty/internal/TcpClientChannelInitializer.java

@@ -86,6 +86,8 @@ public TcpClientChannelInitializer(final ReadOnlyTcpClientConfig config,
                                        final boolean deferSslHandler) {
         ChannelInitializer delegate = ChannelInitializer.defaultInitializer();
 
+        delegate = delegate.andThen(new TransportConfigInitializer(config.transportConfig()));
+
         final ClientSslConfig sslConfig = config.sslConfig();
         if (observer != NoopConnectionObserver.INSTANCE) {
             delegate = delegate.andThen(new ConnectionObserverInitializer(observer,

servicetalk-tcp-netty-internal/src/main/java/io/servicetalk/tcp/netty/internal/TcpServerChannelInitializer.java

@@ -68,6 +68,8 @@ public TcpServerChannelInitializer(final ReadOnlyTcpServerConfig config,
                                        final ExecutionContext<?> executionContext) {
         ChannelInitializer delegate = ChannelInitializer.defaultInitializer();
 
+        delegate = delegate.andThen(new TransportConfigInitializer(config.transportConfig()));
+
         if (observer != NoopConnectionObserver.INSTANCE) {
             delegate = delegate.andThen(new ConnectionObserverInitializer(observer,
                     channel -> new TcpConnectionInfo(channel,

servicetalk-tcp-netty-internal/src/main/java/io/servicetalk/tcp/netty/internal/TransportConfigInitializer.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright © 2024 Apple Inc. and the ServiceTalk project authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 io.servicetalk.tcp.netty.internal;
+
+import io.servicetalk.transport.api.TransportConfig;
+import io.servicetalk.transport.netty.internal.ChannelInitializer;
+
+import io.netty.channel.AdaptiveRecvByteBufAllocator;
+import io.netty.channel.Channel;
+
+import static java.lang.Math.min;
+
+final class TransportConfigInitializer implements ChannelInitializer {
+
+    private final TransportConfig config;
+
+    TransportConfigInitializer(final TransportConfig config) {
+        this.config = config;
+    }
+
+    @Override
+    public void init(final Channel channel) {
+        channel.config().setRecvByteBufAllocator(new AdaptiveRecvByteBufAllocator(
+                min(512, config.maxBytesPerRead()), min(32_768, config.maxBytesPerRead()), config.maxBytesPerRead())
+                .respectMaybeMoreData(false)
+                .maxMessagesPerRead(config.maxReadAttemptsPerSelect()));
+    }
+}

servicetalk-transport-api/src/main/java/io/servicetalk/transport/api/ClientSslConfig.java

@@ -22,6 +22,7 @@
 
 /**
  * Specifies the configuration for client side TLS/SSL.
+ *
  * @see ClientSslConfigBuilder
  */
 public interface ClientSslConfig extends SslConfig {

servicetalk-transport-api/src/main/java/io/servicetalk/transport/api/ServerSslConfig.java

@@ -19,6 +19,7 @@
 
 /**
  * Specifies the configuration for server side TLS/SSL.
+ *
  * @see ServerSslConfigBuilder
  */
 public interface ServerSslConfig extends SslConfig {