Make RequestConcurrencyController public

servicetalk-client-api-internal/src/main/java/io/servicetalk/client/api/internal/AbstractRequestConcurrencyController.java

@@ -16,6 +16,7 @@
 package io.servicetalk.client.api.internal;
 
 import io.servicetalk.client.api.ConsumableEvent;
+import io.servicetalk.client.api.RequestConcurrencyController;
 import io.servicetalk.concurrent.Cancellable;
 import io.servicetalk.concurrent.CompletableSource;
 import io.servicetalk.concurrent.api.Completable;

servicetalk-client-api-internal/src/main/java/io/servicetalk/client/api/internal/AbstractReservableRequestConcurrencyController.java

@@ -16,6 +16,7 @@
 package io.servicetalk.client.api.internal;
 
 import io.servicetalk.client.api.ConsumableEvent;
+import io.servicetalk.client.api.ReservableRequestConcurrencyController;
 import io.servicetalk.concurrent.Cancellable;
 import io.servicetalk.concurrent.CompletableSource.Subscriber;
 import io.servicetalk.concurrent.api.Completable;

servicetalk-client-api-internal/src/main/java/io/servicetalk/client/api/internal/RequestConcurrencyControllerMulti.java

@@ -19,9 +19,9 @@
 import io.servicetalk.concurrent.api.Completable;
 import io.servicetalk.concurrent.api.Publisher;
 
