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 2020/08/07 13:27:09 UTC
[httpcomponents-core] branch 5.1.x updated (3092c6d -> 8838ad0)
This is an automated email from the ASF dual-hosted git repository.
olegk pushed a change to branch 5.1.x
in repository https://gitbox.apache.org/repos/asf/httpcomponents-core.git.
discard 3092c6d HTTPCORE-639: Add a configurable ResponseOutOfOrder strategy for DefaultBHttpClientConnection
new 8838ad0 HTTPCORE-639: Add a configurable ResponseOutOfOrder strategy for DefaultBHttpClientConnection
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (3092c6d)
\
N -- N -- N refs/heads/5.1.x (8838ad0)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
...gResponseOutOfOrderStrategyIntegrationTest.java | 45 ++++++++++++++++------
1 file changed, 33 insertions(+), 12 deletions(-)
[httpcomponents-core] 01/01: HTTPCORE-639: Add a configurable
ResponseOutOfOrder strategy for DefaultBHttpClientConnection
Posted by ol...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
olegk pushed a commit to branch 5.1.x
in repository https://gitbox.apache.org/repos/asf/httpcomponents-core.git
commit 8838ad0285ad49eb6ee4c254f7649660700ade07
Author: Carter Kozak <c4...@gmail.com>
AuthorDate: Mon Jul 27 10:17:45 2020 -0400
HTTPCORE-639: Add a configurable ResponseOutOfOrder strategy for DefaultBHttpClientConnection
This adds a configurable ResponseOutOfOrderStrategy in place of the
previous always-enabled behavior, and uses the no-op
NoResponseOutOfOrderStrategy implementation by default.
The previous behavior can be used by selecting the
MonitoringResponseOutOfOrderStrategy, which has been updated
to support more flexible behavior. Note that this strategy
results in a 1 ms pause for every chunk transferred, limiting
upload speed using the default 8 KiB chunk size to at most
8 MiB/second.
The original discussion can be found on the mailing list:
https://www.mail-archive.com/httpclient-users@hc.apache.org/msg09911.html
This closes #206
---
...gResponseOutOfOrderStrategyIntegrationTest.java | 215 +++++++++++++++++++++
.../http/impl/io/DefaultBHttpClientConnection.java | 105 +++++++++-
.../io/DefaultBHttpClientConnectionFactory.java | 32 ++-
.../io/MonitoringResponseOutOfOrderStrategy.java | 111 +++++++++++
.../http/impl/io/NoResponseOutOfOrderStrategy.java | 61 ++++++
.../http/impl/io/ResponseOutOfOrderException.java | 41 ++++
.../core5/http/io/ResponseOutOfOrderStrategy.java | 66 +++++++
.../java/org/apache/hc/core5/util/Timeout.java | 5 +
.../TestMonitoringResponseOutOfOrderStrategy.java | 133 +++++++++++++
9 files changed, 764 insertions(+), 5 deletions(-)
diff --git a/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/classic/MonitoringResponseOutOfOrderStrategyIntegrationTest.java b/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/classic/MonitoringResponseOutOfOrderStrategyIntegrationTest.java
new file mode 100644
index 0000000..fad4491
--- /dev/null
+++ b/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/classic/MonitoringResponseOutOfOrderStrategyIntegrationTest.java
@@ -0,0 +1,215 @@
+/*
+ * ====================================================================
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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.
+ * ====================================================================
+ *
+ * This software consists of voluntary contributions made by many
+ * individuals on behalf of the Apache Software Foundation. For more
+ * information on the Apache Software Foundation, please see
+ * <http://www.apache.org/>.
+ *
+ */
+
+package org.apache.hc.core5.testing.classic;
+
+import org.apache.hc.core5.http.ClassicHttpRequest;
+import org.apache.hc.core5.http.ClassicHttpResponse;
+import org.apache.hc.core5.http.ContentType;
+import org.apache.hc.core5.http.HttpHost;
+import org.apache.hc.core5.http.Method;
+import org.apache.hc.core5.http.URIScheme;
+import org.apache.hc.core5.http.impl.bootstrap.HttpRequester;
+import org.apache.hc.core5.http.impl.bootstrap.RequesterBootstrap;
+import org.apache.hc.core5.http.impl.io.DefaultBHttpClientConnectionFactory;
+import org.apache.hc.core5.http.impl.io.MonitoringResponseOutOfOrderStrategy;
+import org.apache.hc.core5.http.io.HttpRequestHandler;
+import org.apache.hc.core5.http.io.SocketConfig;
+import org.apache.hc.core5.http.io.entity.AbstractHttpEntity;
+import org.apache.hc.core5.http.io.entity.EntityUtils;
+import org.apache.hc.core5.http.message.BasicClassicHttpRequest;
+import org.apache.hc.core5.http.protocol.HttpContext;
+import org.apache.hc.core5.http.protocol.HttpCoreContext;
+import org.apache.hc.core5.io.CloseMode;
+import org.apache.hc.core5.testing.SSLTestContexts;
+import org.apache.hc.core5.util.Timeout;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExternalResource;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Arrays;
+import java.util.Collection;
+
+@RunWith(Parameterized.class)
+public class MonitoringResponseOutOfOrderStrategyIntegrationTest {
+
+ // Use a 16k buffer for consistent results across systems
+ private static final int BUFFER_SIZE = 16 * 1024;
+ private static final Timeout TIMEOUT = Timeout.ofSeconds(3);
+
+ @Parameterized.Parameters(name = "{0}")
+ public static Collection<Object[]> protocols() {
+ return Arrays.asList(new Object[][]{
+ { URIScheme.HTTP },
+ { URIScheme.HTTPS }
+ });
+ }
+
+ private final URIScheme scheme;
+ private ClassicTestServer server;
+ private HttpRequester requester;
+
+ public MonitoringResponseOutOfOrderStrategyIntegrationTest(final URIScheme scheme) {
+ this.scheme = scheme;
+ }
+
+ @Rule
+ public ExternalResource serverResource = new ExternalResource() {
+
+ @Override
+ protected void before() throws Throwable {
+ server = new ClassicTestServer(
+ scheme == URIScheme.HTTPS ? SSLTestContexts.createServerSSLContext() : null,
+ SocketConfig.custom()
+ .setSoTimeout(TIMEOUT)
+ .setSndBufSize(BUFFER_SIZE)
+ .setRcvBufSize(BUFFER_SIZE)
+ .setSoKeepAlive(false)
+ .build());
+ }
+
+ @Override
+ protected void after() {
+ if (server != null) {
+ try {
+ server.shutdown(CloseMode.IMMEDIATE);
+ server = null;
+ } catch (final Exception ignore) {
+ }
+ }
+ }
+
+ };
+
+ @Rule
+ public ExternalResource requesterResource = new ExternalResource() {
+
+ @Override
+ protected void before() throws Throwable {
+ requester = RequesterBootstrap.bootstrap()
+ .setSslContext(scheme == URIScheme.HTTPS ? SSLTestContexts.createClientSSLContext() : null)
+ .setSocketConfig(SocketConfig.custom()
+ .setSoTimeout(TIMEOUT)
+ .setRcvBufSize(BUFFER_SIZE)
+ .setSndBufSize(BUFFER_SIZE)
+ .setSoKeepAlive(false)
+ .build())
+ .setStreamListener(LoggingHttp1StreamListener.INSTANCE)
+ .setConnPoolListener(LoggingConnPoolListener.INSTANCE)
+ .setConnectionFactory(DefaultBHttpClientConnectionFactory.builder()
+ .responseOutOfOrderStrategy(MonitoringResponseOutOfOrderStrategy.INSTANCE)
+ .build())
+ .create();
+ }
+
+ @Override
+ protected void after() {
+ if (requester != null) {
+ try {
+ requester.close(CloseMode.IMMEDIATE);
+ requester = null;
+ } catch (final Exception ignore) {
+ }
+ }
+ }
+
+ };
+
+ @Test(timeout = 5000) // Failures may hang
+ public void testResponseOutOfOrderWithDefaultStrategy() throws Exception {
+ this.server.registerHandler("*", new HttpRequestHandler() {
+
+ @Override
+ public void handle(
+ final ClassicHttpRequest request,
+ final ClassicHttpResponse response,
+ final HttpContext context) throws IOException {
+ response.setCode(400);
+ response.setEntity(new AllOnesHttpEntity(200000));
+ }
+
+ });
+
+ this.server.start(null, null, null);
+
+ final HttpCoreContext context = HttpCoreContext.create();
+ final HttpHost host = new HttpHost(scheme.id, "localhost", this.server.getPort());
+
+ final ClassicHttpRequest post = new BasicClassicHttpRequest(Method.POST, "/");
+ post.setEntity(new AllOnesHttpEntity(200000));
+
+ try (final ClassicHttpResponse response = requester.execute(host, post, TIMEOUT, context)) {
+ Assert.assertEquals(400, response.getCode());
+ EntityUtils.consumeQuietly(response.getEntity());
+ }
+ }
+
+ private static final class AllOnesHttpEntity extends AbstractHttpEntity {
+ private long remaining;
+
+ protected AllOnesHttpEntity(final long length) {
+ super(ContentType.APPLICATION_OCTET_STREAM, null, true);
+ this.remaining = length;
+ }
+
+ @Override
+ public InputStream getContent() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void writeTo(final OutputStream outStream) throws IOException {
+ final byte[] buf = new byte[1024];
+ while (remaining > 0) {
+ final int writeLength = (int) Math.min(remaining, buf.length);
+ outStream.write(buf, 0, writeLength);
+ outStream.flush();
+ remaining -= writeLength;
+ }
+ }
+
+ @Override
+ public boolean isStreaming() {
+ return true;
+ }
+
+ @Override
+ public void close() {
+ }
+
+ @Override
+ public long getContentLength() {
+ return -1L;
+ }
+ }
+}
diff --git a/httpcore5/src/main/java/org/apache/hc/core5/http/impl/io/DefaultBHttpClientConnection.java b/httpcore5/src/main/java/org/apache/hc/core5/http/impl/io/DefaultBHttpClientConnection.java
index dd5c1e3..d905bab 100644
--- a/httpcore5/src/main/java/org/apache/hc/core5/http/impl/io/DefaultBHttpClientConnection.java
+++ b/httpcore5/src/main/java/org/apache/hc/core5/http/impl/io/DefaultBHttpClientConnection.java
@@ -28,6 +28,7 @@
package org.apache.hc.core5.http.impl.io;
import java.io.IOException;
+import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.nio.charset.CharsetDecoder;
@@ -51,6 +52,7 @@ import org.apache.hc.core5.http.io.HttpMessageParser;
import org.apache.hc.core5.http.io.HttpMessageParserFactory;
import org.apache.hc.core5.http.io.HttpMessageWriter;
import org.apache.hc.core5.http.io.HttpMessageWriterFactory;
+import org.apache.hc.core5.http.io.ResponseOutOfOrderStrategy;
import org.apache.hc.core5.util.Args;
/**
@@ -65,6 +67,7 @@ public class DefaultBHttpClientConnection extends BHttpConnectionBase
private final HttpMessageWriter<ClassicHttpRequest> requestWriter;
private final ContentLengthStrategy incomingContentStrategy;
private final ContentLengthStrategy outgoingContentStrategy;
+ private final ResponseOutOfOrderStrategy responseOutOfOrderStrategy;
private volatile boolean consistent;
/**
@@ -80,6 +83,8 @@ public class DefaultBHttpClientConnection extends BHttpConnectionBase
* {@link DefaultContentLengthStrategy#INSTANCE} will be used.
* @param outgoingContentStrategy outgoing content length strategy. If {@code null}
* {@link DefaultContentLengthStrategy#INSTANCE} will be used.
+ * @param responseOutOfOrderStrategy response out of order strategy. If {@code null}
+ * {@link NoResponseOutOfOrderStrategy#INSTANCE} will be used.
* @param requestWriterFactory request writer factory. If {@code null}
* {@link DefaultHttpRequestWriterFactory#INSTANCE} will be used.
* @param responseParserFactory response parser factory. If {@code null}
@@ -91,6 +96,7 @@ public class DefaultBHttpClientConnection extends BHttpConnectionBase
final CharsetEncoder charEncoder,
final ContentLengthStrategy incomingContentStrategy,
final ContentLengthStrategy outgoingContentStrategy,
+ final ResponseOutOfOrderStrategy responseOutOfOrderStrategy,
final HttpMessageWriterFactory<ClassicHttpRequest> requestWriterFactory,
final HttpMessageParserFactory<ClassicHttpResponse> responseParserFactory) {
super(http1Config, charDecoder, charEncoder);
@@ -99,12 +105,51 @@ public class DefaultBHttpClientConnection extends BHttpConnectionBase
this.responseParser = (responseParserFactory != null ? responseParserFactory :
DefaultHttpResponseParserFactory.INSTANCE).create(http1Config);
this.incomingContentStrategy = incomingContentStrategy != null ? incomingContentStrategy :
- DefaultContentLengthStrategy.INSTANCE;
+ DefaultContentLengthStrategy.INSTANCE;
this.outgoingContentStrategy = outgoingContentStrategy != null ? outgoingContentStrategy :
- DefaultContentLengthStrategy.INSTANCE;
+ DefaultContentLengthStrategy.INSTANCE;
+ this.responseOutOfOrderStrategy = responseOutOfOrderStrategy != null ? responseOutOfOrderStrategy :
+ NoResponseOutOfOrderStrategy.INSTANCE;
this.consistent = true;
}
+ /**
+ * Creates new instance of DefaultBHttpClientConnection.
+ *
+ * @param http1Config Message http1Config. If {@code null}
+ * {@link Http1Config#DEFAULT} will be used.
+ * @param charDecoder decoder to be used for decoding HTTP protocol elements.
+ * If {@code null} simple type cast will be used for byte to char conversion.
+ * @param charEncoder encoder to be used for encoding HTTP protocol elements.
+ * If {@code null} simple type cast will be used for char to byte conversion.
+ * @param incomingContentStrategy incoming content length strategy. If {@code null}
+ * {@link DefaultContentLengthStrategy#INSTANCE} will be used.
+ * @param outgoingContentStrategy outgoing content length strategy. If {@code null}
+ * {@link DefaultContentLengthStrategy#INSTANCE} will be used.
+ * @param requestWriterFactory request writer factory. If {@code null}
+ * {@link DefaultHttpRequestWriterFactory#INSTANCE} will be used.
+ * @param responseParserFactory response parser factory. If {@code null}
+ * {@link DefaultHttpResponseParserFactory#INSTANCE} will be used.
+ */
+ public DefaultBHttpClientConnection(
+ final Http1Config http1Config,
+ final CharsetDecoder charDecoder,
+ final CharsetEncoder charEncoder,
+ final ContentLengthStrategy incomingContentStrategy,
+ final ContentLengthStrategy outgoingContentStrategy,
+ final HttpMessageWriterFactory<ClassicHttpRequest> requestWriterFactory,
+ final HttpMessageParserFactory<ClassicHttpResponse> responseParserFactory) {
+ this(
+ http1Config,
+ charDecoder,
+ charEncoder,
+ incomingContentStrategy,
+ outgoingContentStrategy,
+ null,
+ requestWriterFactory,
+ responseParserFactory);
+ }
+
public DefaultBHttpClientConnection(
final Http1Config http1Config,
final CharsetDecoder charDecoder,
@@ -149,8 +194,62 @@ public class DefaultBHttpClientConnection extends BHttpConnectionBase
if (len == ContentLengthStrategy.UNDEFINED) {
throw new LengthRequiredException();
}
- try (final OutputStream outStream = createContentOutputStream(len, this.outbuffer, socketHolder.getOutputStream(), entity.getTrailers())) {
+ try (final OutputStream outStream = createContentOutputStream(
+ len, this.outbuffer, new OutputStream() {
+
+ final OutputStream socketOutputStream = socketHolder.getOutputStream();
+ final InputStream socketInputStream = socketHolder.getInputStream();
+
+ long totalBytes = 0;
+
+ void checkForEarlyResponse(final long totalBytesSent, final int nextWriteSize) throws IOException {
+ if (responseOutOfOrderStrategy.isEarlyResponseDetected(
+ request,
+ DefaultBHttpClientConnection.this,
+ socketInputStream,
+ totalBytesSent,
+ nextWriteSize)) {
+ throw new ResponseOutOfOrderException();
+ }
+ }
+
+ @Override
+ public void write(final byte[] b) throws IOException {
+ checkForEarlyResponse(totalBytes, b.length);
+ totalBytes += b.length;
+ socketOutputStream.write(b);
+ }
+
+ @Override
+ public void write(final byte[] b, final int off, final int len) throws IOException {
+ checkForEarlyResponse(totalBytes, len);
+ totalBytes += len;
+ socketOutputStream.write(b, off, len);
+ }
+
+ @Override
+ public void write(final int b) throws IOException {
+ checkForEarlyResponse(totalBytes, 1);
+ totalBytes++;
+ socketOutputStream.write(b);
+ }
+
+ @Override
+ public void flush() throws IOException {
+ socketOutputStream.flush();
+ }
+
+ @Override
+ public void close() throws IOException {
+ socketOutputStream.close();
+ }
+
+ }, entity.getTrailers())) {
entity.writeTo(outStream);
+ } catch (final ResponseOutOfOrderException ex) {
+ if (len > 0) {
+ this.consistent = false;
+ }
}
}
diff --git a/httpcore5/src/main/java/org/apache/hc/core5/http/impl/io/DefaultBHttpClientConnectionFactory.java b/httpcore5/src/main/java/org/apache/hc/core5/http/impl/io/DefaultBHttpClientConnectionFactory.java
index 37c2e48..c2becd2 100644
--- a/httpcore5/src/main/java/org/apache/hc/core5/http/impl/io/DefaultBHttpClientConnectionFactory.java
+++ b/httpcore5/src/main/java/org/apache/hc/core5/http/impl/io/DefaultBHttpClientConnectionFactory.java
@@ -41,6 +41,7 @@ import org.apache.hc.core5.http.impl.CharCodingSupport;
import org.apache.hc.core5.http.io.HttpConnectionFactory;
import org.apache.hc.core5.http.io.HttpMessageParserFactory;
import org.apache.hc.core5.http.io.HttpMessageWriterFactory;
+import org.apache.hc.core5.http.io.ResponseOutOfOrderStrategy;
/**
* Default factory for {@link org.apache.hc.core5.http.io.HttpClientConnection}s.
@@ -55,21 +56,23 @@ public class DefaultBHttpClientConnectionFactory
private final CharCodingConfig charCodingConfig;
private final ContentLengthStrategy incomingContentStrategy;
private final ContentLengthStrategy outgoingContentStrategy;
+ private final ResponseOutOfOrderStrategy responseOutOfOrderStrategy;
private final HttpMessageWriterFactory<ClassicHttpRequest> requestWriterFactory;
private final HttpMessageParserFactory<ClassicHttpResponse> responseParserFactory;
- public DefaultBHttpClientConnectionFactory(
+ private DefaultBHttpClientConnectionFactory(
final Http1Config http1Config,
final CharCodingConfig charCodingConfig,
final ContentLengthStrategy incomingContentStrategy,
final ContentLengthStrategy outgoingContentStrategy,
+ final ResponseOutOfOrderStrategy responseOutOfOrderStrategy,
final HttpMessageWriterFactory<ClassicHttpRequest> requestWriterFactory,
final HttpMessageParserFactory<ClassicHttpResponse> responseParserFactory) {
- super();
this.http1Config = http1Config != null ? http1Config : Http1Config.DEFAULT;
this.charCodingConfig = charCodingConfig != null ? charCodingConfig : CharCodingConfig.DEFAULT;
this.incomingContentStrategy = incomingContentStrategy;
this.outgoingContentStrategy = outgoingContentStrategy;
+ this.responseOutOfOrderStrategy = responseOutOfOrderStrategy;
this.requestWriterFactory = requestWriterFactory;
this.responseParserFactory = responseParserFactory;
}
@@ -77,6 +80,23 @@ public class DefaultBHttpClientConnectionFactory
public DefaultBHttpClientConnectionFactory(
final Http1Config http1Config,
final CharCodingConfig charCodingConfig,
+ final ContentLengthStrategy incomingContentStrategy,
+ final ContentLengthStrategy outgoingContentStrategy,
+ final HttpMessageWriterFactory<ClassicHttpRequest> requestWriterFactory,
+ final HttpMessageParserFactory<ClassicHttpResponse> responseParserFactory) {
+ this(
+ http1Config,
+ charCodingConfig,
+ incomingContentStrategy,
+ outgoingContentStrategy,
+ null,
+ requestWriterFactory,
+ responseParserFactory);
+ }
+
+ public DefaultBHttpClientConnectionFactory(
+ final Http1Config http1Config,
+ final CharCodingConfig charCodingConfig,
final HttpMessageWriterFactory<ClassicHttpRequest> requestWriterFactory,
final HttpMessageParserFactory<ClassicHttpResponse> responseParserFactory) {
this(http1Config, charCodingConfig, null, null, requestWriterFactory, responseParserFactory);
@@ -100,6 +120,7 @@ public class DefaultBHttpClientConnectionFactory
CharCodingSupport.createEncoder(this.charCodingConfig),
this.incomingContentStrategy,
this.outgoingContentStrategy,
+ this.responseOutOfOrderStrategy,
this.requestWriterFactory,
this.responseParserFactory);
conn.bind(socket);
@@ -125,6 +146,7 @@ public class DefaultBHttpClientConnectionFactory
private CharCodingConfig charCodingConfig;
private ContentLengthStrategy incomingContentLengthStrategy;
private ContentLengthStrategy outgoingContentLengthStrategy;
+ private ResponseOutOfOrderStrategy responseOutOfOrderStrategy;
private HttpMessageWriterFactory<ClassicHttpRequest> requestWriterFactory;
private HttpMessageParserFactory<ClassicHttpResponse> responseParserFactory;
@@ -150,6 +172,11 @@ public class DefaultBHttpClientConnectionFactory
return this;
}
+ public Builder responseOutOfOrderStrategy(final ResponseOutOfOrderStrategy responseOutOfOrderStrategy) {
+ this.responseOutOfOrderStrategy = responseOutOfOrderStrategy;
+ return this;
+ }
+
public Builder requestWriterFactory(
final HttpMessageWriterFactory<ClassicHttpRequest> requestWriterFactory) {
this.requestWriterFactory = requestWriterFactory;
@@ -168,6 +195,7 @@ public class DefaultBHttpClientConnectionFactory
charCodingConfig,
incomingContentLengthStrategy,
outgoingContentLengthStrategy,
+ responseOutOfOrderStrategy,
requestWriterFactory,
responseParserFactory);
}
diff --git a/httpcore5/src/main/java/org/apache/hc/core5/http/impl/io/MonitoringResponseOutOfOrderStrategy.java b/httpcore5/src/main/java/org/apache/hc/core5/http/impl/io/MonitoringResponseOutOfOrderStrategy.java
new file mode 100644
index 0000000..303e61b
--- /dev/null
+++ b/httpcore5/src/main/java/org/apache/hc/core5/http/impl/io/MonitoringResponseOutOfOrderStrategy.java
@@ -0,0 +1,111 @@
+/*
+ * ====================================================================
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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.
+ * ====================================================================
+ *
+ * This software consists of voluntary contributions made by many
+ * individuals on behalf of the Apache Software Foundation. For more
+ * information on the Apache Software Foundation, please see
+ * <http://www.apache.org/>.
+ *
+ */
+
+package org.apache.hc.core5.http.impl.io;
+
+import org.apache.hc.core5.annotation.Contract;
+import org.apache.hc.core5.annotation.ThreadingBehavior;
+import org.apache.hc.core5.http.ClassicHttpRequest;
+import org.apache.hc.core5.http.io.HttpClientConnection;
+import org.apache.hc.core5.http.io.ResponseOutOfOrderStrategy;
+import org.apache.hc.core5.util.Args;
+import org.apache.hc.core5.util.Timeout;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * A {@link ResponseOutOfOrderStrategy} implementation which checks for premature responses every {@link #chunkSize}
+ * bytes. An 8 KiB chunk size is used by default based on testing using values between 4 KiB and 128 KiB. This is
+ * optimized for correctness and results in a maximum upload speed of 8 MiB/s until {@link #maxChunksToCheck} is
+ * reached.
+ *
+ * @since 5.1
+ */
+@Contract(threading = ThreadingBehavior.IMMUTABLE)
+public final class MonitoringResponseOutOfOrderStrategy implements ResponseOutOfOrderStrategy {
+
+ private static final int DEFAULT_CHUNK_SIZE = 8 * 1024;
+
+ public static final MonitoringResponseOutOfOrderStrategy INSTANCE = new MonitoringResponseOutOfOrderStrategy();
+
+ private final long chunkSize;
+ private final long maxChunksToCheck;
+
+ /**
+ * Instantiates a default {@link MonitoringResponseOutOfOrderStrategy}. {@link #INSTANCE} may be used instead.
+ */
+ public MonitoringResponseOutOfOrderStrategy() {
+ this(DEFAULT_CHUNK_SIZE);
+ }
+
+ /**
+ * Instantiates a {@link MonitoringResponseOutOfOrderStrategy} with unlimited {@link #maxChunksToCheck}.
+ *
+ * @param chunkSize The chunk size after which a response check is executed.
+ */
+ public MonitoringResponseOutOfOrderStrategy(final long chunkSize) {
+ this(chunkSize, Long.MAX_VALUE);
+ }
+
+ /**
+ * Instantiates a {@link MonitoringResponseOutOfOrderStrategy}.
+ *
+ * @param chunkSize The chunk size after which a response check is executed.
+ * @param maxChunksToCheck The maximum number of chunks to check, allowing expensive checks to be avoided
+ * after a sufficient portion of the request entity has been transferred.
+ */
+ public MonitoringResponseOutOfOrderStrategy(final long chunkSize, final long maxChunksToCheck) {
+ this.chunkSize = Args.positive(chunkSize, "chunkSize");
+ this.maxChunksToCheck = Args.positive(maxChunksToCheck, "maxChunksToCheck");
+ }
+
+ @Override
+ public boolean isEarlyResponseDetected(
+ final ClassicHttpRequest request,
+ final HttpClientConnection connection,
+ final InputStream inputStream,
+ final long totalBytesSent,
+ final long nextWriteSize) throws IOException {
+ if (nextWriteStartsNewChunk(totalBytesSent, nextWriteSize)) {
+ final boolean ssl = connection.getSSLSession() != null;
+ return ssl ? connection.isDataAvailable(Timeout.ONE_MILLISECOND) : (inputStream.available() > 0);
+ }
+ return false;
+ }
+
+ private boolean nextWriteStartsNewChunk(final long totalBytesSent, final long nextWriteSize) {
+ final long currentChunkIndex = Math.min(totalBytesSent / chunkSize, maxChunksToCheck);
+ final long newChunkIndex = Math.min((totalBytesSent + nextWriteSize) / chunkSize, maxChunksToCheck);
+ return currentChunkIndex < newChunkIndex;
+ }
+
+ @Override
+ public String toString() {
+ return "DefaultResponseOutOfOrderStrategy{chunkSize=" + chunkSize + ", maxChunksToCheck=" + maxChunksToCheck + '}';
+ }
+}
diff --git a/httpcore5/src/main/java/org/apache/hc/core5/http/impl/io/NoResponseOutOfOrderStrategy.java b/httpcore5/src/main/java/org/apache/hc/core5/http/impl/io/NoResponseOutOfOrderStrategy.java
new file mode 100644
index 0000000..94606e2
--- /dev/null
+++ b/httpcore5/src/main/java/org/apache/hc/core5/http/impl/io/NoResponseOutOfOrderStrategy.java
@@ -0,0 +1,61 @@
+/*
+ * ====================================================================
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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.
+ * ====================================================================
+ *
+ * This software consists of voluntary contributions made by many
+ * individuals on behalf of the Apache Software Foundation. For more
+ * information on the Apache Software Foundation, please see
+ * <http://www.apache.org/>.
+ *
+ */
+
+package org.apache.hc.core5.http.impl.io;
+
+import org.apache.hc.core5.annotation.Contract;
+import org.apache.hc.core5.annotation.ThreadingBehavior;
+import org.apache.hc.core5.http.ClassicHttpRequest;
+import org.apache.hc.core5.http.io.HttpClientConnection;
+import org.apache.hc.core5.http.io.ResponseOutOfOrderStrategy;
+
+import java.io.InputStream;
+
+/**
+ * An implementation of {@link ResponseOutOfOrderStrategy} which does not check for early responses.
+ *
+ * Early response detection requires 1ms blocking reads and incurs a hefty performance cost for
+ * large uploads.
+ *
+ * @see MonitoringResponseOutOfOrderStrategy
+ * @since 5.1
+ */
+@Contract(threading = ThreadingBehavior.IMMUTABLE)
+public final class NoResponseOutOfOrderStrategy implements ResponseOutOfOrderStrategy {
+
+ public static final NoResponseOutOfOrderStrategy INSTANCE = new NoResponseOutOfOrderStrategy();
+
+ @Override
+ public boolean isEarlyResponseDetected(
+ final ClassicHttpRequest request,
+ final HttpClientConnection connection,
+ final InputStream inputStream,
+ final long totalBytesSent,
+ final long nextWriteSize) {
+ return false;
+ }
+}
diff --git a/httpcore5/src/main/java/org/apache/hc/core5/http/impl/io/ResponseOutOfOrderException.java b/httpcore5/src/main/java/org/apache/hc/core5/http/impl/io/ResponseOutOfOrderException.java
new file mode 100644
index 0000000..5744342
--- /dev/null
+++ b/httpcore5/src/main/java/org/apache/hc/core5/http/impl/io/ResponseOutOfOrderException.java
@@ -0,0 +1,41 @@
+/*
+ * ====================================================================
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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.
+ * ====================================================================
+ *
+ * This software consists of voluntary contributions made by many
+ * individuals on behalf of the Apache Software Foundation. For more
+ * information on the Apache Software Foundation, please see
+ * <http://www.apache.org/>.
+ *
+ */
+
+package org.apache.hc.core5.http.impl.io;
+
+import java.io.IOException;
+
+/**
+ * Signals an early (out of order) response.
+ */
+class ResponseOutOfOrderException extends IOException {
+
+ public ResponseOutOfOrderException() {
+ super();
+ }
+
+}
\ No newline at end of file
diff --git a/httpcore5/src/main/java/org/apache/hc/core5/http/io/ResponseOutOfOrderStrategy.java b/httpcore5/src/main/java/org/apache/hc/core5/http/io/ResponseOutOfOrderStrategy.java
new file mode 100644
index 0000000..ff03f68
--- /dev/null
+++ b/httpcore5/src/main/java/org/apache/hc/core5/http/io/ResponseOutOfOrderStrategy.java
@@ -0,0 +1,66 @@
+/*
+ * ====================================================================
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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.
+ * ====================================================================
+ *
+ * This software consists of voluntary contributions made by many
+ * individuals on behalf of the Apache Software Foundation. For more
+ * information on the Apache Software Foundation, please see
+ * <http://www.apache.org/>.
+ *
+ */
+
+package org.apache.hc.core5.http.io;
+
+import org.apache.hc.core5.annotation.Internal;
+import org.apache.hc.core5.http.ClassicHttpRequest;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * Represents a strategy to determine how frequently the client should check for an out of order response.
+ * An out of order response is sent before the server has read the full request. If the client fails to
+ * check for an early response then a {@link java.net.SocketException} or {@link java.net.SocketTimeoutException}
+ * may be thrown while writing the request entity after a timeout is reached on either the client or server.
+ *
+ * @since 5.1
+ */
+@Internal
+public interface ResponseOutOfOrderStrategy {
+
+ /**
+ * Called before each write to the to a socket {@link java.io.OutputStream} with the number of
+ * bytes that have already been sent, and the size of the write that will occur if this check
+ * does not encounter an out of order response.
+ *
+ * @param request The current request.
+ * @param connection The connection used to send the current request.
+ * @param inputStream The response stream, this may be used to check for an early response.
+ * @param totalBytesSent Number of bytes that have already been sent.
+ * @param nextWriteSize The size of a socket write operation that will follow this check.
+ * @return True if an early response was detected, otherwise false.
+ * @throws IOException in case of a network failure while checking for an early response.
+ */
+ boolean isEarlyResponseDetected(
+ ClassicHttpRequest request,
+ HttpClientConnection connection,
+ InputStream inputStream,
+ long totalBytesSent,
+ long nextWriteSize) throws IOException;
+}
diff --git a/httpcore5/src/main/java/org/apache/hc/core5/util/Timeout.java b/httpcore5/src/main/java/org/apache/hc/core5/util/Timeout.java
index f6a835a..77619f6 100644
--- a/httpcore5/src/main/java/org/apache/hc/core5/util/Timeout.java
+++ b/httpcore5/src/main/java/org/apache/hc/core5/util/Timeout.java
@@ -47,6 +47,11 @@ public class Timeout extends TimeValue {
public static final Timeout ZERO_MILLISECONDS = Timeout.of(0, TimeUnit.MILLISECONDS);
/**
+ * A one milliseconds {@link Timeout}.
+ */
+ public static final Timeout ONE_MILLISECOND = Timeout.of(1, TimeUnit.MILLISECONDS);
+
+ /**
* A disabled timeout represented as 0 {@code MILLISECONDS}.
*/
public static final Timeout DISABLED = ZERO_MILLISECONDS;
diff --git a/httpcore5/src/test/java/org/apache/hc/core5/http/impl/io/TestMonitoringResponseOutOfOrderStrategy.java b/httpcore5/src/test/java/org/apache/hc/core5/http/impl/io/TestMonitoringResponseOutOfOrderStrategy.java
new file mode 100644
index 0000000..4c8966e
--- /dev/null
+++ b/httpcore5/src/test/java/org/apache/hc/core5/http/impl/io/TestMonitoringResponseOutOfOrderStrategy.java
@@ -0,0 +1,133 @@
+/*
+ * ====================================================================
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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.
+ * ====================================================================
+ *
+ * This software consists of voluntary contributions made by many
+ * individuals on behalf of the Apache Software Foundation. For more
+ * information on the Apache Software Foundation, please see
+ * <http://www.apache.org/>.
+ *
+ */
+
+package org.apache.hc.core5.http.impl.io;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+import javax.net.ssl.SSLSession;
+
+import org.apache.hc.core5.http.ClassicHttpRequest;
+import org.apache.hc.core5.http.io.HttpClientConnection;
+import org.apache.hc.core5.http.io.ResponseOutOfOrderStrategy;
+import org.apache.hc.core5.http.message.BasicClassicHttpRequest;
+import org.apache.hc.core5.util.Timeout;
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.ArgumentMatchers;
+import org.mockito.Mockito;
+
+public class TestMonitoringResponseOutOfOrderStrategy {
+
+ private static final ClassicHttpRequest REQUEST = new BasicClassicHttpRequest("POST", "/path");
+
+ @Test
+ public void testFirstByteIsNotCheckedSsl() throws IOException {
+ final boolean earlyResponse = MonitoringResponseOutOfOrderStrategy.INSTANCE.isEarlyResponseDetected(
+ REQUEST,
+ connection(true, true),
+ // SSLSocket streams report zero bytes available
+ socketInputStream(0),
+ 0,
+ 1);
+ Assert.assertFalse(earlyResponse);
+ }
+
+ @Test
+ public void testFirstByteIsNotCheckedPlain() throws IOException {
+ final boolean earlyResponse = MonitoringResponseOutOfOrderStrategy.INSTANCE.isEarlyResponseDetected(
+ REQUEST,
+ connection(true, false),
+ socketInputStream(1),
+ 0,
+ 1);
+ Assert.assertFalse(earlyResponse);
+ }
+
+ @Test
+ public void testWritesWithinChunkAreNotChecked() throws IOException {
+ final boolean earlyResponse = MonitoringResponseOutOfOrderStrategy.INSTANCE.isEarlyResponseDetected(
+ REQUEST,
+ connection(true, true),
+ socketInputStream(0),
+ 1,
+ 8190);
+ Assert.assertFalse(
+ "There is data available, but checks shouldn't occur until just prior to the 8192nd byte",
+ earlyResponse);
+ }
+
+ @Test
+ public void testWritesAcrossChunksAreChecked() throws IOException {
+ final boolean earlyResponse = MonitoringResponseOutOfOrderStrategy.INSTANCE.isEarlyResponseDetected(
+ REQUEST,
+ connection(true, true),
+ socketInputStream(0),
+ 8191,
+ 1);
+ Assert.assertTrue(earlyResponse);
+ }
+
+ @Test
+ public void testMaximumChunks() throws IOException {
+ final ResponseOutOfOrderStrategy strategy = new MonitoringResponseOutOfOrderStrategy(1, 2);
+ Assert.assertTrue(strategy.isEarlyResponseDetected(
+ REQUEST,
+ connection(true, true),
+ socketInputStream(0),
+ 0,
+ 1));
+ Assert.assertTrue(strategy.isEarlyResponseDetected(
+ REQUEST,
+ connection(true, true),
+ socketInputStream(0),
+ 1,
+ 2));
+ Assert.assertFalse(strategy.isEarlyResponseDetected(
+ REQUEST,
+ connection(true, true),
+ socketInputStream(0),
+ 2,
+ 3));
+ }
+
+ private static InputStream socketInputStream(final int available) throws IOException {
+ final InputStream stream = Mockito.mock(InputStream.class);
+ Mockito.when(stream.available()).thenReturn(available);
+ return stream;
+ }
+
+ private static HttpClientConnection connection(final boolean dataAvailable, final boolean ssl) throws IOException {
+ final HttpClientConnection connection = Mockito.mock(HttpClientConnection.class);
+ Mockito.when(connection.isDataAvailable(ArgumentMatchers.any(Timeout.class))).thenReturn(dataAvailable);
+ if (ssl) {
+ Mockito.when(connection.getSSLSession()).thenReturn(Mockito.mock(SSLSession.class));
+ }
+ return connection;
+ }
+}