You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hc.apache.org by ol...@apache.org on 2025/09/13 04:59:19 UTC
(httpcomponents-core) 01/02: RFC 9113 conformance: improved H2 stream creation and initialization, improved H2 stream state tracking and tracking of total stream counts
This is an automated email from the ASF dual-hosted git repository.
olegk pushed a commit to branch rfc_9113_conformance
in repository https://gitbox.apache.org/repos/asf/httpcomponents-core.git
commit e9bd620edd01fea505c7052ad027be7a33b88744
Author: Oleg Kalnichevski <ol...@apache.org>
AuthorDate: Sun Sep 7 17:19:47 2025 +0200
RFC 9113 conformance: improved H2 stream creation and initialization, improved H2 stream state tracking and tracking of total stream counts
---
.../impl/nio/AbstractH2StreamMultiplexer.java | 167 +++++++++------------
.../http2/impl/nio/ClientH2StreamMultiplexer.java | 52 +++----
.../apache/hc/core5/http2/impl/nio/H2Stream.java | 78 ++++++++--
.../apache/hc/core5/http2/impl/nio/H2Streams.java | 86 +++++++----
.../http2/impl/nio/ServerH2StreamMultiplexer.java | 42 ++++--
.../impl/nio/TestAbstractH2StreamMultiplexer.java | 43 +++---
6 files changed, 269 insertions(+), 199 deletions(-)
diff --git a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/AbstractH2StreamMultiplexer.java b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/AbstractH2StreamMultiplexer.java
index 861bc9f87..3d35f6655 100644
--- a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/AbstractH2StreamMultiplexer.java
+++ b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/AbstractH2StreamMultiplexer.java
@@ -57,13 +57,14 @@
import org.apache.hc.core5.http.impl.BasicEndpointDetails;
import org.apache.hc.core5.http.impl.BasicHttpConnectionMetrics;
import org.apache.hc.core5.http.impl.CharCodingSupport;
+import org.apache.hc.core5.http.nio.AsyncClientExchangeHandler;
import org.apache.hc.core5.http.nio.AsyncPushConsumer;
import org.apache.hc.core5.http.nio.AsyncPushProducer;
import org.apache.hc.core5.http.nio.HandlerFactory;
import org.apache.hc.core5.http.nio.command.CommandSupport;
-import org.apache.hc.core5.http.nio.command.ExecutableCommand;
+import org.apache.hc.core5.http.nio.command.RequestExecutionCommand;
import org.apache.hc.core5.http.nio.command.ShutdownCommand;
-import org.apache.hc.core5.http.protocol.HttpCoreContext;
+import org.apache.hc.core5.http.protocol.HttpContext;
import org.apache.hc.core5.http.protocol.HttpProcessor;
import org.apache.hc.core5.http2.H2ConnectionException;
import org.apache.hc.core5.http2.H2Error;
@@ -172,6 +173,18 @@ public String getId() {
return ioSession.getId();
}
+ BasicHttpConnectionMetrics getConnMetrics() {
+ return connMetrics;
+ }
+
+ HttpProcessor getHttpProcessor() {
+ return httpProcessor;
+ }
+
+ void submitCommand(final Command command) {
+ ioSession.enqueue(command, Command.Priority.NORMAL);
+ }
+
abstract void validateSetting(H2Param param, int value) throws H2ConnectionException;
abstract H2Setting[] generateSettings(H2Config localConfig);
@@ -182,18 +195,18 @@ public String getId() {
abstract void acceptPushFrame() throws H2ConnectionException;
- abstract H2StreamHandler createRemotelyInitiatedStream(
- H2StreamChannel channel,
+ abstract H2StreamHandler incomingRequest(H2StreamChannel channel) throws IOException;
+
+ abstract H2StreamHandler incomingPushPromise(H2StreamChannel channel,
+ HandlerFactory<AsyncPushConsumer> pushHandlerFactory) throws IOException;
- HttpProcessor httpProcessor,
- BasicHttpConnectionMetrics connMetrics,
- HandlerFactory<AsyncPushConsumer> pushHandlerFactory) throws IOException;
+ abstract H2StreamHandler outgoingRequest(H2StreamChannel channel,
+ AsyncClientExchangeHandler exchangeHandler,
+ HandlerFactory<AsyncPushConsumer> pushHandlerFactory,
+ HttpContext context) throws IOException;
- abstract H2StreamHandler createLocallyInitiatedStream(
- ExecutableCommand command,
- H2StreamChannel channel,
- HttpProcessor httpProcessor,
- BasicHttpConnectionMetrics connMetrics) throws IOException;
+ abstract H2StreamHandler outgoingPushPromise(H2StreamChannel channel,
+ AsyncPushProducer pushProducer) throws IOException;
abstract boolean allowGracefulAbort(H2Stream stream);
@@ -495,11 +508,11 @@ public final void onOutput() throws HttpException, IOException {
executeShutdown((ShutdownCommand) command);
} else if (command instanceof PingCommand) {
executePing((PingCommand) command);
- } else if (command instanceof ExecutableCommand) {
- executeRequest((ExecutableCommand) command);
- if (!outputQueue.isEmpty()) {
- return;
- }
+ } else if (command instanceof RequestExecutionCommand) {
+ executeRequest((RequestExecutionCommand) command);
+ }
+ if (!outputQueue.isEmpty()) {
+ return;
}
}
}
@@ -507,8 +520,8 @@ public final void onOutput() throws HttpException, IOException {
int liveStreams = 0;
for (final Iterator<H2Stream> it = streams.iterator(); it.hasNext(); ) {
final H2Stream stream = it.next();
- if (stream.isLocalClosed() && stream.isRemoteClosed()) {
- streams.release(stream);
+ if (stream.isClosedPastLingerDeadline()) {
+ streams.dropStreamId(stream.getId());
it.remove();
} else {
if (streams.isSameSide(stream.getId()) || stream.getId() <= streams.getLastRemoteId()) {
@@ -555,10 +568,10 @@ public final void onTimeout(final Timeout timeout) throws HttpException, IOExcep
final RawFrame goAway;
if (localSettingState != SettingsHandshake.ACKED) {
goAway = frameFactory.createGoAway(streams.getLastRemoteId(), H2Error.SETTINGS_TIMEOUT,
- "Setting timeout (" + timeout + ")");
+ "Setting timeout (" + timeout + ")");
} else {
goAway = frameFactory.createGoAway(streams.getLastRemoteId(), H2Error.NO_ERROR,
- "Timeout due to inactivity (" + timeout + ")");
+ "Timeout due to inactivity (" + timeout + ")");
}
commitFrame(goAway);
for (final Iterator<H2Stream> it = streams.iterator(); it.hasNext(); ) {
@@ -601,14 +614,13 @@ private void executePing(final PingCommand pingCommand) throws IOException {
commitFrame(ping);
}
- private void executeRequest(final ExecutableCommand executableCommand) throws IOException, HttpException {
+ private void executeRequest(final RequestExecutionCommand requestExecutionCommand) throws IOException, HttpException {
final int streamId = streams.generateStreamId();
- final H2StreamChannel channel = createChannel(streamId, true);
- final H2StreamHandler streamHandler = createLocallyInitiatedStream(
- executableCommand, channel, httpProcessor, connMetrics);
-
- final H2Stream stream = new H2Stream(channel, streamHandler);
- streams.addLocallyInitiated(stream);
+ final H2StreamChannel channel = createChannel(streamId);
+ final H2Stream stream = streams.createActive(channel, outgoingRequest(channel,
+ requestExecutionCommand.getExchangeHandler(),
+ requestExecutionCommand.getPushHandlerFactory(),
+ requestExecutionCommand.getContext()));
if (streamListener != null) {
final int initInputWindow = stream.getInputWindow().get();
@@ -620,23 +632,12 @@ private void executeRequest(final ExecutableCommand executableCommand) throws IO
if (stream.isOutputReady()) {
stream.produceOutput();
}
- final CancellableDependency cancellableDependency = executableCommand.getCancellableDependency();
+ final CancellableDependency cancellableDependency = requestExecutionCommand.getCancellableDependency();
if (cancellableDependency != null) {
cancellableDependency.setDependency(stream::abort);
}
}
- public void executePush(final int promisedStreamId, final AsyncPushProducer pushProducer) {
- final H2StreamChannel channel = createChannel(promisedStreamId, true);
- final HttpCoreContext context = HttpCoreContext.create();
- context.setSSLSession(getSSLSession());
- context.setEndpointDetails(getEndpointDetails());
- final H2StreamHandler streamHandler = new ServerPushH2StreamHandler(
- channel, httpProcessor, connMetrics, pushProducer, context);
- final H2Stream stream = new H2Stream(channel, streamHandler);
- streams.addLocallyInitiated(stream);
- }
-
public final void onException(final Exception cause) {
try {
for (;;) {
@@ -702,8 +703,8 @@ private void consumeFrame(final RawFrame frame) throws HttpException, IOExceptio
stream.localReset(ex, ex.getCause() != null ? H2Error.INTERNAL_ERROR : H2Error.CANCEL);
}
- if (stream.isTerminated()) {
- streams.release(stream);
+ if (stream.isClosed()) {
+ stream.releaseResources();
requestSessionOutput();
}
}
@@ -722,26 +723,20 @@ private void consumeFrame(final RawFrame frame) throws HttpException, IOExceptio
throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "GOAWAY received");
}
- final H2StreamChannel channel = createChannel(streamId, false);
- final H2StreamHandler streamHandler;
+ final H2StreamChannel channel = createChannel(streamId);
if (connState.compareTo(ConnectionHandshake.ACTIVE) <= 0) {
- streamHandler = createRemotelyInitiatedStream(channel, httpProcessor, connMetrics, null);
+ stream = streams.createActive(channel, incomingRequest(channel));
} else {
- streamHandler = NoopH2StreamHandler.INSTANCE;
- channel.markLocalClosed();
- }
-
- stream = new H2Stream(channel, streamHandler);
- if (stream.isOutputReady()) {
- stream.produceOutput();
+ channel.localReset(H2Error.REFUSED_STREAM);
+ stream = streams.createActive(channel, NoopH2StreamHandler.INSTANCE);
}
- streams.addRemotelyInitiated(stream);
} else if (stream.isLocalClosed() && stream.isRemoteClosed()) {
throw new H2ConnectionException(H2Error.STREAM_CLOSED, "Stream closed");
+ } else if (stream.isReserved()) {
+ stream.activate();
}
try {
consumeHeaderFrame(frame, stream);
-
if (stream.isOutputReady()) {
stream.produceOutput();
}
@@ -753,8 +748,8 @@ private void consumeFrame(final RawFrame frame) throws HttpException, IOExceptio
stream.handle(ex);
}
- if (stream.isTerminated()) {
- streams.release(stream);
+ if (stream.isClosed()) {
+ stream.releaseResources();
requestSessionOutput();
}
}
@@ -779,8 +774,8 @@ private void consumeFrame(final RawFrame frame) throws HttpException, IOExceptio
stream.localReset(ex, ex.getCause() != null ? H2Error.INTERNAL_ERROR : H2Error.CANCEL);
}
- if (stream.isTerminated()) {
- streams.release(stream);
+ if (stream.isClosed()) {
+ stream.releaseResources();
requestSessionOutput();
}
}
@@ -829,7 +824,6 @@ private void consumeFrame(final RawFrame frame) throws HttpException, IOExceptio
requestSessionOutput();
} else {
stream.fail(new H2StreamResetException(errorCode, "Stream reset (" + errorCode + ")"));
- streams.release(stream);
requestSessionOutput();
}
}
@@ -900,7 +894,7 @@ private void consumeFrame(final RawFrame frame) throws HttpException, IOExceptio
throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Push is disabled");
}
- final H2Stream stream = streams.lookup(streamId);
+ final H2Stream stream = streams.lookupValid(streamId);
if (stream.isRemoteClosed()) {
stream.localReset(new H2StreamResetException(H2Error.STREAM_CLOSED, "Stream closed"));
break;
@@ -918,19 +912,14 @@ private void consumeFrame(final RawFrame frame) throws HttpException, IOExceptio
throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Stream already open: " + promisedStreamId);
}
- final H2StreamChannel channel = createChannel(promisedStreamId, false);
- final H2StreamHandler streamHandler;
+ final H2StreamChannel channel = createChannel(promisedStreamId);
+ final H2Stream promisedStream;
if (connState.compareTo(ConnectionHandshake.ACTIVE) <= 0) {
- streamHandler = createRemotelyInitiatedStream(channel, httpProcessor, connMetrics,
- stream.getPushHandlerFactory());
+ promisedStream = streams.createReserved(channel, incomingPushPromise(channel, stream.getPushHandlerFactory()));
} else {
- streamHandler = NoopH2StreamHandler.INSTANCE;
- channel.markLocalClosed();
+ channel.localReset(H2Error.REFUSED_STREAM);
+ promisedStream = streams.createActive(channel, NoopH2StreamHandler.INSTANCE);
}
-
- final H2Stream promisedStream = new H2Stream(channel, streamHandler);
- streams.addRemotelyInitiated(promisedStream);
-
try {
consumePushPromiseFrame(frame, payload, promisedStream);
} catch (final H2StreamResetException ex) {
@@ -1133,9 +1122,10 @@ private void produceOutput() throws HttpException, IOException {
if (!stream.isLocalClosed() && stream.getOutputWindow().get() > 0) {
stream.produceOutput();
}
- if (stream.isTerminated()) {
+ if (stream.isClosedPastLingerDeadline()) {
+ streams.dropStreamId(stream.getId());
+ stream.releaseResources();
it.remove();
- streams.release(stream);
requestSessionOutput();
}
if (!outputQueue.isEmpty()) {
@@ -1303,12 +1293,12 @@ ByteBuffer getContent() {
}
- H2StreamChannel createChannel(final int streamId, final boolean idle) {
- return new H2StreamChannelImpl(streamId, idle, initInputWinSize, initOutputWinSize);
+ H2StreamChannel createChannel(final int streamId) {
+ return new H2StreamChannelImpl(streamId, initInputWinSize, initOutputWinSize);
}
- void addStream(final H2Stream stream) throws H2ConnectionException {
- streams.addLocallyInitiated(stream);
+ H2Stream createStream(final H2StreamChannel channel, final H2StreamHandler streamHandler) throws H2ConnectionException {
+ return streams.createActive(channel, streamHandler);
}
class H2StreamChannelImpl implements H2StreamChannel {
@@ -1317,14 +1307,11 @@ class H2StreamChannelImpl implements H2StreamChannel {
private final AtomicInteger inputWindow;
private final AtomicInteger outputWindow;
- private volatile boolean idle;
private volatile boolean localClosed;
-
private volatile long localResetTime;
- H2StreamChannelImpl(final int id, final boolean idle, final int initialInputWindowSize, final int initialOutputWindowSize) {
+ H2StreamChannelImpl(final int id, final int initialInputWindowSize, final int initialOutputWindowSize) {
this.id = id;
- this.idle = idle;
this.inputWindow = new AtomicInteger(initialInputWindowSize);
this.outputWindow = new AtomicInteger(initialOutputWindowSize);
}
@@ -1358,7 +1345,6 @@ public void submit(final List<Header> headers, final boolean endStream) throws I
throw new H2ConnectionException(H2Error.INTERNAL_ERROR, "Message headers are missing");
}
ensureNotClosed();
- idle = false;
commitHeaders(id, headers, endStream);
if (endStream) {
localClosed = true;
@@ -1371,15 +1357,14 @@ public void submit(final List<Header> headers, final boolean endStream) throws I
@Override
public void push(final List<Header> headers, final AsyncPushProducer pushProducer) throws HttpException, IOException {
acceptPushRequest();
-
ioSession.getLock().lock();
try {
ensureNotClosed();
-
final int promisedStreamId = streams.generateStreamId();
- executePush(promisedStreamId, pushProducer);
+ final H2StreamChannel channel = createChannel(promisedStreamId);
+ streams.createReserved(channel, outgoingPushPromise(channel, pushProducer));
+
commitPushPromise(id, promisedStreamId, headers);
- idle = false;
} finally {
ioSession.getLock().unlock();
}
@@ -1450,12 +1435,9 @@ public boolean localReset(final int code) throws IOException {
localClosed = true;
localResetTime = System.currentTimeMillis();
- if (!idle) {
- final RawFrame resetStream = frameFactory.createResetStream(id, code);
- commitFrameInternal(resetStream);
- return true;
- }
- return false;
+ final RawFrame resetStream = frameFactory.createResetStream(id, code);
+ commitFrameInternal(resetStream);
+ return true;
} finally {
ioSession.getLock().unlock();
}
@@ -1484,8 +1466,7 @@ public String toString() {
.append(", inputWindow=").append(inputWindow)
.append(", outputWindow=").append(outputWindow)
.append(", localClosed=").append(localClosed)
- .append(", idle=").append(idle);
- buf.append("]");
+ .append("]");
return buf.toString();
}
diff --git a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientH2StreamMultiplexer.java b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientH2StreamMultiplexer.java
index 9102dd1d8..24ecd70df 100644
--- a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientH2StreamMultiplexer.java
+++ b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientH2StreamMultiplexer.java
@@ -30,12 +30,11 @@
import org.apache.hc.core5.annotation.Internal;
import org.apache.hc.core5.http.config.CharCodingConfig;
-import org.apache.hc.core5.http.impl.BasicHttpConnectionMetrics;
import org.apache.hc.core5.http.nio.AsyncClientExchangeHandler;
import org.apache.hc.core5.http.nio.AsyncPushConsumer;
+import org.apache.hc.core5.http.nio.AsyncPushProducer;
import org.apache.hc.core5.http.nio.HandlerFactory;
-import org.apache.hc.core5.http.nio.command.ExecutableCommand;
-import org.apache.hc.core5.http.nio.command.RequestExecutionCommand;
+import org.apache.hc.core5.http.protocol.HttpContext;
import org.apache.hc.core5.http.protocol.HttpCoreContext;
import org.apache.hc.core5.http.protocol.HttpProcessor;
import org.apache.hc.core5.http2.H2ConnectionException;
@@ -114,7 +113,7 @@ void acceptHeaderFrame() throws H2ConnectionException {
}
@Override
- void acceptPushFrame() throws H2ConnectionException {
+ void acceptPushFrame() {
}
@Override
@@ -123,35 +122,36 @@ void acceptPushRequest() throws H2ConnectionException {
}
@Override
- H2StreamHandler createLocallyInitiatedStream(
- final ExecutableCommand command,
+ H2StreamHandler outgoingRequest(
final H2StreamChannel channel,
- final HttpProcessor httpProcessor,
- final BasicHttpConnectionMetrics connMetrics) throws IOException {
- if (command instanceof RequestExecutionCommand) {
- final RequestExecutionCommand executionCommand = (RequestExecutionCommand) command;
- final AsyncClientExchangeHandler exchangeHandler = executionCommand.getExchangeHandler();
- final HandlerFactory<AsyncPushConsumer> pushHandlerFactory = executionCommand.getPushHandlerFactory();
- final HttpCoreContext context = HttpCoreContext.castOrCreate(executionCommand.getContext());
- context.setSSLSession(getSSLSession());
- context.setEndpointDetails(getEndpointDetails());
- return new ClientH2StreamHandler(channel, httpProcessor, connMetrics, exchangeHandler,
- pushHandlerFactory != null ? pushHandlerFactory : this.pushHandlerFactory,
- context);
- }
- throw new H2ConnectionException(H2Error.INTERNAL_ERROR, "Unexpected executable command");
+ final AsyncClientExchangeHandler exchangeHandler,
+ final HandlerFactory<AsyncPushConsumer> pushHandlerFactory,
+ final HttpContext context) {
+ final HttpCoreContext coreContext = HttpCoreContext.castOrCreate(context);
+ coreContext.setSSLSession(getSSLSession());
+ coreContext.setEndpointDetails(getEndpointDetails());
+ return new ClientH2StreamHandler(channel, getHttpProcessor(), getConnMetrics(), exchangeHandler,
+ pushHandlerFactory != null ? pushHandlerFactory : this.pushHandlerFactory,
+ coreContext);
}
@Override
- H2StreamHandler createRemotelyInitiatedStream(
- final H2StreamChannel channel,
- final HttpProcessor httpProcessor,
- final BasicHttpConnectionMetrics connMetrics,
- final HandlerFactory<AsyncPushConsumer> pushHandlerFactory) throws IOException {
+ H2StreamHandler incomingRequest(final H2StreamChannel channel) throws IOException {
+ throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Illegal incoming request");
+ }
+
+ @Override
+ H2StreamHandler outgoingPushPromise(final H2StreamChannel channel, final AsyncPushProducer pushProducer) throws IOException {
+ throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Illegal attempt to send push promise");
+ }
+
+ @Override
+ H2StreamHandler incomingPushPromise(final H2StreamChannel channel,
+ final HandlerFactory<AsyncPushConsumer> pushHandlerFactory) {
final HttpCoreContext context = HttpCoreContext.create();
context.setSSLSession(getSSLSession());
context.setEndpointDetails(getEndpointDetails());
- return new ClientPushH2StreamHandler(channel, httpProcessor, connMetrics,
+ return new ClientPushH2StreamHandler(channel, getHttpProcessor(), getConnMetrics(),
pushHandlerFactory != null ? pushHandlerFactory : this.pushHandlerFactory,
context);
}
diff --git a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/H2Stream.java b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/H2Stream.java
index 90e69f68d..7ec30a035 100644
--- a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/H2Stream.java
+++ b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/H2Stream.java
@@ -33,6 +33,7 @@
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
import org.apache.hc.core5.http.Header;
import org.apache.hc.core5.http.HttpException;
@@ -45,17 +46,23 @@
class H2Stream {
+ enum State { RESERVED, OPEN, CLOSED }
+
private static final long LINGER_TIME = 1000; // 1 second
private final H2StreamChannel channel;
private final H2StreamHandler handler;
+ private final Consumer<State> stateChangeCallback;
private final AtomicBoolean released;
+ private volatile boolean reserved;
private volatile boolean remoteClosed;
- H2Stream(final H2StreamChannel channel, final H2StreamHandler handler) {
+ H2Stream(final H2StreamChannel channel, final H2StreamHandler handler, final Consumer<State> stateChangeCallback) {
this.channel = channel;
this.handler = handler;
+ this.stateChangeCallback = stateChangeCallback;
+ this.reserved = true;
this.released = new AtomicBoolean();
}
@@ -63,6 +70,27 @@ int getId() {
return channel.getId();
}
+ boolean isReserved() {
+ return reserved;
+ }
+
+ private void triggerOpen() {
+ if (stateChangeCallback != null) {
+ stateChangeCallback.accept(State.OPEN);
+ }
+ }
+
+ private void triggerClosed() {
+ if (stateChangeCallback != null) {
+ stateChangeCallback.accept(State.CLOSED);
+ }
+ }
+
+ void activate() {
+ reserved = false;
+ triggerOpen();
+ }
+
AtomicInteger getOutputWindow() {
return channel.getOutputWindow();
}
@@ -71,13 +99,21 @@ AtomicInteger getInputWindow() {
return channel.getInputWindow();
}
- private boolean isPastResetDeadline() {
+ private boolean isPastLingerDeadline() {
final long localResetTime = channel.getLocalResetTime();
return localResetTime > 0 && localResetTime + LINGER_TIME < System.currentTimeMillis();
}
- boolean isTerminated() {
- return channel.isLocalClosed() && (remoteClosed || isPastResetDeadline());
+ boolean isClosedPastLingerDeadline() {
+ return channel.isLocalClosed() && (remoteClosed || isPastLingerDeadline());
+ }
+
+ boolean isClosed() {
+ return channel.isLocalClosed() && (remoteClosed || channel.isLocalReset());
+ }
+
+ boolean isActive() {
+ return !reserved && !isClosed();
}
boolean isRemoteClosed() {
@@ -134,7 +170,7 @@ void consumeData(final ByteBuffer src, final boolean endOfStream) throws HttpExc
}
boolean isOutputReady() {
- return handler.isOutputReady();
+ return !channel.isLocalClosed() && handler.isOutputReady();
}
void produceOutput() throws HttpException, IOException {
@@ -153,16 +189,24 @@ void fail(final Exception cause) {
remoteClosed = true;
channel.markLocalClosed();
if (released.compareAndSet(false, true)) {
- handler.failed(cause);
- handler.releaseResources();
+ try {
+ handler.failed(cause);
+ handler.releaseResources();
+ } finally {
+ triggerClosed();
+ }
}
}
void localReset(final Exception cause, final int code) throws IOException {
channel.localReset(code);
if (released.compareAndSet(false, true)) {
- handler.failed(cause);
- handler.releaseResources();
+ try {
+ handler.failed(cause);
+ handler.releaseResources();
+ } finally {
+ triggerClosed();
+ }
}
}
@@ -185,8 +229,12 @@ HandlerFactory<AsyncPushConsumer> getPushHandlerFactory() {
boolean abort() {
final boolean cancelled = channel.cancel();
if (released.compareAndSet(false, true)) {
- handler.failed(new StreamClosedException());
- handler.releaseResources();
+ try {
+ handler.failed(new StreamClosedException());
+ handler.releaseResources();
+ } finally {
+ triggerClosed();
+ }
}
return cancelled;
}
@@ -194,7 +242,7 @@ boolean abort() {
boolean abortGracefully() throws IOException {
if (!isLocalClosed() && isRemoteClosed()) {
channel.endStream();
- handler.releaseResources();
+ releaseResources();
return true;
} else {
return abort();
@@ -203,7 +251,11 @@ boolean abortGracefully() throws IOException {
void releaseResources() {
if (released.compareAndSet(false, true)) {
- handler.releaseResources();
+ try {
+ handler.releaseResources();
+ } finally {
+ triggerClosed();
+ }
}
}
diff --git a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/H2Streams.java b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/H2Streams.java
index 53b3de5d7..62b348e4c 100644
--- a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/H2Streams.java
+++ b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/H2Streams.java
@@ -47,6 +47,8 @@ class H2Streams {
private final Queue<H2Stream> streams;
private final AtomicInteger lastLocalId;
private final AtomicInteger lastRemoteId;
+ private final AtomicInteger localCount;
+ private final AtomicInteger remoteCount;
public H2Streams(final StreamIdGenerator idGenerator) {
this.idGenerator = Args.notNull(idGenerator, "Stream id generator");
@@ -54,6 +56,8 @@ public H2Streams(final StreamIdGenerator idGenerator) {
this.streams = new ConcurrentLinkedQueue<>();
this.lastLocalId = new AtomicInteger(0);
this.lastRemoteId = new AtomicInteger(0);
+ this.localCount = new AtomicInteger(0);
+ this.remoteCount = new AtomicInteger(0);
}
public int size() {
@@ -76,37 +80,50 @@ public int getLastRemoteId() {
return lastRemoteId.get();
}
- public void addLocallyInitiated(final H2Stream stream) throws H2ConnectionException {
- final int streamId = stream.getId();
- if (isOtherSide(streamId)) {
- throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Illegal stream id");
+ private H2Stream createStream(final H2StreamChannel channel, final H2StreamHandler streamHandler) {
+ final int streamId = channel.getId();
+ final boolean remoteStream = isOtherSide(streamId);
+ final H2Stream stream = new H2Stream(channel, streamHandler, state -> {
+ final AtomicInteger count = remoteStream ? remoteCount : localCount;
+ switch (state) {
+ case OPEN:
+ count.incrementAndGet();
+ break;
+ case CLOSED:
+ count.decrementAndGet();
+ }
+ });
+ if (remoteStream) {
+ final int currentId = lastRemoteId.get();
+ if (streamId > currentId) {
+ lastRemoteId.compareAndSet(currentId, streamId);
+ }
}
streamMap.put(streamId, stream);
streams.add(stream);
+ return stream;
}
- public void addRemotelyInitiated(final H2Stream stream) throws H2ConnectionException {
- final int streamId = stream.getId();
- if (isSameSide(streamId)) {
- throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Illegal stream id");
+ public H2Stream createActive(final H2StreamChannel channel, final H2StreamHandler streamHandler) {
+ final H2Stream stream = createStream(channel, streamHandler);
+ if (!stream.isClosed()) {
+ stream.activate();
}
- final int currentId = lastRemoteId.get();
- if (streamId > currentId) {
- lastRemoteId.compareAndSet(currentId, streamId);
- }
- streamMap.put(streamId, stream);
- streams.add(stream);
+ return stream;
+ }
+
+ public H2Stream createReserved(final H2StreamChannel channel, final H2StreamHandler streamHandler) {
+ return createStream(channel, streamHandler);
}
- public void release(final H2Stream stream) {
- streamMap.remove(stream.getId());
- stream.releaseResources();
+ public void dropStreamId(final int streamId) {
+ streamMap.remove(streamId);
}
public void shutdownAndReleaseAll() {
for (final Iterator<H2Stream> it = streams.iterator(); it.hasNext(); ) {
final H2Stream stream = it.next();
- if (stream.isLocalClosed() && stream.isRemoteClosed()) {
+ if (stream.isClosed()) {
stream.releaseResources();
} else {
stream.fail(new ConnectionClosedException());
@@ -120,22 +137,26 @@ public H2Stream lookup(final int streamId) {
return streamMap.get(streamId);
}
+ boolean hasBeenSeen(final int streamId) {
+ return streamId <= (isSameSide(streamId) ? lastLocalId : lastRemoteId).get();
+ }
+
+ boolean isClosed(final H2Stream stream, final int streamId) {
+ return stream != null ? stream.isLocalClosed() && stream.isRemoteClosed() : hasBeenSeen(streamId);
+ }
+
public H2Stream lookupValidOrNull(final int streamId) throws H2ConnectionException {
final H2Stream stream = streamMap.get(streamId);
- if (stream == null) {
- if (idGenerator.isSameSide(streamId)) {
- if (streamId <= lastLocalId.get()) {
- throw new H2ConnectionException(H2Error.STREAM_CLOSED, "Stream closed");
- }
- } else {
- if (streamId <= lastRemoteId.get()) {
- throw new H2ConnectionException(H2Error.STREAM_CLOSED, "Stream closed");
- }
- }
- } else {
- if (stream.isLocalClosed() && stream.isRemoteClosed()) {
- throw new H2ConnectionException(H2Error.STREAM_CLOSED, "Stream closed");
- }
+ if (isClosed(stream, streamId)) {
+ throw new H2ConnectionException(H2Error.STREAM_CLOSED, "Stream closed");
+ }
+ return stream;
+ }
+
+ public H2Stream lookupSeen(final int streamId) throws H2ConnectionException {
+ final H2Stream stream = streamMap.get(streamId);
+ if (stream == null && !hasBeenSeen(streamId)) {
+ throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Unexpected stream id: " + streamId);
}
return stream;
}
@@ -166,5 +187,4 @@ public int generateStreamId() {
}
}
-
}
diff --git a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ServerH2StreamMultiplexer.java b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ServerH2StreamMultiplexer.java
index 94669275f..046047398 100644
--- a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ServerH2StreamMultiplexer.java
+++ b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ServerH2StreamMultiplexer.java
@@ -35,11 +35,12 @@
import org.apache.hc.core5.http.HttpException;
import org.apache.hc.core5.http.RequestHeaderFieldsTooLargeException;
import org.apache.hc.core5.http.config.CharCodingConfig;
-import org.apache.hc.core5.http.impl.BasicHttpConnectionMetrics;
+import org.apache.hc.core5.http.nio.AsyncClientExchangeHandler;
import org.apache.hc.core5.http.nio.AsyncPushConsumer;
+import org.apache.hc.core5.http.nio.AsyncPushProducer;
import org.apache.hc.core5.http.nio.AsyncServerExchangeHandler;
import org.apache.hc.core5.http.nio.HandlerFactory;
-import org.apache.hc.core5.http.nio.command.ExecutableCommand;
+import org.apache.hc.core5.http.protocol.HttpContext;
import org.apache.hc.core5.http.protocol.HttpCoreContext;
import org.apache.hc.core5.http.protocol.HttpProcessor;
import org.apache.hc.core5.http2.H2ConnectionException;
@@ -103,11 +104,11 @@ H2Setting[] generateSettings(final H2Config localConfig) {
}
@Override
- void acceptHeaderFrame() throws H2ConnectionException {
+ void acceptHeaderFrame() {
}
@Override
- void acceptPushRequest() throws H2ConnectionException {
+ void acceptPushRequest() {
}
@Override
@@ -116,24 +117,35 @@ void acceptPushFrame() throws H2ConnectionException {
}
@Override
- H2StreamHandler createRemotelyInitiatedStream(
- final H2StreamChannel channel,
- final HttpProcessor httpProcessor,
- final BasicHttpConnectionMetrics connMetrics,
- final HandlerFactory<AsyncPushConsumer> pushHandlerFactory) throws IOException {
+ H2StreamHandler incomingRequest(final H2StreamChannel channel) {
final HttpCoreContext context = HttpCoreContext.create();
context.setSSLSession(getSSLSession());
context.setEndpointDetails(getEndpointDetails());
- return new ServerH2StreamHandler(channel, httpProcessor, connMetrics, exchangeHandlerFactory, context);
+ return new ServerH2StreamHandler(channel, getHttpProcessor(), getConnMetrics(), exchangeHandlerFactory, context);
}
@Override
- H2StreamHandler createLocallyInitiatedStream(
- final ExecutableCommand command,
+ H2StreamHandler outgoingRequest(
final H2StreamChannel channel,
- final HttpProcessor httpProcessor,
- final BasicHttpConnectionMetrics connMetrics) throws IOException {
- throw new H2ConnectionException(H2Error.INTERNAL_ERROR, "Illegal attempt to execute a request");
+ final AsyncClientExchangeHandler exchangeHandler,
+ final HandlerFactory<AsyncPushConsumer> pushHandlerFactory,
+ final HttpContext context) throws IOException {
+ throw new H2ConnectionException(H2Error.INTERNAL_ERROR, "Illegal attempt to send a request");
+ }
+
+ @Override
+ H2StreamHandler incomingPushPromise(final H2StreamChannel channel,
+ final HandlerFactory<AsyncPushConsumer> pushHandlerFactory) throws IOException {
+ throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Illegal incoming push promise");
+ }
+
+ @Override
+ H2StreamHandler outgoingPushPromise(final H2StreamChannel channel,
+ final AsyncPushProducer pushProducer) throws IOException {
+ final HttpCoreContext context = HttpCoreContext.create();
+ context.setSSLSession(getSSLSession());
+ context.setEndpointDetails(getEndpointDetails());
+ return new ServerPushH2StreamHandler(channel, getHttpProcessor(), getConnMetrics(), pushProducer, context);
}
@Override
diff --git a/httpcore5-h2/src/test/java/org/apache/hc/core5/http2/impl/nio/TestAbstractH2StreamMultiplexer.java b/httpcore5-h2/src/test/java/org/apache/hc/core5/http2/impl/nio/TestAbstractH2StreamMultiplexer.java
index 7c7208b48..4025ae28e 100644
--- a/httpcore5-h2/src/test/java/org/apache/hc/core5/http2/impl/nio/TestAbstractH2StreamMultiplexer.java
+++ b/httpcore5-h2/src/test/java/org/apache/hc/core5/http2/impl/nio/TestAbstractH2StreamMultiplexer.java
@@ -27,7 +27,6 @@
package org.apache.hc.core5.http2.impl.nio;
-import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
@@ -37,12 +36,13 @@
import org.apache.hc.core5.function.Supplier;
import org.apache.hc.core5.http.Header;
import org.apache.hc.core5.http.config.CharCodingConfig;
-import org.apache.hc.core5.http.impl.BasicHttpConnectionMetrics;
import org.apache.hc.core5.http.impl.CharCodingSupport;
import org.apache.hc.core5.http.message.BasicHeader;
+import org.apache.hc.core5.http.nio.AsyncClientExchangeHandler;
import org.apache.hc.core5.http.nio.AsyncPushConsumer;
+import org.apache.hc.core5.http.nio.AsyncPushProducer;
import org.apache.hc.core5.http.nio.HandlerFactory;
-import org.apache.hc.core5.http.nio.command.ExecutableCommand;
+import org.apache.hc.core5.http.protocol.HttpContext;
import org.apache.hc.core5.http.protocol.HttpProcessor;
import org.apache.hc.core5.http2.H2ConnectionException;
import org.apache.hc.core5.http2.H2Error;
@@ -141,20 +141,27 @@ void acceptPushFrame() throws H2ConnectionException {
}
@Override
- H2StreamHandler createRemotelyInitiatedStream(
- final H2StreamChannel channel,
- final HttpProcessor httpProcessor,
- final BasicHttpConnectionMetrics connMetrics,
- final HandlerFactory<AsyncPushConsumer> pushHandlerFactory) throws IOException {
+ H2StreamHandler incomingRequest(final H2StreamChannel channel) {
return streamHandlerSupplier.get();
}
@Override
- H2StreamHandler createLocallyInitiatedStream(
- final ExecutableCommand command,
- final H2StreamChannel channel,
- final HttpProcessor httpProcessor,
- final BasicHttpConnectionMetrics connMetrics) throws IOException {
+ H2StreamHandler outgoingRequest(final H2StreamChannel channel,
+ final AsyncClientExchangeHandler exchangeHandler,
+ final HandlerFactory<AsyncPushConsumer> pushHandlerFactory,
+ final HttpContext context) {
+ return null;
+ }
+
+ @Override
+ H2StreamHandler incomingPushPromise(final H2StreamChannel channel,
+ final HandlerFactory<AsyncPushConsumer> pushHandlerFactory) {
+ return streamHandlerSupplier.get();
+ }
+
+ @Override
+ H2StreamHandler outgoingPushPromise(final H2StreamChannel channel,
+ final AsyncPushProducer pushProducer) {
return null;
}
@@ -684,9 +691,8 @@ void testStreamRemoteReset() throws Exception {
h2StreamListener,
() -> streamHandler);
- final H2StreamChannel channel = streamMultiplexer.createChannel(1, false);
- final H2Stream stream = new H2Stream(channel, streamHandler);
- streamMultiplexer.addStream(stream);
+ final H2StreamChannel channel = streamMultiplexer.createChannel(1);
+ final H2Stream stream = streamMultiplexer.createStream(channel, streamHandler);
final ByteArrayBuffer buf = new ByteArrayBuffer(19);
final HPackEncoder encoder = new HPackEncoder(H2Config.INIT.getHeaderTableSize(), CharCodingSupport.createEncoder(CharCodingConfig.DEFAULT));
@@ -731,9 +737,8 @@ void testStreamRemoteResetNoErrorRemoteAlreadyClosed() throws Exception {
h2StreamListener,
() -> streamHandler);
- final H2StreamChannel channel = streamMultiplexer.createChannel(1, false);
- final H2Stream stream = new H2Stream(channel, streamHandler);
- streamMultiplexer.addStream(stream);
+ final H2StreamChannel channel = streamMultiplexer.createChannel(1);
+ final H2Stream stream = streamMultiplexer.createStream(channel, streamHandler);
final ByteArrayBuffer buf = new ByteArrayBuffer(19);
final HPackEncoder encoder = new HPackEncoder(H2Config.INIT.getHeaderTableSize(), CharCodingSupport.createEncoder(CharCodingConfig.DEFAULT));