-import static io.servicetalk.client.api.internal.RequestConcurrencyController.Result.Accepted;
-import static io.servicetalk.client.api.internal.RequestConcurrencyController.Result.RejectedPermanently;
-import static io.servicetalk.client.api.internal.RequestConcurrencyController.Result.RejectedTemporary;
+import static io.servicetalk.client.api.RequestConcurrencyController.Result.Accepted;
+import static io.servicetalk.client.api.RequestConcurrencyController.Result.RejectedPermanently;
+import static io.servicetalk.client.api.RequestConcurrencyController.Result.RejectedTemporary;
 
 final class RequestConcurrencyControllerMulti extends AbstractRequestConcurrencyController {
     private final int maxRequests;

servicetalk-client-api-internal/src/main/java/io/servicetalk/client/api/internal/RequestConcurrencyControllerOnlySingle.java

@@ -19,9 +19,9 @@
 import io.servicetalk.concurrent.api.Completable;
 import io.servicetalk.concurrent.api.Publisher;
 
-import static io.servicetalk.client.api.internal.RequestConcurrencyController.Result.Accepted;
-import static io.servicetalk.client.api.internal.RequestConcurrencyController.Result.RejectedPermanently;
-import static io.servicetalk.client.api.internal.RequestConcurrencyController.Result.RejectedTemporary;
+import static io.servicetalk.client.api.RequestConcurrencyController.Result.Accepted;
+import static io.servicetalk.client.api.RequestConcurrencyController.Result.RejectedPermanently;
+import static io.servicetalk.client.api.RequestConcurrencyController.Result.RejectedTemporary;
 
 final class RequestConcurrencyControllerOnlySingle extends AbstractRequestConcurrencyController {
     RequestConcurrencyControllerOnlySingle(final Publisher<? extends ConsumableEvent<Integer>> maxConcurrency,

servicetalk-client-api-internal/src/main/java/io/servicetalk/client/api/internal/RequestConcurrencyControllers.java

@@ -16,6 +16,7 @@
 package io.servicetalk.client.api.internal;
 
 import io.servicetalk.client.api.ConsumableEvent;
+import io.servicetalk.client.api.RequestConcurrencyController;
 import io.servicetalk.concurrent.api.Completable;
 import io.servicetalk.concurrent.api.Publisher;
 

servicetalk-client-api-internal/src/main/java/io/servicetalk/client/api/internal/ReservableRequestConcurrencyControllerMulti.java

@@ -19,9 +19,9 @@
 import io.servicetalk.concurrent.api.Completable;
 import io.servicetalk.concurrent.api.Publisher;
 
-import static io.servicetalk.client.api.internal.RequestConcurrencyController.Result.Accepted;
-import static io.servicetalk.client.api.internal.RequestConcurrencyController.Result.RejectedPermanently;
-import static io.servicetalk.client.api.internal.RequestConcurrencyController.Result.RejectedTemporary;
+import static io.servicetalk.client.api.RequestConcurrencyController.Result.Accepted;
+import static io.servicetalk.client.api.RequestConcurrencyController.Result.RejectedPermanently;
+import static io.servicetalk.client.api.RequestConcurrencyController.Result.RejectedTemporary;
 
 final class ReservableRequestConcurrencyControllerMulti extends AbstractReservableRequestConcurrencyController {
     private final int maxRequests;

servicetalk-client-api-internal/src/main/java/io/servicetalk/client/api/internal/ReservableRequestConcurrencyControllerOnlySingle.java

@@ -19,9 +19,9 @@
 import io.servicetalk.concurrent.api.Completable;
 import io.servicetalk.concurrent.api.Publisher;
 
-import static io.servicetalk.client.api.internal.RequestConcurrencyController.Result.Accepted;
-import static io.servicetalk.client.api.internal.RequestConcurrencyController.Result.RejectedPermanently;
-import static io.servicetalk.client.api.internal.RequestConcurrencyController.Result.RejectedTemporary;
+import static io.servicetalk.client.api.RequestConcurrencyController.Result.Accepted;
+import static io.servicetalk.client.api.RequestConcurrencyController.Result.RejectedPermanently;
+import static io.servicetalk.client.api.RequestConcurrencyController.Result.RejectedTemporary;
 
 final class ReservableRequestConcurrencyControllerOnlySingle extends AbstractReservableRequestConcurrencyController {
     ReservableRequestConcurrencyControllerOnlySingle(final Publisher<? extends ConsumableEvent<Integer>> maxConcurrency,

servicetalk-client-api-internal/src/main/java/io/servicetalk/client/api/internal/ReservableRequestConcurrencyControllers.java

@@ -16,6 +16,8 @@
 package io.servicetalk.client.api.internal;
 
 import io.servicetalk.client.api.ConsumableEvent;
+import io.servicetalk.client.api.RequestConcurrencyController;
+import io.servicetalk.client.api.ReservableRequestConcurrencyController;
 import io.servicetalk.concurrent.api.Completable;
 import io.servicetalk.concurrent.api.Publisher;
 

servicetalk-client-api-internal/src/test/java/io/servicetalk/client/api/internal/AbstractRequestConcurrencyControllerMultiTest.java

@@ -15,15 +15,16 @@
  */
 package io.servicetalk.client.api.internal;
 
+import io.servicetalk.client.api.RequestConcurrencyController;
 import io.servicetalk.concurrent.api.Completable;
 import io.servicetalk.concurrent.api.Publisher;
 import io.servicetalk.concurrent.api.TestPublisher;
 
 import org.junit.jupiter.api.Test;
 
-import static io.servicetalk.client.api.internal.RequestConcurrencyController.Result.Accepted;
-import static io.servicetalk.client.api.internal.RequestConcurrencyController.Result.RejectedPermanently;
-import static io.servicetalk.client.api.internal.RequestConcurrencyController.Result.RejectedTemporary;
+import static io.servicetalk.client.api.RequestConcurrencyController.Result.Accepted;
+import static io.servicetalk.client.api.RequestConcurrencyController.Result.RejectedPermanently;
+import static io.servicetalk.client.api.RequestConcurrencyController.Result.RejectedTemporary;
 import static io.servicetalk.concurrent.api.Completable.completed;
 import static io.servicetalk.concurrent.api.Completable.never;
 import static io.servicetalk.concurrent.api.Publisher.from;

servicetalk-client-api-internal/src/test/java/io/servicetalk/client/api/internal/AbstractRequestConcurrencyControllerOnlySingleTest.java

@@ -15,15 +15,16 @@
  */
 package io.servicetalk.client.api.internal;
 
+import io.servicetalk.client.api.RequestConcurrencyController;
 import io.servicetalk.concurrent.api.Completable;
 import io.servicetalk.concurrent.api.Publisher;
 import io.servicetalk.concurrent.api.TestPublisher;
 
 import org.junit.jupiter.api.Test;
 
-import static io.servicetalk.client.api.internal.RequestConcurrencyController.Result.Accepted;
-import static io.servicetalk.client.api.internal.RequestConcurrencyController.Result.RejectedPermanently;
-import static io.servicetalk.client.api.internal.RequestConcurrencyController.Result.RejectedTemporary;
+import static io.servicetalk.client.api.RequestConcurrencyController.Result.Accepted;
+import static io.servicetalk.client.api.RequestConcurrencyController.Result.RejectedPermanently;
+import static io.servicetalk.client.api.RequestConcurrencyController.Result.RejectedTemporary;
 import static io.servicetalk.concurrent.api.Completable.completed;
 import static io.servicetalk.concurrent.api.Completable.never;
 import static io.servicetalk.concurrent.api.Publisher.from;

servicetalk-client-api-internal/src/test/java/io/servicetalk/client/api/internal/RequestConcurrencyControllerMultiTest.java

@@ -15,6 +15,7 @@
  */
 package io.servicetalk.client.api.internal;
 
+import io.servicetalk.client.api.RequestConcurrencyController;
 import io.servicetalk.concurrent.api.Completable;
 import io.servicetalk.concurrent.api.Publisher;
 

servicetalk-client-api-internal/src/test/java/io/servicetalk/client/api/internal/RequestConcurrencyControllerOnlySingleTest.java

@@ -15,6 +15,7 @@
  */
 package io.servicetalk.client.api.internal;
 
+import io.servicetalk.client.api.RequestConcurrencyController;
 import io.servicetalk.concurrent.api.Completable;
 import io.servicetalk.concurrent.api.Publisher;
 

servicetalk-client-api-internal/src/test/java/io/servicetalk/client/api/internal/ReservableRequestConcurrencyControllerMultiTest.java

@@ -15,12 +15,13 @@
  */
 package io.servicetalk.client.api.internal;
 
+import io.servicetalk.client.api.ReservableRequestConcurrencyController;
 import io.servicetalk.concurrent.api.Completable;
 import io.servicetalk.concurrent.api.Publisher;
 
 import org.junit.jupiter.api.Test;
 
-import static io.servicetalk.client.api.internal.RequestConcurrencyController.Result.Accepted;
+import static io.servicetalk.client.api.RequestConcurrencyController.Result.Accepted;
 import static io.servicetalk.concurrent.api.Completable.never;
 import static io.servicetalk.concurrent.api.Publisher.from;
 import static org.hamcrest.MatcherAssert.assertThat;

servicetalk-client-api-internal/src/test/java/io/servicetalk/client/api/internal/ReservableRequestConcurrencyControllerOnlySingleTest.java

@@ -15,12 +15,13 @@
  */
 package io.servicetalk.client.api.internal;
 
+import io.servicetalk.client.api.ReservableRequestConcurrencyController;
 import io.servicetalk.concurrent.api.Completable;
 import io.servicetalk.concurrent.api.Publisher;
 
 import org.junit.jupiter.api.Test;
 
-import static io.servicetalk.client.api.internal.RequestConcurrencyController.Result.Accepted;
+import static io.servicetalk.client.api.RequestConcurrencyController.Result.Accepted;
 import static io.servicetalk.client.api.internal.ReservableRequestConcurrencyControllers.newSingleController;
 import static io.servicetalk.concurrent.api.Completable.never;
 import static io.servicetalk.concurrent.api.Publisher.from;

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

@@ -15,13 +15,41 @@
  */
 package io.servicetalk.client.api;
 
+import io.servicetalk.concurrent.api.Completable;
 import io.servicetalk.concurrent.api.ListenableAsyncCloseable;
 
+import static io.servicetalk.concurrent.api.Completable.failed;
+
 /**
  * A connection managed by a {@link LoadBalancer}.
- * <p>
- * Note: {@link ScoreSupplier} contract provides this connection's score, where 0.0 is the lowest score and 1.0 is the
- * highest. {@link LoadBalancer}s prefer connections with a higher score.
+ *
+ * @see ScoreSupplier
+ * @see ReservableRequestConcurrencyController
  */
-public interface LoadBalancedConnection extends ListenableAsyncCloseable, ScoreSupplier {
+public interface LoadBalancedConnection extends ScoreSupplier, ReservableRequestConcurrencyController,
+                                                ListenableAsyncCloseable {
+
+    @Override
+    default Result tryRequest() {
+        throw new UnsupportedOperationException("LoadBalancedConnection#tryRequest() is not supported by " +
+                getClass());
+    }
+
+    @Override
+    default void requestFinished() {
+        throw new UnsupportedOperationException("LoadBalancedConnection#requestFinished() is not supported by " +
+                getClass());
+    }
+
+    @Override
+    default boolean tryReserve() {
+        throw new UnsupportedOperationException("LoadBalancedConnection#tryReserve() is not supported by " +
+                getClass());
+    }
+
+    @Override
+    default Completable releaseAsync() {
+        return failed(new UnsupportedOperationException("LoadBalancedConnection#releaseAsync() is not supported by " +
+                getClass()));
+    }
 }

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

@@ -46,7 +46,7 @@ public interface LoadBalancerFactory<ResolvedAddress, C extends LoadBalancedConn
      * @deprecated In the future only {@link #newLoadBalancer(String, Publisher, ConnectionFactory)} will remain,
      * please use that method instead.
      */
-    @Deprecated
+    @Deprecated // FIXME: 0.43 - remove deprecated method
     default <T extends C> LoadBalancer<T> newLoadBalancer(
             Publisher<? extends ServiceDiscovererEvent<ResolvedAddress>> eventPublisher,
             ConnectionFactory<ResolvedAddress, T> connectionFactory) {

servicetalk-client-api-internal/src/main/java/io/servicetalk/client/api/internal/RequestConcurrencyController.java → servicetalk-client-api/src/main/java/io/servicetalk/client/api/RequestConcurrencyController.java

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.servicetalk.client.api.internal;
+package io.servicetalk.client.api;
 
 import io.servicetalk.concurrent.api.Publisher;
 

servicetalk-client-api-internal/src/main/java/io/servicetalk/client/api/internal/ReservableRequestConcurrencyController.java → servicetalk-client-api/src/main/java/io/servicetalk/client/api/ReservableRequestConcurrencyController.java

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.servicetalk.client.api.internal;
+package io.servicetalk.client.api;
 
 import io.servicetalk.concurrent.api.Completable;
 

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

@@ -19,6 +19,8 @@
 import io.servicetalk.concurrent.api.Executor;
 import io.servicetalk.transport.api.IoExecutor;
 
+import static java.util.Objects.requireNonNull;
+
 /**
  * An implementation of {@link HttpExecutionContext} that delegates all method calls to another
  * {@link HttpExecutionContext}.
@@ -32,7 +34,16 @@ public class DelegatingHttpExecutionContext implements HttpExecutionContext {
      * @param delegate {@link HttpExecutionContext} to delegate all calls.
      */
     public DelegatingHttpExecutionContext(final HttpExecutionContext delegate) {
-        this.delegate = delegate;
+        this.delegate = requireNonNull(delegate);
+    }
+
+    /**
+     * Get the {@link HttpExecutionContext} that this class delegates to.
+     *
+     * @return the {@link HttpExecutionContext} that this class delegates to.
+     */
+    protected final HttpExecutionContext delegate() {
+        return delegate;
     }
 
     @Override

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

@@ -1,5 +1,5 @@
 /*
- * Copyright © 2019 Apple Inc. and the ServiceTalk project authors
+ * Copyright © 2019, 2022 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.
@@ -16,11 +16,41 @@
 package io.servicetalk.http.api;
 
 import io.servicetalk.client.api.LoadBalancedConnection;
+import io.servicetalk.concurrent.api.Completable;
+
+import static io.servicetalk.concurrent.api.Completable.failed;
 
 /**
  * A {@link FilterableStreamingHttpConnection} that supported filtering and implements the {@link
  * LoadBalancedConnection} contract.
  */
-public interface FilterableStreamingHttpLoadBalancedConnection
-        extends FilterableStreamingHttpConnection, LoadBalancedConnection {
+public interface FilterableStreamingHttpLoadBalancedConnection extends FilterableStreamingHttpConnection,
+        LoadBalancedConnection, ReservedStreamingHttpConnection {
+
+    // FIXME: 0.43 - consider removing default implementations
+    @Override
+    default ReservedHttpConnection asConnection() {
+        throw new UnsupportedOperationException(
+                "FilterableStreamingHttpLoadBalancedConnection#asConnection() is not supported by " + getClass());
+    }
+
+    @Override
+    default ReservedBlockingStreamingHttpConnection asBlockingStreamingConnection() {
+        throw new UnsupportedOperationException(
+                "FilterableStreamingHttpLoadBalancedConnection#asBlockingStreamingConnection() is not supported by " +
+                        getClass());
+    }
+
+    @Override
+    default ReservedBlockingHttpConnection asBlockingConnection() {
+        throw new UnsupportedOperationException(
+                "FilterableStreamingHttpLoadBalancedConnection#asBlockingConnection() is not supported by " +
+                        getClass());
+    }
+
+    @Override
+    default Completable releaseAsync() {
+        return failed(new UnsupportedOperationException(
+                "FilterableStreamingHttpLoadBalancedConnection#releaseAsync() is not supported by " + getClass()));
+    }
 }