diff --git a/java/src/org/openqa/selenium/netty/server/TcpTunnelHandler.java b/java/src/org/openqa/selenium/netty/server/TcpTunnelHandler.java index b87b94fd9e95f..7599325845388 100644 --- a/java/src/org/openqa/selenium/netty/server/TcpTunnelHandler.java +++ b/java/src/org/openqa/selenium/netty/server/TcpTunnelHandler.java @@ -26,6 +26,10 @@ /** * Forwards every inbound {@link io.netty.buffer.ByteBuf} to a target {@link Channel}. Used on both * ends of a transparent TCP tunnel once the WebSocket upgrade handshake has been proxied. + * + *

Backpressure is mirrored across the tunnel: when {@code ctx.channel()}'s outbound buffer + * passes its high-water mark, {@code target}'s read side is paused so the peer stops shipping bytes + * the kernel cannot drain. When the buffer drains below the low-water mark the peer is resumed. */ class TcpTunnelHandler extends ChannelInboundHandlerAdapter { @@ -58,6 +62,18 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) { }); } + @Override + public void channelWritabilityChanged(ChannelHandlerContext ctx) { + // Mirror our outbound writability onto the peer's read side. While our buffer is full the + // peer stops reading; when it drains the peer resumes. Symmetric on both legs of the tunnel. + // ChannelConfig.setAutoRead is documented thread-safe: the flag is updated via an atomic + // updater, and the channel.read() that fires on a false→true transition marshalls itself + // onto the channel's event loop. Safe to call directly even if the two tunnel legs are on + // different loops (today TcpUpgradeTunnelHandler places them on the same one). + target.config().setAutoRead(ctx.channel().isWritable()); + ctx.fireChannelWritabilityChanged(); + } + @Override public void channelInactive(ChannelHandlerContext ctx) { target.close(); diff --git a/java/test/org/openqa/selenium/netty/server/BUILD.bazel b/java/test/org/openqa/selenium/netty/server/BUILD.bazel index ab63f9446c736..893d8bfb7ecb2 100644 --- a/java/test/org/openqa/selenium/netty/server/BUILD.bazel +++ b/java/test/org/openqa/selenium/netty/server/BUILD.bazel @@ -5,6 +5,7 @@ SMALL_TEST_SRCS = [ "MessageInboundConverterTest.java", "MessageOutboundConverterTest.java", "RequestConverterTest.java", + "TcpTunnelHandlerTest.java", "WebSocketFrameProxyTest.java", "WebSocketUpgradeHandlerTest.java", ] diff --git a/java/test/org/openqa/selenium/netty/server/TcpTunnelHandlerTest.java b/java/test/org/openqa/selenium/netty/server/TcpTunnelHandlerTest.java new file mode 100644 index 0000000000000..6dc80c7cc155e --- /dev/null +++ b/java/test/org/openqa/selenium/netty/server/TcpTunnelHandlerTest.java @@ -0,0 +1,58 @@ +// Licensed to the Software Freedom Conservancy (SFC) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The SFC licenses this file +// to you 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 org.openqa.selenium.netty.server; + +import static org.assertj.core.api.Assertions.assertThat; + +import io.netty.buffer.Unpooled; +import io.netty.channel.WriteBufferWaterMark; +import io.netty.channel.embedded.EmbeddedChannel; +import org.junit.jupiter.api.Test; + +class TcpTunnelHandlerTest { + + @Test + void writabilityChangesOnTunnelMirrorToPeerAutoRead() { + // Source provides bytes; tunnel handler writes them to the slow-draining target. + EmbeddedChannel target = new EmbeddedChannel(); + EmbeddedChannel source = new EmbeddedChannel(new TcpTunnelHandler(target)); + + // Force the target to flip writability after a single small write. + target.config().setWriteBufferWaterMark(new WriteBufferWaterMark(8, 16)); + + // Don't let EmbeddedChannel's outbound queue auto-drain; we want to observe the watermark. + assertThat(source.config().isAutoRead()).isTrue(); + + // A write that exceeds the high-water mark on the *source* channel triggers the + // channelWritabilityChanged event, which the handler mirrors onto target.setAutoRead(). + source.config().setWriteBufferWaterMark(new WriteBufferWaterMark(8, 16)); + source.write(Unpooled.wrappedBuffer(new byte[64])); + + // Source is now unwritable -> target's autoRead must be paused. + assertThat(source.isWritable()).isFalse(); + assertThat(target.config().isAutoRead()).isFalse(); + + // Drain source; writability flips back and target's autoRead must be re-enabled. + source.flushOutbound(); + source.releaseOutbound(); + assertThat(source.isWritable()).isTrue(); + assertThat(target.config().isAutoRead()).isTrue(); + + target.releaseInbound(); + } +}