Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -1027,10 +1027,21 @@ protected void initChannel(Channel channel) throws Exception {
}

/**
* Checks whether the given channel is an HTTP/2 connection (i.e. has the HTTP/2 multiplex handler installed).
* Checks whether the given channel is an HTTP/2 connection: the parent that multiplexes streams, not one of
* its stream children, which carry neither the multiplex handler nor connection state of their own.
* <p>
* Answered from the {@link Http2ConnectionState} attached to the connection rather than by looking
* {@link #HTTP2_MULTIPLEX} up in the pipeline. The two are attached together, in
* {@link #upgradePipelineToHttp2}, and agree for as long as the connection is live; a pipeline lookup,
* though, compares handler names down the chain, and an HTTP/1.1 connection, which has no such handler, is
* walked to the end to say no.
* <p>
* They part on close, where Netty's own teardown strips the pipeline and leaves the attribute: a closed
* connection answers yes here and would have answered no to a lookup. Callers reach this while deciding
* what to do with a channel they have just taken from the pool, having checked it is active.
*/
public static boolean isHttp2(Channel channel) {
return channel.pipeline().get(HTTP2_MULTIPLEX) != null;
return channel.attr(Http2ConnectionState.HTTP2_STATE_KEY).get() != null;
}

/**
Expand Down Expand Up @@ -1068,6 +1079,21 @@ public void upgradePipelineToHttp2(ChannelPipeline pipeline) {
pipeline.remove(AHC_HTTP_HANDLER);
}

// Attach HTTP/2 connection state for MAX_CONCURRENT_STREAMS tracking and GOAWAY draining. Its
// presence is also what marks the connection as HTTP/2; see isHttp2. Attached before the handlers
// rather than after them, so that there is no instant at which the pipeline speaks HTTP/2 and the
// connection does not yet say so -- a write landing there would take the HTTP/1.1 branch onto an
// HTTP/2 pipeline. Nothing can reach this channel that early today; the ordering is what keeps that
// from being something each new caller has to know.
Http2ConnectionState state = new Http2ConnectionState();
int configMaxStreams = config.getHttp2MaxConcurrentStreams();
if (configMaxStreams > 0) {
// Client's own cap; the server-advertised value (applied by the http2-settings-listener below)
// can only lower the effective limit, never raise it above this.
state.setClientMaxConcurrentStreams(configMaxStreams);
}
pipeline.channel().attr(Http2ConnectionState.HTTP2_STATE_KEY).set(state);

// Add HTTP/2 frame codec (handles connection preface, SETTINGS, PING, flow control, etc.)
Http2Settings settings = new Http2Settings()
.initialWindowSize(config.getHttp2InitialWindowSize())
Expand Down Expand Up @@ -1097,16 +1123,6 @@ protected void initChannel(Channel ch) {
pipeline.addLast(HTTP2_FRAME_CODEC, frameCodec);
pipeline.addLast(HTTP2_MULTIPLEX, multiplexHandler);
Comment thread
pavel-ptashyts marked this conversation as resolved.

// Attach HTTP/2 connection state for MAX_CONCURRENT_STREAMS tracking and GOAWAY draining
Http2ConnectionState state = new Http2ConnectionState();
int configMaxStreams = config.getHttp2MaxConcurrentStreams();
if (configMaxStreams > 0) {
// Client's own cap; the server-advertised value (applied by the http2-settings-listener below)
// can only lower the effective limit, never raise it above this.
state.setClientMaxConcurrentStreams(configMaxStreams);
}
pipeline.channel().attr(Http2ConnectionState.HTTP2_STATE_KEY).set(state);

// Install SETTINGS listener to update MAX_CONCURRENT_STREAMS from server
pipeline.addLast("http2-settings-listener", new ChannelInboundHandlerAdapter() {
@Override
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -41,9 +41,9 @@ public boolean exitAfterHandling100(final Channel channel, final NettyResponseFu

if (channel instanceof Http2StreamChannel) {
// HTTP/2: the HEADERS frame was already sent with endStream=false; now send the deferred body
// as DATA frame(s). writeRequest() can't be reused here — its isHttp2() check looks for the
// parent connection's multiplex handler, which a stream child channel doesn't have, so it would
// mis-route to the HTTP/1.1 writer (UnsupportedMessageTypeException + use-after-free).
// as DATA frame(s). writeRequest() can't be reused here — it routes on the parent connection's
// HTTP/2 state, which a stream child channel doesn't carry, so it would mis-route to the
// HTTP/1.1 writer (UnsupportedMessageTypeException + use-after-free).
//
// Only resume when the body was genuinely deferred. For a request WITHOUT Expect: 100-continue
// the body was already written with endStream=true, so writing DATA now would be a frame after
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -703,9 +703,12 @@ public <T> void writeRequest(NettyResponseFuture<T> future, Channel channel) {
return;
}

// Route to HTTP/2 path if the parent channel has the HTTP/2 multiplex handler installed
if (ChannelManager.isHttp2(channel)) {
writeHttp2Request(future, channel);
// Route to HTTP/2 when the connection carries HTTP/2 state, which is attached where the multiplex
// handler is. Read here rather than asked of ChannelManager.isHttp2, because the HTTP/2 path needs the
// state itself and would otherwise look up what this line has already found.
Http2ConnectionState http2State = channel.attr(Http2ConnectionState.HTTP2_STATE_KEY).get();
if (http2State != null) {
writeHttp2Request(future, channel, http2State);
Comment thread
pavel-ptashyts marked this conversation as resolved.
return;
}

Expand Down Expand Up @@ -771,11 +774,12 @@ public <T> void writeRequest(NettyResponseFuture<T> future, Channel channel) {
* as HTTP/2 frames ({@link DefaultHttp2HeadersFrame} + optional {@link DefaultHttp2DataFrame}).
* The stream child channel has the {@link org.asynchttpclient.netty.handler.Http2Handler} installed
* and the {@link NettyResponseFuture} attached to it, mirroring the HTTP/1.1 channel model.
*
* @param state the connection's HTTP/2 state, which is what identified it as an HTTP/2 connection in the
* first place, so the caller has it in hand
*/
private <T> void writeHttp2Request(NettyResponseFuture<T> future, Channel parentChannel) {
Http2ConnectionState state = parentChannel.attr(Http2ConnectionState.HTTP2_STATE_KEY).get();

if (state != null && !state.tryAcquireStream()) {
private <T> void writeHttp2Request(NettyResponseFuture<T> future, Channel parentChannel, Http2ConnectionState state) {
if (!state.tryAcquireStream()) {
if (state.isDraining()) {
// Connection is draining from GOAWAY — fail the future so it retries on a new connection.
// Don't close the parent channel since it may still have active streams. sendHttp2Frames
Expand Down Expand Up @@ -850,14 +854,12 @@ protected void initChannel(Http2StreamChannel streamCh) {
if (openedRequest != null) {
openedRequest.release();
}
if (state != null) {
state.releaseStream();
// Close the parent once it has no active streams AND it is either draining
// (GOAWAY) or a redundant duplicate (#10 thundering-herd loser) — neither
// will serve further requests, so it must not linger open.
if ((state.isDraining() || state.isRedundant()) && state.getActiveStreams() <= 0) {
channelManager.closeChannel(parentChannel);
}
state.releaseStream();
// Close the parent once it has no active streams AND it is either draining
// (GOAWAY) or a redundant duplicate (#10 thundering-herd loser) — neither
// will serve further requests, so it must not linger open.
if ((state.isDraining() || state.isRedundant()) && state.getActiveStreams() <= 0) {
channelManager.closeChannel(parentChannel);
}
});

Expand Down Expand Up @@ -896,9 +898,7 @@ protected void initChannel(Http2StreamChannel streamCh) {
} else {
// Stream channel was never opened — no closeFuture will fire, so release the
// acquired slot and the unsent request body inline.
if (state != null) {
state.releaseStream();
}
state.releaseStream();
releaseHttp2Request(future);
// Fail ONLY this future (future.abort, not abort(parentChannel, ...)): opening one stream
// can fail for a stream-local reason (e.g. Netty rejecting it as the outbound max-streams
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,113 @@
/*
* Copyright (c) 2026 AsyncHttpClient Project. All rights reserved.
*
* 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 org.asynchttpclient.netty.channel;

import io.netty.channel.Channel;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.embedded.EmbeddedChannel;
import io.netty.handler.codec.http2.Http2StreamChannelBootstrap;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timer;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

import static org.asynchttpclient.Dsl.config;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;

/**
* {@link ChannelManager#isHttp2(Channel)} answers from the {@link Http2ConnectionState} attached to a
* connection, while the thing it stands for is the multiplex handler in the pipeline. These pin the two
* together: either both say HTTP/2 or neither does, whichever way a later change to the upgrade attaches them.
*/
class ChannelManagerHttp2MarkerTest {

// One per class: the upgrade is what is under test and it needs a ChannelManager only to be called. Building
// one per test costs an SslContext and an event loop group each time, for state that lives on the channel.
private static ChannelManager channelManager;
private static Timer timer;

private EmbeddedChannel channel;

@BeforeAll
static void startManager() {
timer = new HashedWheelTimer();
channelManager = new ChannelManager(config().build(), timer);
Comment thread
pavel-ptashyts marked this conversation as resolved.
}

@AfterAll
static void stopManager() {
if (channelManager != null) {
channelManager.close();
}
if (timer != null) {
timer.stop();
}
}

@BeforeEach
void setUp() {
channel = new EmbeddedChannel();
}

@AfterEach
void tearDown() {
if (channel != null) {
channel.finishAndReleaseAll();
}
}

@Test
void aConnectionThatWasNeverUpgradedIsNotHttp2() {
assertNull(channel.pipeline().get(ChannelManager.HTTP2_MULTIPLEX),
"an untouched pipeline should not carry the multiplex handler");
assertFalse(ChannelManager.isHttp2(channel), "and should not be reported as HTTP/2");
}

@Test
void upgradingAConnectionBothInstallsTheHandlerAndReportsHttp2() {
channelManager.upgradePipelineToHttp2(channel.pipeline());

assertNotNull(channel.pipeline().get(ChannelManager.HTTP2_MULTIPLEX),
"the upgrade should install the multiplex handler");
assertTrue(ChannelManager.isHttp2(channel), "and should report the connection as HTTP/2");
}

@Test
void aStreamOfAnHttp2ConnectionIsNotTheConnection() {
// A real stream child rather than a bare channel: what is worth pinning is that a stream does not
// inherit the connection state its parent carries, since that is now what identifies an HTTP/2
// connection. The stream is where a request is written, so mistaking it for its parent would loop.
channelManager.upgradePipelineToHttp2(channel.pipeline());
channel.runPendingTasks();

Channel stream = new Http2StreamChannelBootstrap(channel)
.handler(new ChannelInboundHandlerAdapter())
.open().syncUninterruptibly().getNow();
try {
assertNull(stream.pipeline().get(ChannelManager.HTTP2_MULTIPLEX),
"a stream child carries no multiplex handler of its own");
assertFalse(ChannelManager.isHttp2(stream), "and is not the connection that multiplexes it");
} finally {
stream.close().syncUninterruptibly();
}
}
}
Loading