Give access to the underlying TCP ConnectionContext from HTTP/2 servi…

…ce (#2319)

Motivation:

Server-side users have access only to HTTP/2 stream `ConnectionContext`
from the `HttpServiceContext`. In some cases, it's important to access
a `ConnectionContext` of the underlying TCP connection.

Modifications:

- Add `@Nullable ConnectionContext#parent()` method;
- Implement support for a new method;
- Add tests;

Result:

Users can access the underlying TCP `ConnectionContext` from the HTTP/2
service endpoint.

servicetalk-benchmarks/src/jmh/java/io/servicetalk/http/netty/NettyPipelinedConnectionBenchmark.java

@@ -23,6 +23,7 @@
 import io.servicetalk.concurrent.api.Publisher;
 import io.servicetalk.concurrent.api.Single;
 import io.servicetalk.http.api.HttpProtocolVersion;
+import io.servicetalk.transport.api.ConnectionContext;
 import io.servicetalk.transport.api.ExecutionContext;
 import io.servicetalk.transport.api.SslConfig;
 import io.servicetalk.transport.netty.internal.FlushStrategies;
@@ -150,6 +151,7 @@ public void onComplete() {
 
     private NettyConnection<Object, Object> newNettyConnection() {
         return new NettyConnection<Object, Object>() {
+
             private final AtomicInteger offloadCount = new AtomicInteger();
             @Override
             public Publisher<Object> read() {
@@ -261,6 +263,12 @@ public Protocol protocol() {
                 return HttpProtocolVersion.HTTP_1_1;
             }
 
+            @Nullable
+            @Override
+            public ConnectionContext parent() {
+                return null;
+            }
+
             @Override
             public Completable onClose() {
                 return Completable.never();

servicetalk-grpc-api/src/main/java/io/servicetalk/grpc/api/DefaultGrpcServiceContext.java

@@ -91,6 +91,12 @@ public GrpcProtocol protocol() {
         return protocol;
     }
 
+    @Nullable
+    @Override
+    public ConnectionContext parent() {
+        return connectionContext.parent();
+    }
+
     @Override
     public Completable onClose() {
         return connectionContext.onClose();

servicetalk-grpc-netty/src/test/java/io/servicetalk/grpc/customtransport/Utils.java

@@ -37,6 +37,7 @@
 import io.servicetalk.serializer.api.StreamingDeserializer;
 import io.servicetalk.serializer.api.StreamingSerializer;
 import io.servicetalk.serializer.utils.FramedDeserializerOperator;
+import io.servicetalk.transport.api.ConnectionContext;
 import io.servicetalk.transport.api.IoExecutor;
 import io.servicetalk.transport.api.SslConfig;
 
@@ -194,6 +195,12 @@ public GrpcProtocol protocol() {
             return InMemoryGrpcProtocol.INSTANCE;
         }
 
+        @Nullable
+        @Override
+        public ConnectionContext parent() {
+            return null;
+        }
+
         @Deprecated
         @Override
         public List<ContentCodec> supportedMessageCodings() {

servicetalk-grpc-netty/src/test/java/io/servicetalk/grpc/netty/ParentConnectionContextTest.java

@@ -0,0 +1,144 @@
+/*
+ * Copyright © 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.
+ * 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.grpc.netty;
+
+import io.servicetalk.concurrent.BlockingIterable;
+import io.servicetalk.concurrent.api.Publisher;
+import io.servicetalk.concurrent.api.Single;
+import io.servicetalk.grpc.api.GrpcPayloadWriter;
+import io.servicetalk.grpc.api.GrpcServiceContext;
+import io.servicetalk.grpc.netty.TesterProto.TestRequest;
+import io.servicetalk.grpc.netty.TesterProto.TestResponse;
+import io.servicetalk.grpc.netty.TesterProto.Tester.BlockingTesterClient;
+import io.servicetalk.grpc.netty.TesterProto.Tester.BlockingTesterService;
+import io.servicetalk.grpc.netty.TesterProto.Tester.TesterService;
+import io.servicetalk.transport.api.ConnectionContext;
+import io.servicetalk.transport.api.ServerContext;
+
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import static io.servicetalk.concurrent.api.Publisher.from;
+import static io.servicetalk.concurrent.api.Single.succeeded;
+import static io.servicetalk.test.resources.TestUtils.assertNoAsyncErrors;
+import static io.servicetalk.transport.netty.internal.AddressUtils.localAddress;
+import static io.servicetalk.transport.netty.internal.AddressUtils.serverHostAndPort;
+import static java.util.Collections.singleton;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.notNullValue;
+import static org.hamcrest.Matchers.nullValue;
+import static org.hamcrest.Matchers.sameInstance;
+
+class ParentConnectionContextTest {
+
+    private final BlockingQueue<Throwable> errors = new LinkedBlockingQueue<>();
+
+    @ParameterizedTest(name = "{index}: blocking={0}")
+    @ValueSource(booleans = {false, true})
+    void test(boolean blocking) throws Exception {
+        try (ServerContext serverContext = GrpcServers.forAddress(localAddress(0))
+                .listenAndAwait(blocking ? new BlockingTesterServiceImpl() : new TesterServiceImpl());
+             BlockingTesterClient client = GrpcClients.forAddress(serverHostAndPort(serverContext))
+                     .buildBlocking(new TesterProto.Tester.ClientFactory())) {
+
+            client.test(TestRequest.getDefaultInstance());
+            client.testBiDiStream(singleton(TestRequest.getDefaultInstance())).forEach(__ -> { /* noop */ });
+            client.testResponseStream(TestRequest.getDefaultInstance()).forEach(__ -> { /* noop */ });
+            client.testRequestStream(singleton(TestRequest.getDefaultInstance()));
+        }
+        assertNoAsyncErrors(errors);
+    }
+
+    private final class TesterServiceImpl implements TesterService {
+
+        @Override
+        public Single<TestResponse> test(GrpcServiceContext ctx, TestRequest request) {
+            assertServiceContext(ctx, errors);
+            return succeeded(TestResponse.getDefaultInstance());
+        }
+
+        @Override
+        public Publisher<TestResponse> testBiDiStream(GrpcServiceContext ctx, Publisher<TestRequest> request) {
+            assertServiceContext(ctx, errors);
+            return request.ignoreElements().concat(from(TestResponse.getDefaultInstance()));
+        }
+
+        @Override
+        public Publisher<TestResponse> testResponseStream(GrpcServiceContext ctx, TestRequest request) {
+            assertServiceContext(ctx, errors);
+            return from(TestResponse.getDefaultInstance());
+        }
+
+        @Override
+        public Single<TestResponse> testRequestStream(GrpcServiceContext ctx, Publisher<TestRequest> request) {
+            assertServiceContext(ctx, errors);
+            return request.ignoreElements().concat(succeeded(TestResponse.getDefaultInstance()));
+        }
+    }
+
+    private final class BlockingTesterServiceImpl implements BlockingTesterService {
+
+        @Override
+        public TestResponse test(GrpcServiceContext ctx, TestRequest request) {
+            assertServiceContext(ctx, errors);
+            return TestResponse.getDefaultInstance();
+        }
+
+        @Override
+        public void testBiDiStream(GrpcServiceContext ctx, BlockingIterable<TestRequest> request,
+                                   GrpcPayloadWriter<TestResponse> responseWriter) throws Exception {
+            assertServiceContext(ctx, errors);
+            request.forEach(__ -> { /* noop */ });
+            responseWriter.write(TestResponse.getDefaultInstance());
+            responseWriter.close();
+        }
+
+        @Override
+        public void testResponseStream(GrpcServiceContext ctx, TestRequest request,
+                                       GrpcPayloadWriter<TestResponse> responseWriter) throws Exception {
+            assertServiceContext(ctx, errors);
+            responseWriter.write(TestResponse.getDefaultInstance());
+            responseWriter.close();
+        }
+
+        @Override
+        public TestResponse testRequestStream(GrpcServiceContext ctx, BlockingIterable<TestRequest> request) {
+            assertServiceContext(ctx, errors);
+            request.forEach(__ -> { /* noop */ });
+            return TestResponse.getDefaultInstance();
+        }
+    }
+
+    private static void assertServiceContext(GrpcServiceContext ctx, BlockingQueue<Throwable> errors) {
+        try {
+            assertThat(ctx, is(notNullValue()));
+            ConnectionContext parent = ctx.parent();
+            assertThat("gRPC stream must have reference to its parent ConnectionContext",
+                    parent, is(notNullValue()));
+            assertThat("Unexpected localAddress",
+                    parent.localAddress(), is(sameInstance(ctx.localAddress())));
+            assertThat("Unexpected remoteAddress",
+                    parent.remoteAddress(), is(sameInstance(ctx.remoteAddress())));
+            assertThat("Unexpected parent.parent()", parent.parent(), is(nullValue()));
+        } catch (Throwable t) {
+            errors.add(t);
+        }
+    }
+}

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

@@ -16,6 +16,7 @@
 package io.servicetalk.http.api;
 
 import io.servicetalk.concurrent.api.Completable;
+import io.servicetalk.transport.api.ConnectionContext;
 import io.servicetalk.transport.api.SslConfig;
 
 import java.net.SocketAddress;
@@ -93,6 +94,12 @@ public HttpProtocolVersion protocol() {
         return delegate.protocol();
     }
 
+    @Nullable
+    @Override
+    public ConnectionContext parent() {
+        return delegate.parent();
+    }
+
     @Override
     public Completable onClose() {
         return delegate.onClose();

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

@@ -27,4 +27,9 @@ public interface HttpConnectionContext extends ConnectionContext {
 
     @Override
     HttpProtocolVersion protocol();
+
+    // Do not override the return type of parent() method to HttpConnectionContext as there is a possibility to have an
+    // HttpConnectionContext over a non-HTTP ConnectionContext. Also, with HttpConnectionContext as a return type it
+    // makes implementation a lot harder because we receive a non-HTTP NettyConnectionContext from the transport and
+    // wrap it later.
 }

servicetalk-http-api/src/testFixtures/java/io/servicetalk/http/api/TestHttpServiceContext.java

@@ -16,6 +16,7 @@
 package io.servicetalk.http.api;
 
 import io.servicetalk.concurrent.api.Completable;
+import io.servicetalk.transport.api.ConnectionContext;
 import io.servicetalk.transport.api.SslConfig;
 import io.servicetalk.transport.netty.internal.AddressUtils;
 
@@ -100,6 +101,12 @@ public HttpProtocolVersion protocol() {
         return HTTP_1_0;
     }
 
+    @Nullable
+    @Override
+    public ConnectionContext parent() {
+        return null;
+    }
+
     @Override
     public Completable onClose() {
         return completed();

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

@@ -21,12 +21,15 @@
 import io.servicetalk.http.api.HttpConnectionContext;
 import io.servicetalk.http.api.HttpExecutionContext;
 import io.servicetalk.http.api.HttpProtocolVersion;
+import io.servicetalk.transport.api.ConnectionContext;
 import io.servicetalk.transport.api.DelegatingConnectionContext;
 import io.servicetalk.transport.netty.internal.FlushStrategy;
 import io.servicetalk.transport.netty.internal.NettyConnectionContext;
 
 import io.netty.channel.Channel;
 
+import javax.annotation.Nullable;
+
 import static java.util.Objects.requireNonNull;
 
 final class DefaultNettyHttpConnectionContext extends DelegatingConnectionContext
@@ -52,6 +55,12 @@ public HttpProtocolVersion protocol() {
         return (HttpProtocolVersion) nettyConnectionContext.protocol();
     }
 
+    @Nullable
+    @Override
+    public ConnectionContext parent() {
+        return nettyConnectionContext.parent();
+    }
+
     @Override
     public Cancellable updateFlushStrategy(final FlushStrategyProvider strategyProvider) {
         return nettyConnectionContext.updateFlushStrategy(strategyProvider);

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

@@ -40,6 +40,7 @@
 import io.servicetalk.http.api.StreamingHttpResponseFactory;
 import io.servicetalk.http.netty.LoadBalancedStreamingHttpClient.OnStreamClosedRunnable;
 import io.servicetalk.http.netty.ReservableRequestConcurrencyControllers.IgnoreConsumedEvent;
+import io.servicetalk.transport.api.ConnectionContext;
 import io.servicetalk.transport.api.ConnectionObserver;
 import io.servicetalk.transport.api.ConnectionObserver.MultiplexedObserver;
 import io.servicetalk.transport.api.ConnectionObserver.StreamObserver;
@@ -348,13 +349,11 @@ private void childChannelActive(Future<Http2StreamChannel> future,
                             closeHandler, streamObserver));
                     DefaultNettyConnection<Object, Object> nettyConnection =
                             DefaultNettyConnection.initChildChannel(streamChannel,
-                                    parentContext.executionContext(),
+                                    parentContext,
                                     closeHandler,
-                                    parentContext.flushStrategyHolder.currentStrategy(),
+                                    parentContext.defaultFlushStrategy(),
                                     parentContext.idleTimeoutMs,
                                     HTTP_2_0,
-                                    parentContext.sslConfig(),
-                                    parentContext.sslSession(),
                                     parentContext.nettyChannel().config(),
                                     streamObserver,
                                     true, OBJ_EXPECT_CONTINUE,
@@ -440,6 +439,12 @@ public HttpProtocolVersion protocol() {
             return parentContext.protocol();
         }
 
+        @Nullable
+        @Override
+        public ConnectionContext parent() {
+            return parentContext.parent();
+        }
+
         @Override
         public StreamingHttpResponseFactory httpResponseFactory() {
             return reqRespFactory;

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

@@ -25,6 +25,7 @@
 import io.servicetalk.http.api.HttpConnectionContext;
 import io.servicetalk.http.api.HttpExecutionContext;
 import io.servicetalk.http.api.HttpProtocolVersion;
+import io.servicetalk.transport.api.ConnectionContext;
 import io.servicetalk.transport.api.ConnectionObserver;
 import io.servicetalk.transport.api.SslConfig;
 import io.servicetalk.transport.netty.internal.FlushStrategy;
@@ -149,6 +150,12 @@ public HttpProtocolVersion protocol() {
         return HTTP_2_0;
     }
 
+    @Nullable
+    @Override
+    public ConnectionContext parent() {
+        return null;
+    }
+
     @Override
     public final Channel nettyChannel() {
         return channel();

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

@@ -165,17 +165,15 @@ protected void initChannel(final Http2StreamChannel streamChannel) {
                                 // ServiceTalk <-> Netty netty utilities
                                 DefaultNettyConnection<Object, Object> streamConnection =
                                         DefaultNettyConnection.initChildChannel(streamChannel,
-                                                connection.executionContext(),
+                                                connection,
                                                 closeHandler,
                                                 // TODO(scott): after flushStrategy is no longer on the connection
                                                 // level we can use DefaultNettyConnection.initChannel instead of this
                                                 // custom method.
-                                                connection.flushStrategyHolder.currentStrategy(),
+                                                connection.defaultFlushStrategy(),
                                                 connection.idleTimeoutMs,
                                                 HTTP_2_0,
-                                                connection.sslConfig(),
-                                                connection.sslSession(),
-                                                channel.config(),
+                                                connection.nettyChannel().config(),
                                                 streamObserver,
                                                 false, __ -> false,
                                                 NettyHttp2ExceptionUtils::wrapIfNecessary);

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

@@ -47,6 +47,7 @@
 import io.servicetalk.tcp.netty.internal.ReadOnlyTcpServerConfig;
 import io.servicetalk.tcp.netty.internal.TcpServerBinder;
 import io.servicetalk.tcp.netty.internal.TcpServerChannelInitializer;
+import io.servicetalk.transport.api.ConnectionContext;
 import io.servicetalk.transport.api.ConnectionObserver;
 import io.servicetalk.transport.api.ServerContext;
 import io.servicetalk.transport.api.SslConfig;
@@ -510,6 +511,12 @@ public HttpProtocolVersion protocol() {
             return (HttpProtocolVersion) connection.protocol();
         }
 
+        @Nullable
+        @Override
+        public ConnectionContext parent() {
+            return connection.parent();
+        }
+
         @Override
         public Single<Throwable> transportError() {
             return connection.transportError();