diff --git a/servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/DeferredServerChannelBinder.java b/servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/DeferredServerChannelBinder.java index b03af5abf6..8a5e8c6354 100644 --- a/servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/DeferredServerChannelBinder.java +++ b/servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/DeferredServerChannelBinder.java @@ -69,11 +69,16 @@ static Single bind(final HttpExecutionContext executionContex return TcpServerBinder.bind(listenAddress, tcpConfig, executionContext, connectionAcceptor, channelInit, serverConnection -> { - // Start processing requests on http/1.1 connection: + // Notify connection established after all acceptors have completed: if (serverConnection instanceof NettyHttpServerConnection) { - ((NettyHttpServerConnection) serverConnection).process(true); + ((NettyHttpServerConnection) serverConnection).notifyConnectionEstablishedAndProcess(); + } else if (serverConnection instanceof H2ServerParentConnectionContext) { + ((H2ServerParentConnectionContext) serverConnection) + .notifyConnectionEstablishedAndEnableAutoRead(); + } else { + throw new IllegalStateException("Unexpected connection type: " + + serverConnection.getClass().getName()); } - // Nothing to do otherwise as h2 uses auto read on the parent channel }, earlyConnectionAcceptor, lateConnectionAcceptor) .map(delegate -> { LOGGER.debug("Started HTTP server with ALPN for address {}", delegate.listenAddress()); diff --git a/servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/H2ClientParentConnectionContext.java b/servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/H2ClientParentConnectionContext.java index 6a48875de1..500c8e623e 100644 --- a/servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/H2ClientParentConnectionContext.java +++ b/servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/H2ClientParentConnectionContext.java @@ -187,7 +187,8 @@ private static final class DefaultH2ClientParentConnection extends AbstractH2Par HttpHeadersFactory headersFactory, StreamingHttpRequestResponseFactory reqRespFactory, ConnectionObserver observer) { - super(connection, delayedCancellable, waitForSslHandshake, observer); + super(connection, delayedCancellable, waitForSslHandshake, observer, + false /* deferAutoRead: client enables auto-read immediately */); this.subscriber = requireNonNull(subscriber); this.headersFactory = requireNonNull(headersFactory); this.reqRespFactory = requireNonNull(reqRespFactory); diff --git a/servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/H2ParentConnectionContext.java b/servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/H2ParentConnectionContext.java index c393a0eaee..87630817ed 100644 --- a/servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/H2ParentConnectionContext.java +++ b/servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/H2ParentConnectionContext.java @@ -194,15 +194,18 @@ abstract static class AbstractH2ParentConnection extends ChannelInboundHandlerAd final boolean waitForSslHandshake; private final DelayedCancellable delayedCancellable; final ConnectionObserver observer; + private final boolean deferAutoRead; AbstractH2ParentConnection(H2ParentConnectionContext parentContext, DelayedCancellable delayedCancellable, boolean waitForSslHandshake, - ConnectionObserver observer) { + ConnectionObserver observer, + boolean deferAutoRead) { this.parentContext = parentContext; this.delayedCancellable = delayedCancellable; this.waitForSslHandshake = waitForSslHandshake; this.observer = observer; + this.deferAutoRead = deferAutoRead; } abstract void tryCompleteSubscriber(); @@ -226,8 +229,8 @@ public final void handlerAdded(ChannelHandlerContext ctx) { if (channel.isActive()) { doChannelActive(ctx); } - if (!channel.config().isAutoRead()) { - // auto read is required for h2 + if (!deferAutoRead && !channel.config().isAutoRead()) { + // auto read is required for h2, but might be deferred due to connection acceptors channel.config().setAutoRead(true); } } diff --git a/servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/H2ServerParentConnectionContext.java b/servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/H2ServerParentConnectionContext.java index 0e12325f77..f05b914eec 100644 --- a/servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/H2ServerParentConnectionContext.java +++ b/servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/H2ServerParentConnectionContext.java @@ -44,6 +44,7 @@ import io.netty.channel.Channel; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelPipeline; +import io.netty.channel.EventLoop; import io.netty.handler.codec.http2.Http2SettingsFrame; import io.netty.handler.codec.http2.Http2StreamChannel; import org.slf4j.Logger; @@ -66,15 +67,21 @@ final class H2ServerParentConnectionContext extends H2ParentConnectionContext implements ServerContext { private static final Logger LOGGER = LoggerFactory.getLogger(H2ServerParentConnectionContext.class); private final SocketAddress listenAddress; + @Nullable + private ConnectionObserver observer; + @Nullable + private DefaultH2ServerParentConnection parentConnectionHandler; private H2ServerParentConnectionContext(final Channel channel, final HttpExecutionContext executionContext, final FlushStrategy flushStrategy, final long idleTimeoutMs, @Nullable final SslConfig sslConfig, @Nullable final SSLSession sslSession, final SocketAddress listenAddress, - final KeepAliveManager keepAliveManager) { + final KeepAliveManager keepAliveManager, + final ConnectionObserver observer) { super(channel, executionContext, flushStrategy, idleTimeoutMs, sslConfig, sslSession, keepAliveManager); this.listenAddress = requireNonNull(listenAddress); + this.observer = requireNonNull(observer); } @Override @@ -87,6 +94,29 @@ public SocketAddress listenAddress() { return listenAddress; } + /** + * Notifies the observer and enables auto-read. Must be called after all connection acceptors complete. + * Both actions are combined because the {@code multiplexedObserver} must be set before any frames arrive, + * and enabling auto-read is what triggers frame delivery. Always dispatches to the event loop to ensure + * the observer write and {@code setAutoRead(true)} both execute on the thread that processes frames. + */ + void notifyConnectionEstablishedAndEnableAutoRead() { + final EventLoop eventLoop = nettyChannel().eventLoop(); + if (eventLoop.inEventLoop()) { + doNotifyAndEnableAutoRead(); + } else { + eventLoop.execute(this::doNotifyAndEnableAutoRead); + } + } + + private void doNotifyAndEnableAutoRead() { + assert observer != null && parentConnectionHandler != null; + parentConnectionHandler.multiplexedObserver = observer.multiplexedConnectionEstablished(this); + observer = null; + parentConnectionHandler = null; + nettyChannel().config().setAutoRead(true); + } + static Single bind(final HttpExecutionContext executionContext, final ReadOnlyHttpServerConfig config, final SocketAddress listenAddress, @@ -98,11 +128,12 @@ static Single bind(final HttpExecutionContext executionContex return failed(newH2ConfigException()); } final ReadOnlyTcpServerConfig tcpServerConfig = config.tcpConfig(); + // Called AFTER all connection acceptors have completed. return TcpServerBinder.bind(listenAddress, tcpServerConfig, executionContext, connectionAcceptor, (channel, connectionObserver) -> initChannel(listenAddress, channel, executionContext, config, new TcpServerChannelInitializer(tcpServerConfig, connectionObserver, executionContext), service, connectionObserver), - serverConnection -> { /* nothing to do as h2 uses auto read on the parent channel */ }, + H2ServerParentConnectionContext::notifyConnectionEstablishedAndEnableAutoRead, earlyConnectionAcceptor, lateConnectionAcceptor) .map(delegate -> { LOGGER.debug("Started HTTP/2 server with prior-knowledge for address {}", delegate.listenAddress()); @@ -148,11 +179,13 @@ protected void handleSubscribe(final Subscriber() { @@ -190,7 +223,7 @@ protected void initChannel(final Http2StreamChannel streamChannel) { // ServiceTalk HTTP service handler new NettyHttpServerConnection(streamConnection, service, HTTP_2_0, h2ServerConfig.headersFactory(), - config.allowDropTrailersReadFromTransport()).process(false); + config.allowDropTrailersReadFromTransport(), null).process(false); } }).init(channel); } catch (Throwable cause) { @@ -221,8 +254,9 @@ private static final class DefaultH2ServerParentConnection extends AbstractH2Par final Subscriber subscriber, final DelayedCancellable delayedCancellable, final boolean waitForSslHandshake, - final ConnectionObserver observer) { - super(parentContext, delayedCancellable, waitForSslHandshake, observer); + final ConnectionObserver observer, + final boolean deferAutoRead) { + super(parentContext, delayedCancellable, waitForSslHandshake, observer, deferAutoRead); this.subscriber = requireNonNull(subscriber); } @@ -231,7 +265,9 @@ void tryCompleteSubscriber() { if (subscriber != null) { Subscriber subscriberCopy = subscriber; subscriber = null; - multiplexedObserver = observer.multiplexedConnectionEstablished(parentContext); + // multiplexedObserver will be set via notifyConnectionEstablishedAndEnableAutoRead() in + // connectionConsumer, + // after all connection acceptors pass. Safe default: NoopMultiplexedObserver.INSTANCE subscriberCopy.onSuccess((H2ServerParentConnectionContext) parentContext); } } diff --git a/servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/NettyHttpServer.java b/servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/NettyHttpServer.java index f4e732d250..8e92745325 100644 --- a/servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/NettyHttpServer.java +++ b/servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/NettyHttpServer.java @@ -63,7 +63,6 @@ import io.servicetalk.transport.netty.internal.FlushStrategy; import io.servicetalk.transport.netty.internal.FlushStrategyHolder; import io.servicetalk.transport.netty.internal.InfluencerConnectionAcceptor; -import io.servicetalk.transport.netty.internal.NettyConnection; import io.servicetalk.transport.netty.internal.NettyConnectionContext; import io.servicetalk.transport.netty.internal.NettyConnectionContext.FlushStrategyProvider; @@ -144,7 +143,7 @@ static Single bind(final HttpExecutionContext executionContex (channel, connectionObserver) -> initChannel(channel, executionContext, config, new TcpServerChannelInitializer(tcpServerConfig, connectionObserver, executionContext), service, connectionObserver), - serverConnection -> serverConnection.process(true), + serverConnection -> serverConnection.notifyConnectionEstablishedAndProcess(), earlyConnectionAcceptor, lateConnectionAcceptor) .map(delegate -> { LOGGER.debug("Started HTTP/1.1 server for address {}.", delegate.listenAddress()); @@ -185,10 +184,10 @@ private static Single initChannel(final Channel chann closeHandler, tcpConfig.flushStrategy(), tcpConfig.idleTimeoutMs(), tcpConfig.sslConfig(), initializer.andThen(getChannelInitializer( getByteBufAllocator(builderExecutionContext.bufferAllocator()), h1Config, closeHandler)), - HTTP_1_1, observer, false, __ -> false) + HTTP_1_1, observer, false, __ -> false, true) .map(conn -> new NettyHttpServerConnection(conn, service, - HTTP_1_1, h1Config.headersFactory(), - config.allowDropTrailersReadFromTransport())), + HTTP_1_1, h1Config.headersFactory(), + config.allowDropTrailersReadFromTransport(), observer)), HTTP_1_1, channel); } @@ -265,17 +264,20 @@ public String toString() { static final class NettyHttpServerConnection extends HttpServiceContext implements NettyConnectionContext { private final StreamingHttpService service; - private final NettyConnection connection; + private final DefaultNettyConnection connection; private final HttpHeadersFactory headersFactory; private final HttpExecutionContext executionContext; private final ChangingFlushStrategy flushStrategy; private final boolean requireTrailerHeader; + @Nullable + private ConnectionObserver observer; - NettyHttpServerConnection(final NettyConnection connection, + NettyHttpServerConnection(final DefaultNettyConnection connection, final StreamingHttpService service, final HttpProtocolVersion version, final HttpHeadersFactory headersFactory, - final boolean requireTrailerHeader) { + final boolean requireTrailerHeader, + @Nullable final ConnectionObserver observer) { super(headersFactory, new DefaultHttpResponseFactory(headersFactory, connection.executionContext().bufferAllocator(), version), @@ -292,6 +294,7 @@ static final class NettyHttpServerConnection extends HttpServiceContext implemen this.flushStrategy = new ChangingFlushStrategy(new FlushStrategyHolder(connection.defaultFlushStrategy())); connection.updateFlushStrategy((current, isCurrentOriginal) -> flushStrategy); this.requireTrailerHeader = requireTrailerHeader; + this.observer = observer; } void process(final boolean handleMultipleRequests) { @@ -309,6 +312,14 @@ void process(final boolean handleMultipleRequests) { .subscribe(new ErrorLoggingHttpSubscriber(this)); } + void notifyConnectionEstablishedAndProcess() { + if (observer != null) { + connection.notifyConnectionEstablished(observer); + observer = null; + } + process(true); + } + @Override public Cancellable updateFlushStrategy(final FlushStrategyProvider strategyProvider) { return flushStrategy.updateFlushStrategy(strategyProvider); diff --git a/servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/OptionalSslNegotiator.java b/servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/OptionalSslNegotiator.java index cc7028d602..3b88645552 100644 --- a/servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/OptionalSslNegotiator.java +++ b/servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/OptionalSslNegotiator.java @@ -102,7 +102,14 @@ static Single bind(final HttpExecutionContext executionContex final Consumer connectionConsumer = serverConnection -> { if (serverConnection instanceof NettyHttpServer.NettyHttpServerConnection) { - ((NettyHttpServer.NettyHttpServerConnection) serverConnection).process(true); + ((NettyHttpServer.NettyHttpServerConnection) serverConnection) + .notifyConnectionEstablishedAndProcess(); + } else if (serverConnection instanceof H2ServerParentConnectionContext) { + ((H2ServerParentConnectionContext) serverConnection) + .notifyConnectionEstablishedAndEnableAutoRead(); + } else { + throw new IllegalStateException("Unexpected connection type: " + + serverConnection.getClass().getName()); } }; diff --git a/servicetalk-http-netty/src/test/java/io/servicetalk/http/netty/EarlyAndLateConnectionAcceptorTest.java b/servicetalk-http-netty/src/test/java/io/servicetalk/http/netty/EarlyAndLateConnectionAcceptorTest.java index ce57648ff1..75bdb2677e 100644 --- a/servicetalk-http-netty/src/test/java/io/servicetalk/http/netty/EarlyAndLateConnectionAcceptorTest.java +++ b/servicetalk-http-netty/src/test/java/io/servicetalk/http/netty/EarlyAndLateConnectionAcceptorTest.java @@ -47,6 +47,11 @@ import java.net.InetSocketAddress; import java.util.Queue; import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; @@ -344,6 +349,111 @@ void lateConnectionAcceptorCanReject() throws Exception { } } + /** + * Verifies that the HTTP service is NOT invoked when a {@link LateConnectionAcceptor} rejects + * for HTTP/2 prior-knowledge (no TLS). Previously, auto-read was enabled before acceptors + * completed, allowing requests to reach the service before rejection. + *

+ * Note: TLS-based protocols (H2_TLS, H2_ALPN) are excluded because with TLS the SSL handshake + * read may also deliver application data (H2 preface) in the same {@code channelRead} call that + * completes the handshake. This is a pre-existing issue unrelated to the auto-read fix — our + * fix prevents new reads via auto-read, but cannot prevent data already decoded by the + * SslHandler during the handshake read cycle. + */ + @Test + void h2ServiceNotInvokedWhenLateAcceptorRejects() throws Exception { + final AtomicBoolean serviceInvoked = new AtomicBoolean(false); + + HttpServerBuilder builder = serverBuilder() + .protocols(h2Default()) + .appendLateConnectionAcceptor(info -> Completable.failed(DELIBERATE_EXCEPTION)); + + final HttpService service = (ctx, request, responseFactory) -> { + serviceInvoked.set(true); + return succeeded(responseFactory.ok().payloadBody("Hello World!", textSerializerUtf8())); + }; + try (ServerContext server = builder.listenAndAwait(service)) { + try (BlockingHttpClient client = HttpClients.forSingleAddress(serverHostAndPort(server)) + .protocols(h2Default()).buildBlocking()) { + assertThrows(Exception.class, () -> client.request(client.get("/sayHello"))); + } + } + + assertThat("Service should not be invoked when late acceptor rejects", + serviceInvoked.get(), is(false)); + } + + /** + * Verifies that auto-read is not enabled before the late acceptor completes for HTTP/2 prior-knowledge. + * The late acceptor blocks, verifying the server does not process frames prematurely. + */ + @Test + void h2NoFramesSentBeforeLateAcceptorCompletes() throws Exception { + final CountDownLatch acceptorStarted = new CountDownLatch(1); + final CountDownLatch acceptorRelease = new CountDownLatch(1); + final AtomicBoolean serviceInvoked = new AtomicBoolean(false); + + HttpServerBuilder builder = serverBuilder() + .protocols(h2Default()) + .appendLateConnectionAcceptor(new LateConnectionAcceptor() { + @Override + public Completable accept(final ConnectionInfo info) { + return accept((ConnectionContext) info); + } + + @Override + public Completable accept(final ConnectionContext context) { + return Completable.defer(() -> { + acceptorStarted.countDown(); + try { + acceptorRelease.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return Completable.completed(); + }); + } + + @Override + public ConnectExecutionStrategy requiredOffloads() { + return ConnectExecutionStrategy.offloadAll(); + } + }); + + final HttpService service = (ctx, request, responseFactory) -> { + serviceInvoked.set(true); + return succeeded(responseFactory.ok().payloadBody("Hello World!", textSerializerUtf8())); + }; + try (ServerContext server = builder.listenAndAwait(service)) { + try (BlockingHttpClient client = HttpClients.forSingleAddress(serverHostAndPort(server)) + .protocols(h2Default()).buildBlocking()) { + // Send request in a background thread — the late acceptor will block + final ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + final Future responseFuture = executor.submit( + () -> client.request(client.get("/sayHello"))); + + // Wait for the acceptor to start executing + acceptorStarted.await(); + + // While the acceptor is blocking, service should NOT have been invoked + assertThat("Service should not be invoked while late acceptor is blocking", + serviceInvoked.get(), is(false)); + + // Release the acceptor to let the connection complete + acceptorRelease.countDown(); + + // Verify the request completed successfully + HttpResponse response = responseFuture.get(); + assertNotNull(response, "No response received"); + assertThat(response.status(), is(HttpResponseStatus.OK)); + } finally { + executor.shutdownNow(); + } + } + } + } + private static HttpServerBuilder serverBuilder() { return HttpServers.forAddress(localAddress(0)); } diff --git a/servicetalk-http-netty/src/test/java/io/servicetalk/http/netty/HttpTransportObserverConnectionAcceptorTest.java b/servicetalk-http-netty/src/test/java/io/servicetalk/http/netty/HttpTransportObserverConnectionAcceptorTest.java new file mode 100644 index 0000000000..3b3fd40f21 --- /dev/null +++ b/servicetalk-http-netty/src/test/java/io/servicetalk/http/netty/HttpTransportObserverConnectionAcceptorTest.java @@ -0,0 +1,229 @@ +/* + * Copyright © 2026 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.http.netty; + +import io.servicetalk.concurrent.api.Completable; +import io.servicetalk.http.api.BlockingHttpClient; +import io.servicetalk.http.api.HttpServerBuilder; +import io.servicetalk.http.api.SingleAddressHttpClientBuilder; +import io.servicetalk.test.resources.DefaultTestCerts; +import io.servicetalk.transport.api.ClientSslConfig; +import io.servicetalk.transport.api.ClientSslConfigBuilder; +import io.servicetalk.transport.api.ConnectionInfo; +import io.servicetalk.transport.api.ConnectionObserver; +import io.servicetalk.transport.api.ConnectionObserver.DataObserver; +import io.servicetalk.transport.api.ConnectionObserver.MultiplexedObserver; +import io.servicetalk.transport.api.ConnectionObserver.ReadObserver; +import io.servicetalk.transport.api.ConnectionObserver.StreamObserver; +import io.servicetalk.transport.api.ConnectionObserver.WriteObserver; +import io.servicetalk.transport.api.HostAndPort; +import io.servicetalk.transport.api.ServerContext; +import io.servicetalk.transport.api.ServerSslConfig; +import io.servicetalk.transport.api.ServerSslConfigBuilder; +import io.servicetalk.transport.api.TransportObserver; + +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; +import org.mockito.Mockito; +import org.mockito.verification.VerificationWithTimeout; + +import java.net.InetSocketAddress; +import java.util.concurrent.atomic.AtomicBoolean; + +import static io.servicetalk.concurrent.api.Single.succeeded; +import static io.servicetalk.concurrent.internal.DeliberateException.DELIBERATE_EXCEPTION; +import static io.servicetalk.http.api.HttpResponseStatus.OK; +import static io.servicetalk.http.api.HttpSerializers.textSerializerUtf8; +import static io.servicetalk.http.netty.HttpProtocolConfigs.h1Default; +import static io.servicetalk.http.netty.HttpProtocolConfigs.h2Default; +import static io.servicetalk.test.resources.DefaultTestCerts.serverPemHostname; +import static io.servicetalk.transport.netty.internal.AddressUtils.localAddress; +import static io.servicetalk.transport.netty.internal.AddressUtils.serverHostAndPort; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.is; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.lenient; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * Tests that verify {@link ConnectionObserver#connectionEstablished} and + * {@link ConnectionObserver#multiplexedConnectionEstablished} callbacks fire at the correct time + * relative to connection acceptors. + */ +class HttpTransportObserverConnectionAcceptorTest { + + /** + * Protocol configurations covering all distinct server bind paths. + */ + private enum Protocol { + /** HTTP/1.1 plain — uses {@link NettyHttpServer#bind}. */ + H1, + /** HTTP/2 prior-knowledge — uses {@link H2ServerParentConnectionContext#bind}. */ + H2, + /** HTTP/2 with ALPN — uses {@link DeferredServerChannelBinder#bind}. */ + H2_ALPN + } + + private TransportObserver svrTransportObserver; + private ConnectionObserver svrConnectionObserver; + private DataObserver svrDataObserver; + private MultiplexedObserver svrMultiplexedObserver; + + private void setUpMocks() { + svrTransportObserver = mock(TransportObserver.class, "svrTransportObserver"); + svrConnectionObserver = mock(ConnectionObserver.class, "svrConnectionObserver"); + svrDataObserver = mock(DataObserver.class, "svrDataObserver"); + svrMultiplexedObserver = mock(MultiplexedObserver.class, "svrMultiplexedObserver"); + when(svrTransportObserver.onNewConnection(any(), any())).thenReturn(svrConnectionObserver); + lenient().when(svrConnectionObserver.connectionEstablished(any(ConnectionInfo.class))) + .thenReturn(svrDataObserver); + lenient().when(svrConnectionObserver.multiplexedConnectionEstablished(any(ConnectionInfo.class))) + .thenReturn(svrMultiplexedObserver); + } + + private HttpServerBuilder configureServer(Protocol protocol) { + HttpServerBuilder builder = HttpServers.forAddress(localAddress(0)) + .transportObserver(svrTransportObserver); + switch (protocol) { + case H1: + builder.protocols(h1Default()); + break; + case H2: + builder.protocols(h2Default()); + break; + case H2_ALPN: + ServerSslConfig serverSslConfig = new ServerSslConfigBuilder( + DefaultTestCerts::loadServerPem, DefaultTestCerts::loadServerKey).build(); + builder.protocols(h2Default(), h1Default()).sslConfig(serverSslConfig); + break; + default: + throw new IllegalArgumentException("Unsupported protocol: " + protocol); + } + return builder; + } + + private BlockingHttpClient configureClient(Protocol protocol, ServerContext server) { + final SingleAddressHttpClientBuilder clientBuilder = + HttpClients.forSingleAddress(serverHostAndPort(server)); + switch (protocol) { + case H1: + clientBuilder.protocols(h1Default()); + break; + case H2: + clientBuilder.protocols(h2Default()); + break; + case H2_ALPN: + ClientSslConfig clientSslConfig = new ClientSslConfigBuilder(DefaultTestCerts::loadServerCAPem) + .peerHost(serverPemHostname()).build(); + clientBuilder.protocols(h2Default(), h1Default()).sslConfig(clientSslConfig); + break; + default: + throw new IllegalArgumentException("Unsupported protocol: " + protocol); + } + return clientBuilder.buildBlocking(); + } + + @ParameterizedTest(name = "{displayName} [{index}] protocol={0}") + @EnumSource(Protocol.class) + void connectionEstablishedNotCalledWhenAcceptorRejects(Protocol protocol) throws Exception { + setUpMocks(); + HttpServerBuilder serverBuilder = configureServer(protocol) + .appendLateConnectionAcceptor(info -> Completable.failed(DELIBERATE_EXCEPTION)); + + try (ServerContext server = serverBuilder.listenAndAwait((ctx, req, factory) -> + succeeded(factory.ok().payloadBody("Hello", textSerializerUtf8())))) { + try (BlockingHttpClient client = configureClient(protocol, server)) { + assertThrows(Exception.class, () -> client.request(client.get("/"))); + } + } + + verify(svrTransportObserver, await()).onNewConnection(any(), any()); + verify(svrConnectionObserver, never()).connectionEstablished(any(ConnectionInfo.class)); + verify(svrConnectionObserver, never()).multiplexedConnectionEstablished(any(ConnectionInfo.class)); + verify(svrConnectionObserver, await()).connectionClosed(DELIBERATE_EXCEPTION); + } + + @ParameterizedTest(name = "{displayName} [{index}] protocol={0}") + @EnumSource(Protocol.class) + void connectionEstablishedNotCalledWhenEarlyAcceptorRejects(Protocol protocol) throws Exception { + setUpMocks(); + HttpServerBuilder serverBuilder = configureServer(protocol) + .appendEarlyConnectionAcceptor(info -> Completable.failed(DELIBERATE_EXCEPTION)); + + try (ServerContext server = serverBuilder.listenAndAwait((ctx, req, factory) -> + succeeded(factory.ok().payloadBody("Hello", textSerializerUtf8())))) { + try (BlockingHttpClient client = configureClient(protocol, server)) { + assertThrows(Exception.class, () -> client.request(client.get("/"))); + } + } + + verify(svrTransportObserver, await()).onNewConnection(any(), any()); + verify(svrConnectionObserver, never()).connectionEstablished(any(ConnectionInfo.class)); + verify(svrConnectionObserver, never()).multiplexedConnectionEstablished(any(ConnectionInfo.class)); + // Note: connectionClosed is NOT verified here because when the early acceptor rejects, + // the connectionFunction never executes, so the ConnectionObserverHandler is never added + // to the pipeline and the observer never receives a connectionClosed callback. + } + + @ParameterizedTest(name = "{displayName} [{index}] protocol={0}") + @EnumSource(Protocol.class) + void connectionEstablishedAfterLateAcceptor(Protocol protocol) throws Exception { + final AtomicBoolean lateAcceptorCompleted = new AtomicBoolean(false); + final AtomicBoolean establishedAfterAcceptor = new AtomicBoolean(false); + + setUpMocks(); + // Override default mock setup to track ordering + StreamObserver svrStreamObserver = mock(StreamObserver.class, "svrStreamObserver"); + when(svrConnectionObserver.connectionEstablished(any(ConnectionInfo.class))) + .thenAnswer(invocation -> { + establishedAfterAcceptor.set(lateAcceptorCompleted.get()); + return svrDataObserver; + }); + when(svrConnectionObserver.multiplexedConnectionEstablished(any(ConnectionInfo.class))) + .thenAnswer(invocation -> { + establishedAfterAcceptor.set(lateAcceptorCompleted.get()); + return svrMultiplexedObserver; + }); + lenient().when(svrMultiplexedObserver.onNewStream()).thenReturn(svrStreamObserver); + lenient().when(svrStreamObserver.streamEstablished()).thenReturn(svrDataObserver); + lenient().when(svrDataObserver.onNewRead()).thenReturn(mock(ReadObserver.class)); + lenient().when(svrDataObserver.onNewWrite()).thenReturn(mock(WriteObserver.class)); + + HttpServerBuilder serverBuilder = configureServer(protocol) + .appendLateConnectionAcceptor(info -> { + lateAcceptorCompleted.set(true); + return Completable.completed(); + }); + + try (ServerContext server = serverBuilder.listenAndAwait((ctx, req, factory) -> + succeeded(factory.ok().payloadBody("Hello", textSerializerUtf8())))) { + try (BlockingHttpClient client = configureClient(protocol, server)) { + assertThat(client.request(client.get("/")).status(), is(OK)); + } + } + + assertThat("connectionEstablished should fire after late acceptor", + establishedAfterAcceptor.get(), is(true)); + } + + static VerificationWithTimeout await() { + return Mockito.timeout(Long.MAX_VALUE); + } +} diff --git a/servicetalk-transport-netty-internal/src/main/java/io/servicetalk/transport/netty/internal/DefaultNettyConnection.java b/servicetalk-transport-netty-internal/src/main/java/io/servicetalk/transport/netty/internal/DefaultNettyConnection.java index ad859a8817..9677536db6 100644 --- a/servicetalk-transport-netty-internal/src/main/java/io/servicetalk/transport/netty/internal/DefaultNettyConnection.java +++ b/servicetalk-transport-netty-internal/src/main/java/io/servicetalk/transport/netty/internal/DefaultNettyConnection.java @@ -30,6 +30,7 @@ import io.servicetalk.concurrent.api.internal.SubscribableSingle; import io.servicetalk.concurrent.internal.DelayedCancellable; import io.servicetalk.transport.api.ConnectionContext; +import io.servicetalk.transport.api.ConnectionInfo; import io.servicetalk.transport.api.ConnectionObserver; import io.servicetalk.transport.api.ConnectionObserver.DataObserver; import io.servicetalk.transport.api.ConnectionObserver.ReadObserver; @@ -145,6 +146,7 @@ public final class DefaultNettyConnection extends NettyChannelListe private final ChannelConfig parentChannelConfig; private volatile DataObserver dataObserver; private final boolean isClient; + private final boolean deferConnectionEstablished; private final Predicate shouldWait; private final UnaryOperator enrichProtocolError; private final TerminalSignalConsumer cleanupStateConsumer = new TerminalSignalConsumer() { @@ -176,7 +178,8 @@ private DefaultNettyConnection( long idleTimeoutMs, Protocol protocol, @Nullable SslConfig sslConfig, @Nullable SSLSession sslSession, @Nullable ChannelConfig parentChannelConfig, DataObserver dataObserver, boolean isClient, - Predicate shouldWait, UnaryOperator enrichProtocolError) { + Predicate shouldWait, UnaryOperator enrichProtocolError, + boolean deferConnectionEstablished) { super(channel, executionContext.executionStrategy().isCloseOffloaded() ? executionContext.executor() : immediate()); nettyChannelPublisher = new NettyChannelPublisher<>(channel, closeHandler); @@ -203,10 +206,20 @@ private DefaultNettyConnection( this.protocol = requireNonNull(protocol); this.dataObserver = dataObserver; this.isClient = isClient; + this.deferConnectionEstablished = deferConnectionEstablished; this.shouldWait = requireNonNull(shouldWait); this.enrichProtocolError = requireNonNull(enrichProtocolError); } + /** + * Notifies the observer that the connection has been established and sets the {@link DataObserver}. + * + * @param observer the {@link ConnectionObserver} to notify. + */ + public void notifyConnectionEstablished(final ConnectionObserver observer) { + this.dataObserver = observer.connectionEstablished(this); + } + /** * Given a {@link Channel} this will initialize the {@link ChannelPipeline} just to create a * {@link DefaultNettyConnection}. It is assumed this is a child channel and all TLS handshaking is completed. @@ -354,7 +367,7 @@ private static DefaultNettyConnection initChildChanne DefaultNettyConnection connection = new DefaultNettyConnection<>(channel, parent, executionContext, closeHandler, flushStrategy, idleTimeoutMs, protocol, sslConfig, sslSession, parentChannelConfig, - streamObserver.streamEstablished(), isClient, shouldWait, enrichProtocolError); + streamObserver.streamEstablished(), isClient, shouldWait, enrichProtocolError, false); channel.pipeline().addLast(new NettyToStChannelHandler<>(connection, null, null, false, NoopConnectionObserver.INSTANCE)); return connection; @@ -493,6 +506,41 @@ public static Single> initChan FlushStrategy flushStrategy, long idleTimeoutMs, @Nullable SslConfig sslConfig, ChannelInitializer initializer, Protocol protocol, ConnectionObserver observer, boolean isClient, Predicate shouldWait) { + return initChannel(channel, executionContext, closeHandler, flushStrategy, idleTimeoutMs, sslConfig, + initializer, protocol, observer, isClient, shouldWait, false); + } + + /** + * Given a {@link Channel} this will initialize the {@link ChannelPipeline} and create a + * {@link DefaultNettyConnection}. The resulting single will complete after the TLS handshake has completed + * (if applicable) or otherwise after the channel is active and ready to use. + * @param channel A newly created {@link Channel}. + * @param executionContext The {@link ExecutionContext} to use for the {@link DefaultNettyConnection}. Note: + * {@link ExecutionContext#ioExecutor()} must be backed by a single {@link EventLoop} thread identical to + * {@link Channel#eventLoop()} for the specified channel. + * @param closeHandler Manages the half closure of the {@link DefaultNettyConnection}. + * @param flushStrategy Manages flushing of data for the {@link DefaultNettyConnection}. + * @param idleTimeoutMs Value for {@link ServiceTalkSocketOptions#IDLE_TIMEOUT IDLE_TIMEOUT} socket option. + * @param sslConfig The {@link SslConfig} to use for the {@link DefaultNettyConnection}. + * @param initializer Synchronously initializes the pipeline upon subscribe. + * @param protocol {@link Protocol} for the returned {@link DefaultNettyConnection}. + * @param observer {@link ConnectionObserver} to report network events. + * @param isClient tells if this {@link Channel} is for the client. + * @param shouldWait predicate that tells when request payload body should wait for continuation signal. + * @param deferConnectionEstablished if {@code true}, the + * {@link ConnectionObserver#connectionEstablished(ConnectionInfo)} callback will not be + * invoked when the connection Single resolves. Instead, the caller is responsible for calling + * {@link #notifyConnectionEstablished(ConnectionObserver)} after all connection acceptors have completed. + * @param Type of objects read from the {@link NettyConnection}. + * @param Type of objects written to the {@link NettyConnection}. + * @return A {@link Single} that completes with a {@link DefaultNettyConnection} after the channel is activated and + * ready to use. + */ + public static Single> initChannel( + Channel channel, ExecutionContext executionContext, CloseHandler closeHandler, + FlushStrategy flushStrategy, long idleTimeoutMs, @Nullable SslConfig sslConfig, + ChannelInitializer initializer, Protocol protocol, ConnectionObserver observer, boolean isClient, + Predicate shouldWait, boolean deferConnectionEstablished) { assert channel.eventLoop() == toEventLoopAwareNettyIoExecutor(executionContext.ioExecutor()).eventLoopGroup(); return new SubscribableSingle>() { @Override @@ -515,7 +563,8 @@ protected void handleSubscribe( final SSLSession sslSession = extractSslSession(sslConfig, pipeline); DefaultNettyConnection connection = new DefaultNettyConnection<>(channel, null, executionContext, closeHandler, flushStrategy, idleTimeoutMs, protocol, sslConfig, - sslSession, null, NoopDataObserver.INSTANCE, isClient, shouldWait, identity()); + sslSession, null, NoopDataObserver.INSTANCE, isClient, shouldWait, identity(), + deferConnectionEstablished); channel.attr(CHANNEL_CLOSEABLE_KEY).set(connection); delayedCancellable = new DelayedCancellable(); nettyInboundHandler = new NettyToStChannelHandler<>(connection, subscriber, @@ -1020,9 +1069,11 @@ private void completeSubscriber() { assert subscriber != null; SingleSource.Subscriber> subscriberCopy = subscriber; subscriber = null; - // TODO: how can we make sure we have the correct context information here. - // See HttpTransportObserverAsyncContextTest. It has some assertions for 'broken' behavior. - connection.dataObserver = observer.connectionEstablished(connection); + if (!connection.deferConnectionEstablished) { + // TODO: how can we make sure we have the correct context information here. + // See HttpTransportObserverAsyncContextTest. It has some assertions for 'broken' behavior. + connection.notifyConnectionEstablished(observer); + } subscriberCopy.onSuccess(connection); }