Reactor-Netty HttpClient changes

sdk/cosmos/microsoft-azure-cosmos-benchmark/src/main/java/com/azure/data/cosmos/benchmark/Configuration.java

@@ -55,7 +55,7 @@ class Configuration {
     private int documentDataFieldSize = 20;
 
     @Parameter(names = "-maxConnectionPoolSize", description = "Max Connection Pool Size")
-    private Integer maxConnectionPoolSize = 1000;
+    private Integer maxConnectionPoolSize = 500;
 
     @Parameter(names = "-consistencyLevel", description = "Consistency Level", converter = ConsistencyLevelConverter.class)
     private ConsistencyLevel consistencyLevel = ConsistencyLevel.SESSION;

sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/ConnectionPolicy.java

@@ -18,7 +18,7 @@ public final class ConnectionPolicy {
     private static final int DEFAULT_MEDIA_REQUEST_TIMEOUT_IN_MILLIS = 300 * 1000;
     private static final int DEFAULT_IDLE_CONNECTION_TIMEOUT_IN_MILLIS = 60 * 1000;
 
-    private static final int DEFAULT_MAX_POOL_SIZE = 1000;
+    private static final int DEFAULT_MAX_POOL_SIZE = 500;
 
     private static ConnectionPolicy default_policy = null;
     private int requestTimeoutInMillis;
@@ -38,7 +38,7 @@ public final class ConnectionPolicy {
      * Constructor.
      */
     public ConnectionPolicy() {
-        this.connectionMode = ConnectionMode.GATEWAY;
+        this.connectionMode = ConnectionMode.DIRECT;
         this.enableReadRequestsFallback = null;
         this.idleConnectionTimeoutInMillis = DEFAULT_IDLE_CONNECTION_TIMEOUT_IN_MILLIS;
         this.maxPoolSize = DEFAULT_MAX_POOL_SIZE;
@@ -62,7 +62,7 @@ public static ConnectionPolicy defaultPolicy() {
 
     /**
      * Gets the request timeout (time to wait for response from network peer) in
-     * milliseconds. 
+     * milliseconds.
      *
      * @return the request timeout in milliseconds.
      */

sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/RxGatewayStoreModel.java

@@ -24,7 +24,6 @@
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
-import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.net.URI;
@@ -43,7 +42,6 @@
  */
 class RxGatewayStoreModel implements RxStoreModel {
 
-    private final static int INITIAL_RESPONSE_BUFFER_SIZE = 1024;
     private final Logger logger = LoggerFactory.getLogger(RxGatewayStoreModel.class);
     private final Map<String, String> defaultHeaders;
     private final HttpClient httpClient;
@@ -229,22 +227,6 @@ private String ensureSlashPrefixed(String path) {
         return "/" + path;
     }
 
-    private Mono<String> toString(Flux<ByteBuf> contentObservable) {
-        return contentObservable
-            .reduce(
-                new ByteArrayOutputStream(INITIAL_RESPONSE_BUFFER_SIZE),
-                (out, bb) -> {
-                    try {
-                        bb.readBytes(out, bb.readableBytes());
-                        return out;
-                    }
-                    catch (IOException e) {
-                        throw new RuntimeException(e);
-                    }
-                })
-            .map(out -> new String(out.toByteArray(), StandardCharsets.UTF_8));
-    }
-
     /**
      * Transforms the reactor netty's client response Observable to RxDocumentServiceResponse Observable.
      *
@@ -272,11 +254,7 @@ private Flux<RxDocumentServiceResponse> toDocumentServiceResponse(Mono<HttpRespo
                     // for delete we don't expect any body
                     inputStreamObservable = Flux.just(IOUtils.toInputStream("", StandardCharsets.UTF_8));
                 } else {
-                    // transforms the ByteBufFlux to Flux<InputStream>
-                    inputStreamObservable = httpResponse
-                            .body()
-                            .flatMap(byteBuf ->
-                                    Flux.just(IOUtils.toInputStream(byteBuf.toString(StandardCharsets.UTF_8), StandardCharsets.UTF_8)));
+                    inputStreamObservable = httpResponse.bodyAsInputStream();
                 }
 
                 return inputStreamObservable
@@ -315,7 +293,7 @@ private Flux<RxDocumentServiceResponse> toDocumentServiceResponse(Mono<HttpRespo
                     contentObservable = Flux.just(StringUtils.EMPTY);
                 } else {
                     // transforms the ByteBufFlux to Flux<String>
-                    contentObservable = toString(httpResponse.body()).flux();
+                    contentObservable = httpResponse.bodyAsString().flux();
                 }
 
                 return contentObservable
@@ -498,4 +476,4 @@ private void applySessionToken(RxDocumentServiceRequest request) {
             headers.put(HttpConstants.HttpHeaders.SESSION_TOKEN, sessionToken);
         }
     }
-}
+}

sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/ErrorUtils.java

@@ -17,7 +17,7 @@ public class ErrorUtils {
     private static final Logger logger = LoggerFactory.getLogger(ErrorUtils.class);
 
     static Mono<String> getErrorResponseAsync(HttpResponse responseMessage, HttpRequest request) {
-        Mono<String> responseAsString = ResponseUtils.toString(responseMessage.body());
+        Mono<String> responseAsString = responseMessage.bodyAsString();
         if (request.httpMethod() == HttpMethod.DELETE) {
             return Mono.just(StringUtils.EMPTY);
         }

sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/HttpClientUtils.java

@@ -31,7 +31,7 @@ static Mono<RxDocumentServiceResponse> parseResponseAsync(Mono<HttpResponse> htt
     }
 
     private static Mono<CosmosClientException> createDocumentClientException(HttpResponse httpResponse) {
-        Mono<String> readStream = ResponseUtils.toString(httpResponse.body());
+        Mono<String> readStream = httpResponse.bodyAsString();
 
         return readStream.map(body -> {
             CosmosError cosmosError = BridgeInternal.createCosmosError(body);

sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/ResponseUtils.java

@@ -6,33 +6,11 @@
 import com.azure.data.cosmos.internal.http.HttpHeaders;
 import com.azure.data.cosmos.internal.http.HttpRequest;
 import com.azure.data.cosmos.internal.http.HttpResponse;
-import io.netty.buffer.ByteBuf;
 import io.netty.handler.codec.http.HttpMethod;
 import org.apache.commons.lang3.StringUtils;
-import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
-import java.nio.charset.StandardCharsets;
-
 class ResponseUtils {
-    private final static int INITIAL_RESPONSE_BUFFER_SIZE = 1024;
-
-    public static Mono<String> toString(Flux<ByteBuf> contentObservable) {
-        return contentObservable
-                .reduce(
-                        new ByteArrayOutputStream(INITIAL_RESPONSE_BUFFER_SIZE),
-                        (out, bb) -> {
-                            try {
-                                bb.readBytes(out, bb.readableBytes());
-                                return out;
-                            } catch (IOException e) {
-                                throw new RuntimeException(e);
-                            }
-                        })
-                .map(out -> new String(out.toByteArray(), StandardCharsets.UTF_8));
-    }
 
     static Mono<StoreResponse> toStoreResponse(HttpResponse httpClientResponse, HttpRequest httpRequest) {
 
@@ -44,7 +22,7 @@ static Mono<StoreResponse> toStoreResponse(HttpResponse httpClientResponse, Http
             // for delete we don't expect any body
             contentObservable = Mono.just(StringUtils.EMPTY);
         } else {
-            contentObservable = toString(httpClientResponse.body());
+            contentObservable = httpClientResponse.bodyAsString();
         }
 
         return contentObservable.flatMap(content -> {

sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/http/BufferedHttpResponse.java

@@ -7,6 +7,8 @@
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
 import java.nio.charset.Charset;
 import java.nio.charset.StandardCharsets;
 
@@ -53,6 +55,11 @@ public Flux<ByteBuf> body() {
         return bodyAsByteArray().flatMapMany(bytes -> Flux.just(Unpooled.wrappedBuffer(bytes)));
     }
 
+    @Override
+    public Flux<InputStream> bodyAsInputStream() {
+        return bodyAsByteArray().flatMapMany(bytes -> Flux.just(new ByteArrayInputStream(bytes)));
+    }
+
     @Override
     public Mono<String> bodyAsString() {
         return bodyAsByteArray()

sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/http/HttpClient.java

@@ -5,6 +5,8 @@
 import reactor.core.publisher.Mono;
 import reactor.netty.resources.ConnectionProvider;
 
+import java.time.Duration;
+
 /**
  * A generic interface for sending HTTP requests and getting responses.
  */
@@ -19,7 +21,7 @@ public interface HttpClient {
     Mono<HttpResponse> send(HttpRequest request);
 
     /**
-     * Create fixed HttpClient with {@link HttpClientConfig}
+     * Create HttpClient with FixedChannelPool {@link HttpClientConfig}
      *
      * @return the HttpClient
      */
@@ -28,12 +30,40 @@ static HttpClient createFixed(HttpClientConfig httpClientConfig) {
             throw new IllegalArgumentException("HttpClientConfig is null");
         }
 
-        if (httpClientConfig.getMaxPoolSize() == null) {
-            return new ReactorNettyClient(ConnectionProvider.fixed(httpClientConfig.getConfigs().getReactorNettyConnectionPoolName()), httpClientConfig);
+        Integer maxIdleConnectionTimeoutInMillis = 60 * 1000;
+        if (httpClientConfig.getMaxIdleConnectionTimeoutInMillis() != null) {
+            maxIdleConnectionTimeoutInMillis = httpClientConfig.getMaxIdleConnectionTimeoutInMillis();
+        }
+
+        //  Default pool size
+        Integer maxPoolSize = 500;
+        if (httpClientConfig.getMaxPoolSize() != null) {
+            maxPoolSize = httpClientConfig.getMaxPoolSize();
         }
-        return new ReactorNettyClient(ConnectionProvider.fixed(httpClientConfig.getConfigs().getReactorNettyConnectionPoolName(), httpClientConfig.getMaxPoolSize()), httpClientConfig);
+
+        int connectionAcquireTimeoutInMillis = 45 * 1000;
+
+        ConnectionProvider fixedConnectionProvider =
+            ConnectionProvider.fixed(httpClientConfig.getConfigs().getReactorNettyConnectionPoolName(),
+            maxPoolSize, connectionAcquireTimeoutInMillis, Duration.ofMillis(maxIdleConnectionTimeoutInMillis));
+
+        return ReactorNettyClient.createWithConnectionProvider(fixedConnectionProvider, httpClientConfig);
     }
 
+    /**
+     * Create HttpClient with un-pooled connection {@link HttpClientConfig}
+     *
+     * @return the HttpClient
+     */
+    static HttpClient create(HttpClientConfig httpClientConfig) {
+        if (httpClientConfig.getConfigs() == null) {
+            throw new IllegalArgumentException("HttpClientConfig is null");
+        }
+
+        return ReactorNettyClient.create(httpClientConfig);
+    }
+
+
     /**
      * Shutdown the Http Client and clean up resources
      */

sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/http/HttpClientConfig.java

@@ -18,6 +18,7 @@ public class HttpClientConfig {
     private Integer maxIdleConnectionTimeoutInMillis;
     private Integer requestTimeoutInMillis;
     private InetSocketAddress proxy;
+    private boolean connectionKeepAlive = true;
 
     public HttpClientConfig(Configs configs) {
         this.configs = configs;
@@ -43,6 +44,11 @@ public HttpClientConfig withRequestTimeoutInMillis(int requestTimeoutInMillis) {
         return this;
     }
 
+    public HttpClientConfig withConnectionKeepAlive(boolean connectionKeepAlive) {
+        this.connectionKeepAlive = connectionKeepAlive;
+        return this;
+    }
+
     public Configs getConfigs() {
         return configs;
     }
@@ -62,4 +68,8 @@ public Integer getRequestTimeoutInMillis() {
     public InetSocketAddress getProxy() {
         return proxy;
     }
+
+    public boolean isConnectionKeepAlive() {
+        return connectionKeepAlive;
+    }
 }

sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/http/HttpResponse.java

@@ -8,6 +8,7 @@
 import reactor.core.publisher.Mono;
 import reactor.netty.Connection;
 
+import java.io.InputStream;
 import java.nio.charset.Charset;
 
 /**
@@ -68,6 +69,13 @@ public abstract class HttpResponse implements AutoCloseable {
      */
     public abstract Flux<ByteBuf> body();
 
+    /**
+     * Get the response content as InputStream.
+     *
+     * @return this response content as InputStream
+     */
+    public abstract Flux<InputStream> bodyAsInputStream();
+
     /**
      * Get the response content as a byte[].
      *