MultiAddressClientBuilder execution strategy control

servicetalk-client-api/src/main/java/io/servicetalk/client/api/DefaultClientGroup.java

@@ -123,17 +123,19 @@ public Client get(final Key key) {
         try {
             client = requireNonNull(clientFactory.apply(key), "Newly created client can not be null");
         } catch (Throwable t) {
-            clientMap.remove(key); // PLACEHOLDER_CLIENT
+            final boolean removed = clientMap.remove(key, PLACEHOLDER_CLIENT);
+            assert removed : "Expected to remove PLACEHOLDER_CLIENT";
             throw new IllegalArgumentException("Failed to create new client", t);
         }
 
-        clientMap.put(key, client); // Overwrite PLACEHOLDER_CLIENT
+        final boolean replaced = clientMap.replace(key, PLACEHOLDER_CLIENT, client);
+        assert replaced : "Expected to replace PLACEHOLDER_CLIENT";
         toSource(client.onClose()).subscribe(new RemoveClientOnClose(key, client));
         LOGGER.debug("A new client {} was created", client);
 
         if (closed) {
             // group has been closed after a new client was created
-            if (clientMap.remove(key) != null) { // not closed by closing thread
+            if (clientMap.remove(key, client)) { // not closed by closing thread
                 client.closeAsync().subscribe();
                 LOGGER.debug("Recently created client {} was removed and closed, group {} closed", client, this);
             }

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

@@ -49,7 +49,7 @@ enum DefaultHttpExecutionStrategy implements HttpExecutionStrategy {
     OFFLOAD_SEND_EVENT_STRATEGY(EnumSet.of(OFFLOAD_SEND, OFFLOAD_EVENT)),
     OFFLOAD_RECEIVE_META_AND_SEND_EVENT_STRATEGY(EnumSet.of(OFFLOAD_RECEIVE_META, OFFLOAD_SEND, OFFLOAD_EVENT)),
     OFFLOAD_RECEIVE_DATA_AND_SEND_EVENT_STRATEGY(EnumSet.of(OFFLOAD_RECEIVE_DATA, OFFLOAD_SEND, OFFLOAD_EVENT)),
-    OFFLOADD_ALL_REQRESP_EVENT_STRATEGY(
+    OFFLOAD_ALL_REQRESP_EVENT_STRATEGY(
             EnumSet.of(OFFLOAD_RECEIVE_META, OFFLOAD_RECEIVE_DATA, OFFLOAD_SEND, OFFLOAD_EVENT)),
 
     OFFLOAD_CLOSE_STRATEGY(EnumSet.of(OFFLOAD_CLOSE)),

servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/FilterableClientToClient.java → servicetalk-http-api/src/main/java/io/servicetalk/http/api/FilterableClientToClient.java

@@ -13,88 +13,83 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.servicetalk.http.netty;
+package io.servicetalk.http.api;
 
 import io.servicetalk.concurrent.api.Completable;
 import io.servicetalk.concurrent.api.Publisher;
 import io.servicetalk.concurrent.api.Single;
-import io.servicetalk.http.api.BlockingHttpClient;
-import io.servicetalk.http.api.BlockingStreamingHttpClient;
-import io.servicetalk.http.api.FilterableStreamingHttpClient;
-import io.servicetalk.http.api.HttpClient;
-import io.servicetalk.http.api.HttpConnectionContext;
-import io.servicetalk.http.api.HttpEventKey;
-import io.servicetalk.http.api.HttpExecutionContext;
-import io.servicetalk.http.api.HttpExecutionStrategy;
-import io.servicetalk.http.api.HttpRequestMetaData;
-import io.servicetalk.http.api.HttpRequestMethod;
-import io.servicetalk.http.api.ReservedBlockingHttpConnection;
-import io.servicetalk.http.api.ReservedBlockingStreamingHttpConnection;
-import io.servicetalk.http.api.ReservedHttpConnection;
-import io.servicetalk.http.api.ReservedStreamingHttpConnection;
-import io.servicetalk.http.api.StreamingHttpClient;
-import io.servicetalk.http.api.StreamingHttpRequest;
-import io.servicetalk.http.api.StreamingHttpResponse;
-import io.servicetalk.http.api.StreamingHttpResponseFactory;
 
 import static io.servicetalk.http.api.HttpApiConversions.toBlockingClient;
 import static io.servicetalk.http.api.HttpApiConversions.toBlockingStreamingClient;
 import static io.servicetalk.http.api.HttpApiConversions.toClient;
 import static io.servicetalk.http.api.HttpApiConversions.toReservedBlockingConnection;
 import static io.servicetalk.http.api.HttpApiConversions.toReservedBlockingStreamingConnection;
 import static io.servicetalk.http.api.HttpApiConversions.toReservedConnection;
+import static io.servicetalk.http.api.HttpContextKeys.HTTP_CLIENT_API_KEY;
 import static io.servicetalk.http.api.HttpContextKeys.HTTP_EXECUTION_STRATEGY_KEY;
 
 final class FilterableClientToClient implements StreamingHttpClient {
     private final FilterableStreamingHttpClient client;
-    private final HttpExecutionStrategy strategy;
+    private final DelegatingHttpExecutionContext executionContext;
 
     FilterableClientToClient(FilterableStreamingHttpClient filteredClient, HttpExecutionStrategy strategy) {
         client = filteredClient;
-        this.strategy = strategy;
+        this.executionContext = new DelegatingHttpExecutionContext(client.executionContext()) {
+            @Override
+            public HttpExecutionStrategy executionStrategy() {
+                return strategy;
+            }
+        };
     }
 
     @Override
     public HttpClient asClient() {
-        return toClient(this, strategy);
+        return toClient(this, executionContext.executionStrategy());
     }
 
     @Override
     public BlockingStreamingHttpClient asBlockingStreamingClient() {
-        return toBlockingStreamingClient(this, strategy);
+        return toBlockingStreamingClient(this, executionContext.executionStrategy());
     }
 
     @Override
     public BlockingHttpClient asBlockingClient() {
-        return toBlockingClient(this, strategy);
+        return toBlockingClient(this, executionContext.executionStrategy());
     }
 
     @Override
     public Single<StreamingHttpResponse> request(final StreamingHttpRequest request) {
         return Single.defer(() -> {
-            request.context().putIfAbsent(HTTP_EXECUTION_STRATEGY_KEY, strategy);
+            if (null == request.context().putIfAbsent(HTTP_EXECUTION_STRATEGY_KEY,
+                    executionContext().executionStrategy())) {
+                request.context().put(HTTP_CLIENT_API_KEY, HttpApiConversions.ClientAPI.ASYNC_STREAMING);
+            }
             return client.request(request).shareContextOnSubscribe();
         });
     }
 
     @Override
     public Single<ReservedStreamingHttpConnection> reserveConnection(final HttpRequestMetaData metaData) {
         return Single.defer(() -> {
-            metaData.context().putIfAbsent(HTTP_EXECUTION_STRATEGY_KEY, strategy);
+            if (null == metaData.context().putIfAbsent(HTTP_EXECUTION_STRATEGY_KEY,
+                    executionContext().executionStrategy())) {
+                metaData.context().put(HTTP_CLIENT_API_KEY, HttpApiConversions.ClientAPI.ASYNC_STREAMING);
+            }
+
             return client.reserveConnection(metaData).map(rc -> new ReservedStreamingHttpConnection() {
                 @Override
                 public ReservedHttpConnection asConnection() {
-                    return toReservedConnection(this, strategy);
+                    return toReservedConnection(this, executionContext.executionStrategy());
                 }
 
                 @Override
                 public ReservedBlockingStreamingHttpConnection asBlockingStreamingConnection() {
-                    return toReservedBlockingStreamingConnection(this, strategy);
+                    return toReservedBlockingStreamingConnection(this, executionContext.executionStrategy());
                 }
 
                 @Override
                 public ReservedBlockingHttpConnection asBlockingConnection() {
-                    return toReservedBlockingConnection(this, strategy);
+                    return toReservedBlockingConnection(this, executionContext.executionStrategy());
                 }
 
                 @Override
@@ -108,8 +103,10 @@ public Single<StreamingHttpResponse> request(final StreamingHttpRequest request)
                     // created and hence could have an incorrect default strategy. Doing this makes sure we never call
                     // the method without strategy just as we do for the regular connection.
                     return Single.defer(() -> {
-                        request.context().putIfAbsent(HTTP_EXECUTION_STRATEGY_KEY,
-                                FilterableClientToClient.this.strategy);
+                        if (null == request.context().putIfAbsent(HTTP_EXECUTION_STRATEGY_KEY,
+                                FilterableClientToClient.this.executionContext().executionStrategy())) {
+                            request.context().put(HTTP_CLIENT_API_KEY, HttpApiConversions.ClientAPI.ASYNC_STREAMING);
+                        }
                         return rc.request(request).shareContextOnSubscribe();
                     });
                 }
@@ -159,7 +156,7 @@ public StreamingHttpRequest newRequest(final HttpRequestMethod method, final Str
 
     @Override
     public HttpExecutionContext executionContext() {
-        return client.executionContext();
+        return executionContext;
     }
 
     @Override

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

@@ -20,12 +20,35 @@
 import io.servicetalk.http.api.StreamingHttpClientToHttpClient.ReservedStreamingHttpConnectionToReservedHttpConnection;
 
 import static io.servicetalk.http.api.HttpContextKeys.HTTP_EXECUTION_STRATEGY_KEY;
+import static io.servicetalk.http.api.StreamingHttpConnectionToBlockingHttpConnection.DEFAULT_BLOCKING_CONNECTION_STRATEGY;
+import static io.servicetalk.http.api.StreamingHttpConnectionToBlockingStreamingHttpConnection.DEFAULT_BLOCKING_STREAMING_CONNECTION_STRATEGY;
+import static io.servicetalk.http.api.StreamingHttpConnectionToHttpConnection.DEFAULT_ASYNC_CONNECTION_STRATEGY;
 
 /**
  * Conversion routines to {@link StreamingHttpService}.
  */
 public final class HttpApiConversions {
 
+    /**
+     * The "flavors" of client API available.
+     */
+    public enum ClientAPI {
+        BLOCKING_AGGREGATED(DEFAULT_BLOCKING_CONNECTION_STRATEGY),
+        BLOCKING_STREAMING(DEFAULT_BLOCKING_STREAMING_CONNECTION_STRATEGY),
+        ASYNC_AGGREGATED(DEFAULT_ASYNC_CONNECTION_STRATEGY),
+        ASYNC_STREAMING(DefaultHttpExecutionStrategy.OFFLOAD_ALL_REQRESP_EVENT_STRATEGY);
+
+        private final HttpExecutionStrategy defaultApiStrategy;
+
+        ClientAPI(HttpExecutionStrategy defaultApiStrategy) {
+            this.defaultApiStrategy = defaultApiStrategy;
+        }
+
+        public HttpExecutionStrategy defaultStrategy() {
+            return defaultApiStrategy;
+        }
+    }
+
     private HttpApiConversions() {
         // no instances
     }
@@ -39,6 +62,7 @@ private HttpApiConversions() {
      * @return The conversion result.
      * @deprecated Use overload with {@link HttpExecutionStrategy} rather than {@link HttpExecutionStrategyInfluencer}
      */
+    // FIXME: 0.43 - remove deprecated method
     @Deprecated
     public static ReservedHttpConnection toReservedConnection(ReservedStreamingHttpConnection original,
                                                               HttpExecutionStrategyInfluencer influencer) {
@@ -66,6 +90,7 @@ public static ReservedHttpConnection toReservedConnection(ReservedStreamingHttpC
      * @return The conversion result.
      * @deprecated Use overload with {@link HttpExecutionStrategy} rather than {@link HttpExecutionStrategyInfluencer}
      */
+    // FIXME: 0.43 - remove deprecated method
     @Deprecated
     public static ReservedBlockingHttpConnection toReservedBlockingConnection(
             ReservedStreamingHttpConnection original, HttpExecutionStrategyInfluencer influencer) {
@@ -94,6 +119,7 @@ public static ReservedBlockingHttpConnection toReservedBlockingConnection(
      * @return The conversion result.
      * @deprecated Use overload with {@link HttpExecutionStrategy} rather than {@link HttpExecutionStrategyInfluencer}
      */
+    // FIXME: 0.43 - remove deprecated method
     @Deprecated
     public static ReservedBlockingStreamingHttpConnection toReservedBlockingStreamingConnection(
             ReservedStreamingHttpConnection original, HttpExecutionStrategyInfluencer influencer) {
@@ -122,6 +148,7 @@ public static ReservedBlockingStreamingHttpConnection toReservedBlockingStreamin
      * @return The conversion result.
      * @deprecated Use overload with {@link HttpExecutionStrategy} rather than {@link HttpExecutionStrategyInfluencer}
      */
+    // FIXME: 0.43 - remove deprecated method
     @Deprecated
     public static HttpConnection toConnection(StreamingHttpConnection original,
                                               HttpExecutionStrategyInfluencer influencer) {
@@ -148,6 +175,7 @@ public static HttpConnection toConnection(StreamingHttpConnection original, Http
      * @return The conversion result.
      * @deprecated Use overload with {@link HttpExecutionStrategy} rather than {@link HttpExecutionStrategyInfluencer}
      */
+    // FIXME: 0.43 - remove deprecated method
     @Deprecated
     public static BlockingHttpConnection toBlockingConnection(StreamingHttpConnection original,
                                                               HttpExecutionStrategyInfluencer influencer) {
@@ -175,6 +203,7 @@ public static BlockingHttpConnection toBlockingConnection(StreamingHttpConnectio
      * @return The conversion result.
      * @deprecated Use overload with {@link HttpExecutionStrategy} rather than {@link HttpExecutionStrategyInfluencer}
      */
+    // FIXME: 0.43 - remove deprecated method
     @Deprecated
     public static BlockingStreamingHttpConnection toBlockingStreamingConnection(
             StreamingHttpConnection original, HttpExecutionStrategyInfluencer influencer) {
@@ -203,6 +232,7 @@ public static BlockingStreamingHttpConnection toBlockingStreamingConnection(
      * @return The conversion result.
      * @deprecated Use overload with {@link HttpExecutionStrategy} rather than {@link HttpExecutionStrategyInfluencer}
      */
+    // FIXME: 0.43 - remove deprecated method
     @Deprecated
     public static HttpClient toClient(StreamingHttpClient original, HttpExecutionStrategyInfluencer influencer) {
         return toClient(original, influencer.requiredOffloads());
@@ -219,6 +249,18 @@ public static HttpClient toClient(StreamingHttpClient original, HttpExecutionStr
         return new StreamingHttpClientToHttpClient(original, strategy);
     }
 
+    /**
+     * Convert from {@link FilterableStreamingHttpClient} to {@link StreamingHttpClient}.
+     *
+     * @param client Original {@link FilterableClientToClient} to convert.
+     * @param strategy required strategy for the service when invoking the resulting {@link HttpClient}
+     * @return The conversion result.
+     */
+    public static StreamingHttpClient toStreamingClient(final FilterableStreamingHttpClient client,
+                                                        final HttpExecutionStrategy strategy) {
+        return new FilterableClientToClient(client, strategy);
+    }
+
     /**
      * Convert from {@link StreamingHttpClient} to {@link BlockingHttpClient}.
      *
@@ -228,6 +270,7 @@ public static HttpClient toClient(StreamingHttpClient original, HttpExecutionStr
      * @return The conversion result.
      * @deprecated Use overload with {@link HttpExecutionStrategy} rather than {@link HttpExecutionStrategyInfluencer}
      */
+    // FIXME: 0.43 - remove deprecated method
     @Deprecated
     public static BlockingHttpClient toBlockingClient(StreamingHttpClient original,
                                                       HttpExecutionStrategyInfluencer influencer) {
@@ -254,6 +297,7 @@ public static BlockingHttpClient toBlockingClient(StreamingHttpClient original,
      * @return The conversion result.
      * @deprecated Use overload with {@link HttpExecutionStrategy} rather than {@link HttpExecutionStrategyInfluencer}
      */
+    // FIXME: 0.43 - remove deprecated method
     @Deprecated
     public static BlockingStreamingHttpClient toBlockingStreamingClient(StreamingHttpClient original,
                                                                         HttpExecutionStrategyInfluencer influencer) {
@@ -281,6 +325,7 @@ public static BlockingStreamingHttpClient toBlockingStreamingClient(StreamingHtt
      * @return {@link ServiceAdapterHolder} containing the service adapted to the streaming programming model.
      * @deprecated Use overload with {@link HttpExecutionStrategy} rather than {@link HttpExecutionStrategyInfluencer}
      */
+    // FIXME: 0.43 - remove deprecated method
     @Deprecated
     public static ServiceAdapterHolder toStreamingHttpService(HttpService service,
                                                               HttpExecutionStrategyInfluencer influencer) {
@@ -307,6 +352,7 @@ public static ServiceAdapterHolder toStreamingHttpService(HttpService service, H
      * @return {@link ServiceAdapterHolder} containing the service adapted to the streaming programming model.
      * @deprecated Use overload with {@link HttpExecutionStrategy} rather than {@link HttpExecutionStrategyInfluencer}
      */
+    // FIXME: 0.43 - remove deprecated method
     @Deprecated
     public static ServiceAdapterHolder toStreamingHttpService(BlockingStreamingHttpService service,
                                                               HttpExecutionStrategyInfluencer influencer) {
@@ -334,6 +380,7 @@ public static ServiceAdapterHolder toStreamingHttpService(BlockingStreamingHttpS
      * @return {@link ServiceAdapterHolder} containing the service adapted to the streaming programming model.
      * @deprecated Use overload with {@link HttpExecutionStrategy} rather than {@link HttpExecutionStrategyInfluencer}
      */
+    // FIXME: 0.43 - remove deprecated method
     @Deprecated
     public static ServiceAdapterHolder toStreamingHttpService(BlockingHttpService service,
                                                               HttpExecutionStrategyInfluencer influencer) {

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

@@ -30,31 +30,31 @@
 interface HttpClientBuilder<U, R, SDE extends ServiceDiscovererEvent<R>> {
 
     /**
-     * Sets the {@link Executor} for all connections created from this builder.
+     * Sets the {@link Executor} for all clients created from this builder.
      *
      * @param executor {@link IoExecutor} to use.
      * @return {@code this}.
      */
     HttpClientBuilder<U, R, SDE> executor(Executor executor);
 
     /**
-     * Sets the {@link IoExecutor} for all connections created from this builder.
+     * Sets the {@link IoExecutor} for all clients created from this builder.
      *
      * @param ioExecutor {@link IoExecutor} to use.
      * @return {@code this}.
      */
     HttpClientBuilder<U, R, SDE> ioExecutor(IoExecutor ioExecutor);
 
     /**
-     * Sets the {@link BufferAllocator} for all connections created from this builder.
+     * Sets the {@link BufferAllocator} for all clients created from this builder.
      *
      * @param allocator {@link BufferAllocator} to use.
      * @return {@code this}.
      */
     HttpClientBuilder<U, R, SDE> bufferAllocator(BufferAllocator allocator);
 
     /**
-     * Sets the {@link HttpExecutionStrategy} for all connections created from this builder.
+     * Sets the {@link HttpExecutionStrategy} for all clients created from this builder.
      *
      * @param strategy {@link HttpExecutionStrategy} to use.
      * @return {@code this}.

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

@@ -33,6 +33,12 @@ public final class HttpContextKeys {
     public static final Key<HttpExecutionStrategy> HTTP_EXECUTION_STRATEGY_KEY =
             newKey("HTTP_EXECUTION_STRATEGY_KEY", HttpExecutionStrategy.class);
 
+    /**
+     * Tracks the original API mode used for client API conversions
+     */
+    public static final Key<HttpApiConversions.ClientAPI> HTTP_CLIENT_API_KEY =
+            newKey("HTTP_CLIENT_API_KEY", HttpApiConversions.ClientAPI.class);
+
     private HttpContextKeys() {
         // No instances
     }