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 2014/05/11 16:09:31 UTC
svn commit: r1593793 [1/2] - in /httpcomponents/httpcore/trunk:
httpcore-nio/src/main/java/org/apache/http/nio/protocol/
httpcore-nio/src/test/java/org/apache/http/nio/integration/
httpcore-nio/src/test/java/org/apache/http/nio/protocol/ httpcore-nio/s...
Author: olegk
Date: Sun May 11 14:09:30 2014
New Revision: 1593793
URL: http://svn.apache.org/r1593793
Log:
Initial implementation of the client side request pipelining; many bug fixes in the server side implementation; unit and integration tests
Added:
httpcomponents/httpcore/trunk/httpcore-nio/src/main/java/org/apache/http/nio/protocol/Pipelined.java
- copied, changed from r1590623, httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/RndTestPatternGenerator.java
httpcomponents/httpcore/trunk/httpcore-nio/src/main/java/org/apache/http/nio/protocol/PipeliningClientExchangeHandler.java
- copied, changed from r1590623, httpcomponents/httpcore/trunk/httpcore-nio/src/main/java/org/apache/http/nio/protocol/BasicAsyncClientExchangeHandler.java
httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/TestHttpAsyncHandlersPipelining.java (with props)
httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/TestHttpsAsyncHandlersPipelining.java
- copied, changed from r1590623, httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/RndTestPatternGenerator.java
httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/TestServerSidePipelining.java (contents, props changed)
- copied, changed from r1590623, httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/TestPipelining.java
Removed:
httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/TestPipelining.java
Modified:
httpcomponents/httpcore/trunk/httpcore-nio/src/main/java/org/apache/http/nio/protocol/BasicAsyncClientExchangeHandler.java
httpcomponents/httpcore/trunk/httpcore-nio/src/main/java/org/apache/http/nio/protocol/HttpAsyncRequestExecutor.java
httpcomponents/httpcore/trunk/httpcore-nio/src/main/java/org/apache/http/nio/protocol/HttpAsyncRequester.java
httpcomponents/httpcore/trunk/httpcore-nio/src/main/java/org/apache/http/nio/protocol/HttpAsyncService.java
httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/RndTestPatternGenerator.java
httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/SimpleRequestHandler.java
httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/protocol/TestHttpAsyncRequestExecutor.java
httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/protocol/TestHttpAsyncService.java
httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/testserver/HttpClientNio.java
httpcomponents/httpcore/trunk/httpcore/src/main/java/org/apache/http/util/Args.java
httpcomponents/httpcore/trunk/httpcore/src/main/java/org/apache/http/util/Asserts.java
Modified: httpcomponents/httpcore/trunk/httpcore-nio/src/main/java/org/apache/http/nio/protocol/BasicAsyncClientExchangeHandler.java
URL: http://svn.apache.org/viewvc/httpcomponents/httpcore/trunk/httpcore-nio/src/main/java/org/apache/http/nio/protocol/BasicAsyncClientExchangeHandler.java?rev=1593793&r1=1593792&r2=1593793&view=diff
==============================================================================
--- httpcomponents/httpcore/trunk/httpcore-nio/src/main/java/org/apache/http/nio/protocol/BasicAsyncClientExchangeHandler.java (original)
+++ httpcomponents/httpcore/trunk/httpcore-nio/src/main/java/org/apache/http/nio/protocol/BasicAsyncClientExchangeHandler.java Sun May 11 14:09:30 2014
@@ -140,6 +140,9 @@ public class BasicAsyncClientExchangeHan
@Override
public HttpRequest generateRequest() throws IOException, HttpException {
+ if (isDone()) {
+ return null;
+ }
final HttpRequest request = this.requestProducer.generateRequest();
this.localContext.setAttribute(HttpCoreContext.HTTP_REQUEST, request);
this.localContext.setAttribute(HttpCoreContext.HTTP_CONNECTION, this.conn);
@@ -182,7 +185,7 @@ public class BasicAsyncClientExchangeHan
this.responseConsumer.responseCompleted(this.localContext);
final T result = this.responseConsumer.getResult();
final Exception ex = this.responseConsumer.getException();
- if (ex == null) {
+ if (result != null) {
this.future.completed(result);
} else {
this.future.failed(ex);
Modified: httpcomponents/httpcore/trunk/httpcore-nio/src/main/java/org/apache/http/nio/protocol/HttpAsyncRequestExecutor.java
URL: http://svn.apache.org/viewvc/httpcomponents/httpcore/trunk/httpcore-nio/src/main/java/org/apache/http/nio/protocol/HttpAsyncRequestExecutor.java?rev=1593793&r1=1593792&r2=1593793&view=diff
==============================================================================
--- httpcomponents/httpcore/trunk/httpcore-nio/src/main/java/org/apache/http/nio/protocol/HttpAsyncRequestExecutor.java (original)
+++ httpcomponents/httpcore/trunk/httpcore-nio/src/main/java/org/apache/http/nio/protocol/HttpAsyncRequestExecutor.java Sun May 11 14:09:30 2014
@@ -31,12 +31,15 @@ import java.io.IOException;
import java.net.SocketTimeoutException;
import org.apache.http.ConnectionClosedException;
+import org.apache.http.HttpEntity;
import org.apache.http.HttpEntityEnclosingRequest;
import org.apache.http.HttpException;
import org.apache.http.HttpRequest;
import org.apache.http.HttpResponse;
import org.apache.http.HttpStatus;
+import org.apache.http.HttpVersion;
import org.apache.http.ProtocolException;
+import org.apache.http.ProtocolVersion;
import org.apache.http.annotation.Immutable;
import org.apache.http.nio.ContentDecoder;
import org.apache.http.nio.ContentEncoder;
@@ -136,40 +139,55 @@ public class HttpAsyncRequestExecutor im
@Override
public void requestReady(
final NHttpClientConnection conn) throws IOException, HttpException {
- final State state = ensureNotNull(getState(conn));
+ final State state = getState(conn);
+ Asserts.notNull(state, "Connection state");
+ Asserts.check(state.getRequestState() == MessageState.READY ||
+ state.getRequestState() == MessageState.COMPLETED,
+ "Unexpected request state %s", state.getRequestState());
+
if (state.getRequestState() != MessageState.READY) {
return;
}
- HttpAsyncClientExchangeHandler handler = getHandler(conn);
- if (handler != null && handler.isDone()) {
- closeHandler(handler);
- state.reset();
- handler = null;
- }
- if (handler == null) {
+ final HttpAsyncClientExchangeHandler handler = getHandler(conn);
+ if (handler == null || handler.isDone()) {
return;
}
+ final boolean pipelined = handler.getClass().getAnnotation(Pipelined.class) != null;
final HttpRequest request = handler.generateRequest();
if (request == null) {
return;
}
+ final ProtocolVersion version = request.getRequestLine().getProtocolVersion();
+ if (pipelined && version.lessEquals(HttpVersion.HTTP_1_0)) {
+ throw new ProtocolException(version + " cannot be used with request pipelining");
+ }
state.setRequest(request);
- conn.submitRequest(request);
-
if (request instanceof HttpEntityEnclosingRequest) {
- if (((HttpEntityEnclosingRequest) request).expectContinue()) {
+ final boolean expectContinue = ((HttpEntityEnclosingRequest) request).expectContinue();
+ if (expectContinue && pipelined) {
+ throw new ProtocolException("Expect-continue handshake cannot be used with request pipelining");
+ }
+ conn.submitRequest(request);
+ if (expectContinue) {
final int timeout = conn.getSocketTimeout();
state.setTimeout(timeout);
conn.setSocketTimeout(this.waitForContinue);
state.setRequestState(MessageState.ACK_EXPECTED);
} else {
- state.setRequestState(MessageState.BODY_STREAM);
+ final HttpEntity entity = ((HttpEntityEnclosingRequest) request).getEntity();
+ if (entity != null) {
+ state.setRequestState(MessageState.BODY_STREAM);
+ } else {
+ handler.requestCompleted();
+ state.setRequestState(pipelined ? MessageState.READY : MessageState.COMPLETED);
+ }
}
} else {
+ conn.submitRequest(request);
handler.requestCompleted();
- state.setRequestState(MessageState.COMPLETED);
+ state.setRequestState(pipelined ? MessageState.READY : MessageState.COMPLETED);
}
}
@@ -177,29 +195,40 @@ public class HttpAsyncRequestExecutor im
public void outputReady(
final NHttpClientConnection conn,
final ContentEncoder encoder) throws IOException, HttpException {
- final State state = ensureNotNull(getState(conn));
- final HttpAsyncClientExchangeHandler handler = ensureNotNull(getHandler(conn));
+ final State state = getState(conn);
+ Asserts.notNull(state, "Connection state");
+ Asserts.check(state.getRequestState() == MessageState.BODY_STREAM ||
+ state.getRequestState() == MessageState.ACK_EXPECTED,
+ "Unexpected request state %s", state.getRequestState());
+
+ final HttpAsyncClientExchangeHandler handler = getHandler(conn);
+ Asserts.notNull(handler, "Client exchange handler");
if (state.getRequestState() == MessageState.ACK_EXPECTED) {
conn.suspendOutput();
return;
}
handler.produceContent(encoder, conn);
- state.setRequestState(MessageState.BODY_STREAM);
if (encoder.isCompleted()) {
handler.requestCompleted();
- state.setRequestState(MessageState.COMPLETED);
+ final boolean pipelined = handler.getClass().getAnnotation(Pipelined.class) != null;
+ state.setRequestState(pipelined ? MessageState.READY : MessageState.COMPLETED);
}
}
@Override
public void responseReceived(
final NHttpClientConnection conn) throws HttpException, IOException {
- final State state = ensureNotNull(getState(conn));
+ final State state = getState(conn);
+ Asserts.notNull(state, "Connection state");
+ Asserts.check(state.getResponseState() == MessageState.READY,
+ "Unexpected request state %s", state.getResponseState());
+
final HttpRequest request = state.getRequest();
if (request == null) {
throw new HttpException("Out of sequence response");
}
- final HttpAsyncClientExchangeHandler handler = ensureNotNull(getHandler(conn));
+ final HttpAsyncClientExchangeHandler handler = getHandler(conn);
+ Asserts.notNull(handler, "Client exchange handler");
final HttpResponse response = conn.getHttpResponse();
final int statusCode = response.getStatusLine().getStatusCode();
@@ -213,7 +242,7 @@ public class HttpAsyncRequestExecutor im
final int timeout = state.getTimeout();
conn.setSocketTimeout(timeout);
conn.requestOutput();
- state.setRequestState(MessageState.ACK);
+ state.setRequestState(MessageState.BODY_STREAM);
}
return;
}
@@ -245,10 +274,14 @@ public class HttpAsyncRequestExecutor im
public void inputReady(
final NHttpClientConnection conn,
final ContentDecoder decoder) throws IOException, HttpException {
- final State state = ensureNotNull(getState(conn));
- final HttpAsyncClientExchangeHandler handler = ensureNotNull(getHandler(conn));
+ final State state = getState(conn);
+ Asserts.notNull(state, "Connection state");
+ Asserts.check(state.getResponseState() == MessageState.BODY_STREAM,
+ "Unexpected request state %s", state.getResponseState());
+
+ final HttpAsyncClientExchangeHandler handler = getHandler(conn);
+ Asserts.notNull(handler, "Client exchange handler");
handler.consumeContent(decoder, conn);
- state.setResponseState(MessageState.BODY_STREAM);
if (decoder.isCompleted()) {
processResponse(conn, state, handler);
}
@@ -290,6 +323,7 @@ public class HttpAsyncRequestExecutor im
conn.setSocketTimeout(timeout);
conn.requestOutput();
state.setRequestState(MessageState.BODY_STREAM);
+ state.setTimeout(0);
return;
} else {
state.invalidate();
@@ -326,20 +360,10 @@ public class HttpAsyncRequestExecutor im
return (State) conn.getContext().getAttribute(HTTP_EXCHANGE_STATE);
}
- private State ensureNotNull(final State state) {
- Asserts.notNull(state, "HTTP exchange state");
- return state;
- }
-
private HttpAsyncClientExchangeHandler getHandler(final NHttpConnection conn) {
return (HttpAsyncClientExchangeHandler) conn.getContext().getAttribute(HTTP_HANDLER);
}
- private HttpAsyncClientExchangeHandler ensureNotNull(final HttpAsyncClientExchangeHandler handler) {
- Asserts.notNull(handler, "HTTP exchange handler");
- return handler;
- }
-
private void shutdownConnection(final NHttpConnection conn) {
try {
conn.shutdown();
@@ -366,7 +390,14 @@ public class HttpAsyncRequestExecutor im
conn.close();
}
handler.responseCompleted();
- state.reset();
+
+ final boolean pipelined = handler.getClass().getAnnotation(Pipelined.class) != null;
+ if (!pipelined) {
+ state.setRequestState(MessageState.READY);
+ state.setRequest(null);
+ }
+ state.setResponseState(MessageState.READY);
+ state.setResponse(null);
if (!handler.isDone() && conn.isOpen()) {
conn.requestOutput();
}
@@ -447,14 +478,6 @@ public class HttpAsyncRequestExecutor im
this.timeout = timeout;
}
- public void reset() {
- this.responseState = MessageState.READY;
- this.requestState = MessageState.READY;
- this.response = null;
- this.request = null;
- this.timeout = 0;
- }
-
public boolean isValid() {
return this.valid;
}
Modified: httpcomponents/httpcore/trunk/httpcore-nio/src/main/java/org/apache/http/nio/protocol/HttpAsyncRequester.java
URL: http://svn.apache.org/viewvc/httpcomponents/httpcore/trunk/httpcore-nio/src/main/java/org/apache/http/nio/protocol/HttpAsyncRequester.java?rev=1593793&r1=1593792&r2=1593793&view=diff
==============================================================================
--- httpcomponents/httpcore/trunk/httpcore-nio/src/main/java/org/apache/http/nio/protocol/HttpAsyncRequester.java (original)
+++ httpcomponents/httpcore/trunk/httpcore-nio/src/main/java/org/apache/http/nio/protocol/HttpAsyncRequester.java Sun May 11 14:09:30 2014
@@ -26,7 +26,9 @@
*/
package org.apache.http.nio.protocol;
+import java.io.Closeable;
import java.io.IOException;
+import java.util.List;
import java.util.concurrent.Future;
import org.apache.http.ConnectionClosedException;
@@ -100,8 +102,8 @@ public class HttpAsyncRequester {
* Initiates asynchronous HTTP request execution.
*
* @param <T> the result type of request execution.
- * @param requestProducer request producer callback.
- * @param responseConsumer response consumer callaback.
+ * @param requestProducer request producer.
+ * @param responseConsumer response consumer.
* @param conn underlying HTTP connection.
* @param context HTTP context
* @param callback future callback.
@@ -143,8 +145,8 @@ public class HttpAsyncRequester {
* Initiates asynchronous HTTP request execution.
*
* @param <T> the result type of request execution.
- * @param requestProducer request producer callback.
- * @param responseConsumer response consumer callaback.
+ * @param requestProducer request producer.
+ * @param responseConsumer response consumer.
* @param conn underlying HTTP connection.
* @param context HTTP context
* @return future representing pending completion of the operation.
@@ -161,8 +163,8 @@ public class HttpAsyncRequester {
* Initiates asynchronous HTTP request execution.
*
* @param <T> the result type of request execution.
- * @param requestProducer request producer callback.
- * @param responseConsumer response consumer callaback.
+ * @param requestProducer request producer.
+ * @param responseConsumer response consumer.
* @param conn underlying HTTP connection.
* @return future representing pending completion of the operation.
*/
@@ -178,8 +180,8 @@ public class HttpAsyncRequester {
*
* @param <T> the result type of request execution.
* @param <E> the connection pool entry type.
- * @param requestProducer request producer callback.
- * @param responseConsumer response consumer callaback.
+ * @param requestProducer request producer.
+ * @param responseConsumer response consumer.
* @param connPool pool of persistent reusable connections.
* @param context HTTP context
* @param callback future callback.
@@ -203,13 +205,46 @@ public class HttpAsyncRequester {
}
/**
+ * Initiates asynchronous pipelined HTTP request execution.
+ *
+ * @param <T> the result type of request execution.
+ * @param <E> the connection pool entry type.
+ * @param target target host.
+ * @param requestProducers list of request producers.
+ * @param responseConsumers list of response consumers.
+ * @param connPool pool of persistent reusable connections.
+ * @param context HTTP context
+ * @param callback future callback.
+ * @return future representing pending completion of the operation.
+ *
+ * @since 4.4
+ */
+ public <T, E extends PoolEntry<HttpHost, NHttpClientConnection>> Future<List<T>> executePipelined(
+ final HttpHost target,
+ final List<HttpAsyncRequestProducer> requestProducers,
+ final List<HttpAsyncResponseConsumer<T>> responseConsumers,
+ final ConnPool<HttpHost, E> connPool,
+ final HttpContext context,
+ final FutureCallback<List<T>> callback) {
+ Args.notNull(target, "HTTP target");
+ Args.notEmpty(requestProducers, "Request producer list");
+ Args.notEmpty(responseConsumers, "Response consumer list");
+ Args.notNull(connPool, "HTTP connection pool");
+ Args.notNull(context, "HTTP context");
+ final BasicFuture<List<T>> future = new BasicFuture<List<T>>(callback);
+ connPool.lease(target, null, new ConnPipelinedRequestCallback<T, E>(
+ future, requestProducers, responseConsumers, connPool, context));
+ return future;
+ }
+
+ /**
* Initiates asynchronous HTTP request execution. This method automatically releases
* the given pool entry once request execution is completed (successfully or unsuccessfully).
*
* @param <T> the result type of request execution.
* @param <E> the connection pool entry type.
- * @param requestProducer request producer callback.
- * @param responseConsumer response consumer callaback.
+ * @param requestProducer request producer.
+ * @param responseConsumer response consumer.
* @param poolEntry leased pool entry. It will be automatically released
* back to the pool when execution is completed.
* @param connPool pool of persistent reusable connections.
@@ -243,12 +278,52 @@ public class HttpAsyncRequester {
}
/**
+ * Initiates asynchronous pipelined HTTP request execution. This method automatically releases
+ * the given pool entry once request execution is completed (successfully or unsuccessfully).
+ *
+ * @param <T> the result type of request execution.
+ * @param <E> the connection pool entry type.
+ * @param requestProducers list of request producers.
+ * @param responseConsumer list of response consumers.
+ * @param poolEntry leased pool entry. It will be automatically released
+ * back to the pool when execution is completed.
+ * @param connPool pool of persistent reusable connections.
+ * @param context HTTP context
+ * @param callback future callback.
+ * @return future representing pending completion of the operation.
+ *
+ * @since 4.4
+ */
+ public <T, E extends PoolEntry<HttpHost, NHttpClientConnection>> Future<List<T>> executePipelined(
+ final List<HttpAsyncRequestProducer> requestProducers,
+ final List<HttpAsyncResponseConsumer<T>> responseConsumers,
+ final E poolEntry,
+ final ConnPool<HttpHost, E> connPool,
+ final HttpContext context,
+ final FutureCallback<List<T>> callback) {
+ Args.notEmpty(requestProducers, "Request producer list");
+ Args.notEmpty(responseConsumers, "Response consumer list");
+ Args.notNull(connPool, "HTTP connection pool");
+ Args.notNull(poolEntry, "Pool entry");
+ Args.notNull(context, "HTTP context");
+ final BasicFuture<List<T>> future = new BasicFuture<List<T>>(callback);
+ final NHttpClientConnection conn = poolEntry.getConnection();
+ final PipeliningClientExchangeHandler<T> handler = new PipeliningClientExchangeHandler<T>(
+ requestProducers, responseConsumers,
+ new RequestExecutionCallback<List<T>, E>(future, poolEntry, connPool),
+ context, conn,
+ this.httpprocessor, this.connReuseStrategy);
+ initExection(handler, conn);
+ return future;
+ }
+
+ /**
* Initiates asynchronous HTTP request execution.
*
* @param <T> the result type of request execution.
* @param <E> the connection pool entry type.
- * @param requestProducer request producer callback.
- * @param responseConsumer response consumer callaback.
+ * @param requestProducer request producer.
+ * @param responseConsumer response consumer.
* @param connPool pool of persistent reusable connections.
* @param context HTTP context
* @return future representing pending completion of the operation.
@@ -266,8 +341,8 @@ public class HttpAsyncRequester {
*
* @param <T> the result type of request execution.
* @param <E> the connection pool entry type.
- * @param requestProducer request producer callback.
- * @param responseConsumer response consumer callaback.
+ * @param requestProducer request producer.
+ * @param responseConsumer response consumer.
* @param connPool pool of persistent reusable connections.
* @return future representing pending completion of the operation.
*/
@@ -341,15 +416,84 @@ public class HttpAsyncRequester {
}
public void releaseResources() {
+ close(requestProducer);
+ close(responseConsumer);
+ }
+
+ }
+
+ class ConnPipelinedRequestCallback<T, E extends PoolEntry<HttpHost, NHttpClientConnection>> implements FutureCallback<E> {
+
+ private final BasicFuture<List<T>> requestFuture;
+ private final List<HttpAsyncRequestProducer> requestProducers;
+ private final List<HttpAsyncResponseConsumer<T>> responseConsumers;
+ private final ConnPool<HttpHost, E> connPool;
+ private final HttpContext context;
+
+ ConnPipelinedRequestCallback(
+ final BasicFuture<List<T>> requestFuture,
+ final List<HttpAsyncRequestProducer> requestProducers,
+ final List<HttpAsyncResponseConsumer<T>> responseConsumers,
+ final ConnPool<HttpHost, E> connPool,
+ final HttpContext context) {
+ super();
+ this.requestFuture = requestFuture;
+ this.requestProducers = requestProducers;
+ this.responseConsumers = responseConsumers;
+ this.connPool = connPool;
+ this.context = context;
+ }
+
+ @Override
+ public void completed(final E result) {
+ if (this.requestFuture.isDone()) {
+ this.connPool.release(result, true);
+ return;
+ }
+ final NHttpClientConnection conn = result.getConnection();
+ final PipeliningClientExchangeHandler<T> handler = new PipeliningClientExchangeHandler<T>(
+ this.requestProducers, this.responseConsumers,
+ new RequestExecutionCallback<List<T>, E>(this.requestFuture, result, this.connPool),
+ this.context, conn, httpprocessor, connReuseStrategy);
+ initExection(handler, conn);
+ }
+
+ @Override
+ public void failed(final Exception ex) {
try {
- this.requestProducer.close();
- } catch (final IOException ioex) {
- log(ioex);
+ try {
+ for (HttpAsyncResponseConsumer<T> responseConsumer: this.responseConsumers) {
+ responseConsumer.failed(ex);
+ }
+ } finally {
+ releaseResources();
+ }
+ } finally {
+ this.requestFuture.failed(ex);
}
+ }
+
+ @Override
+ public void cancelled() {
try {
- this.responseConsumer.close();
- } catch (final IOException ioex) {
- log(ioex);
+ try {
+ for (HttpAsyncResponseConsumer<T> responseConsumer: this.responseConsumers) {
+ responseConsumer.cancel();
+ }
+ } finally {
+ releaseResources();
+ }
+ } finally {
+ this.requestFuture.cancel(true);
+ }
+ }
+
+ public void releaseResources() {
+ for (HttpAsyncRequestProducer requestProducer: this.requestProducers) {
+ close(requestProducer);
+ }
+ for (HttpAsyncResponseConsumer<T> responseConsumer: this.responseConsumers) {
+ close(responseConsumer);
}
}
@@ -411,4 +555,12 @@ public class HttpAsyncRequester {
protected void log(final Exception ex) {
}
+ private void close(final Closeable closeable) {
+ try {
+ closeable.close();
+ } catch (IOException ex) {
+ log(ex);
+ }
+ }
+
}
Modified: httpcomponents/httpcore/trunk/httpcore-nio/src/main/java/org/apache/http/nio/protocol/HttpAsyncService.java
URL: http://svn.apache.org/viewvc/httpcomponents/httpcore/trunk/httpcore-nio/src/main/java/org/apache/http/nio/protocol/HttpAsyncService.java?rev=1593793&r1=1593792&r2=1593793&view=diff
==============================================================================
--- httpcomponents/httpcore/trunk/httpcore-nio/src/main/java/org/apache/http/nio/protocol/HttpAsyncService.java (original)
+++ httpcomponents/httpcore/trunk/httpcore-nio/src/main/java/org/apache/http/nio/protocol/HttpAsyncService.java Sun May 11 14:09:30 2014
@@ -212,7 +212,7 @@ public class HttpAsyncService implements
@Override
public void closed(final NHttpServerConnection conn) {
- final State state = getState(conn);
+ final State state = (State) conn.getContext().removeAttribute(HTTP_EXCHANGE_STATE);
if (state != null) {
state.setTerminated();
closeHandlers(state);
@@ -220,9 +220,6 @@ public class HttpAsyncService implements
if (cancellable != null) {
cancellable.cancel();
}
- state.setRequestState(MessageState.READY);
- state.setResponseState(MessageState.READY);
- conn.getContext().removeAttribute(HTTP_EXCHANGE_STATE);
}
}
@@ -276,6 +273,9 @@ public class HttpAsyncService implements
final NHttpServerConnection conn) throws IOException, HttpException {
final State state = getState(conn);
Asserts.notNull(state, "Connection state");
+ Asserts.check(state.getRequestState() == MessageState.READY,
+ "Unexpected request state %s", state.getRequestState());
+
final HttpRequest request = conn.getHttpRequest();
final HttpContext context = new BasicHttpContext();
@@ -304,6 +304,7 @@ public class HttpAsyncService implements
HttpStatus.SC_CONTINUE, context);
if (this.expectationVerifier != null) {
conn.suspendInput();
+ conn.suspendOutput();
final HttpAsyncExchange httpAsyncExchange = new HttpAsyncExchangeImpl(
request, ack, state, conn, context);
this.expectationVerifier.verify(httpAsyncExchange, context);
@@ -326,11 +327,13 @@ public class HttpAsyncService implements
final ContentDecoder decoder) throws IOException, HttpException {
final State state = getState(conn);
Asserts.notNull(state, "Connection state");
+ Asserts.check(state.getRequestState() == MessageState.BODY_STREAM,
+ "Unexpected request state %s", state.getRequestState());
+
final Incoming incoming = state.getIncoming();
Asserts.notNull(incoming, "Incoming request");
final HttpAsyncRequestConsumer<?> consumer = incoming.getConsumer();
consumer.consumeContent(decoder, conn);
- state.setRequestState(MessageState.BODY_STREAM);
if (decoder.isCompleted()) {
completeRequest(incoming, conn, state);
}
@@ -341,22 +344,16 @@ public class HttpAsyncService implements
final NHttpServerConnection conn) throws IOException, HttpException {
final State state = getState(conn);
Asserts.notNull(state, "Connection state");
- Outgoing outgoing = state.getOutgoing();
- if (outgoing == null) {
- final Queue<PipelineEntry> pipeline = state.getPipeline();
- final PipelineEntry pipelineEntry = pipeline.poll();
- if (pipelineEntry == null) {
- return;
- }
- outgoing = respondToRequest(pipelineEntry, conn, state);
- if (outgoing == null) {
- return;
- }
- state.setOutgoing(outgoing);
- }
- final HttpResponse response = outgoing.getResponse();
- final int status = response.getStatusLine().getStatusCode();
+ Asserts.check(state.getResponseState() == MessageState.READY ||
+ state.getResponseState() == MessageState.INIT,
+ "Unexpected response state %s", state.getResponseState());
+
if (state.getRequestState() == MessageState.ACK_EXPECTED) {
+ final Outgoing outgoing = state.getOutgoing();
+ Asserts.notNull(outgoing, "Outgoing response");
+
+ final HttpResponse response = outgoing.getResponse();
+ final int status = response.getStatusLine().getStatusCode();
if (status == 100) {
final HttpContext context = outgoing.getContext();
final HttpAsyncResponseProducer responseProducer = outgoing.getProducer();
@@ -379,6 +376,40 @@ public class HttpAsyncService implements
throw new HttpException("Invalid response: " + response.getStatusLine());
}
} else {
+ if (state.getResponseState() == MessageState.READY) {
+ final Queue<PipelineEntry> pipeline = state.getPipeline();
+ final PipelineEntry pipelineEntry = pipeline.poll();
+ if (pipelineEntry == null) {
+ conn.suspendOutput();
+ return;
+ }
+ final Object result = pipelineEntry.getResult();
+ final HttpRequest request = pipelineEntry.getRequest();
+ final HttpContext context = pipelineEntry.getContext();
+ if (result != null) {
+ state.setResponseState(MessageState.INIT);
+ final HttpResponse response = this.responseFactory.newHttpResponse(HttpVersion.HTTP_1_1,
+ HttpStatus.SC_OK, context);
+ final HttpAsyncExchangeImpl httpExchange = new HttpAsyncExchangeImpl(
+ request, response, state, conn, context);
+ final HttpAsyncRequestHandler<Object> handler = pipelineEntry.getHandler();
+ conn.suspendOutput();
+ handler.handle(result, httpExchange, context);
+ } else {
+ final Exception exception = pipelineEntry.getException();
+ final HttpAsyncResponseProducer responseProducer = handleException(
+ exception != null ? exception : new HttpException("Internal error processing request"),
+ context);
+ final HttpResponse error = responseProducer.generateResponse();
+ state.setOutgoing(new Outgoing(request, error, responseProducer, context));
+ }
+ }
+ final Outgoing outgoing = state.getOutgoing();
+ if (outgoing == null) {
+ return;
+ }
+ final HttpResponse response = outgoing.getResponse();
+ final int status = response.getStatusLine().getStatusCode();
if (status >= 200) {
commitFinalResponse(conn, state);
} else {
@@ -393,12 +424,14 @@ public class HttpAsyncService implements
final ContentEncoder encoder) throws HttpException, IOException {
final State state = getState(conn);
Asserts.notNull(state, "Connection state");
+ Asserts.check(state.getResponseState() == MessageState.BODY_STREAM,
+ "Unexpected response state %s", state.getResponseState());
+
final Outgoing outgoing = state.getOutgoing();
Asserts.notNull(outgoing, "Outgoing response");
final HttpAsyncResponseProducer responseProducer = outgoing.getProducer();
responseProducer.produceContent(encoder, conn);
- state.setResponseState(MessageState.BODY_STREAM);
if (encoder.isCompleted()) {
completeResponse(outgoing, conn, state);
@@ -552,7 +585,7 @@ public class HttpAsyncService implements
final Incoming incoming,
final NHttpServerConnection conn,
final State state) throws IOException, HttpException {
- state.setRequestState(MessageState.COMPLETED);
+ state.setRequestState(MessageState.READY);
state.setIncoming(null);
final PipelineEntry pipelineEntry;
@@ -570,41 +603,8 @@ public class HttpAsyncService implements
consumer.close();
}
final Queue<PipelineEntry> pipeline = state.getPipeline();
- if (!pipeline.isEmpty() || state.getResponseState().compareTo(MessageState.READY) > 0) {
- pipeline.add(pipelineEntry);
- } else {
- final Outgoing outgoing = respondToRequest(pipelineEntry, conn, state);
- if (outgoing != null) {
- state.setOutgoing(outgoing);
- conn.requestOutput();
- }
- }
- }
-
- private Outgoing respondToRequest(
- final PipelineEntry entry,
- final NHttpServerConnection conn,
- final State state) throws IOException, HttpException {
- final Object result = entry.getResult();
- final HttpRequest request = entry.getRequest();
- final HttpContext context = entry.getContext();
- state.setResponseState(MessageState.INIT);
- if (result != null) {
- final HttpResponse response = this.responseFactory.newHttpResponse(HttpVersion.HTTP_1_1,
- HttpStatus.SC_OK, context);
- final HttpAsyncExchangeImpl httpExchange = new HttpAsyncExchangeImpl(
- request, response, state, conn, context);
- final HttpAsyncRequestHandler<Object> handler = entry.getHandler();
- handler.handle(result, httpExchange, context);
- return null;
- } else {
- final Exception exception = entry.getException();
- final HttpAsyncResponseProducer responseProducer = handleException(
- exception != null ? exception : new HttpException("Internal error processing request"),
- context);
- final HttpResponse error = responseProducer.generateResponse();
- return new Outgoing(request, error, responseProducer, context);
- }
+ pipeline.add(pipelineEntry);
+ conn.requestOutput();
}
private void commitFinalResponse(
Copied: httpcomponents/httpcore/trunk/httpcore-nio/src/main/java/org/apache/http/nio/protocol/Pipelined.java (from r1590623, httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/RndTestPatternGenerator.java)
URL: http://svn.apache.org/viewvc/httpcomponents/httpcore/trunk/httpcore-nio/src/main/java/org/apache/http/nio/protocol/Pipelined.java?p2=httpcomponents/httpcore/trunk/httpcore-nio/src/main/java/org/apache/http/nio/protocol/Pipelined.java&p1=httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/RndTestPatternGenerator.java&r1=1590623&r2=1593793&rev=1593793&view=diff
==============================================================================
--- httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/RndTestPatternGenerator.java (original)
+++ httpcomponents/httpcore/trunk/httpcore-nio/src/main/java/org/apache/http/nio/protocol/Pipelined.java Sun May 11 14:09:30 2014
@@ -24,27 +24,16 @@
* <http://www.apache.org/>.
*
*/
+package org.apache.http.nio.protocol;
-package org.apache.http.nio.integration;
-
-import java.util.Random;
-
-final class RndTestPatternGenerator {
-
- private static final Random RND = new Random();
- private static final String TEST_CHARS = "0123456789ABCDEF";
-
- public static String generateText() {
- final StringBuilder buffer = new StringBuilder();
- for (int i = 0; i < 5; i++) {
- final char rndchar = TEST_CHARS.charAt(RND.nextInt(TEST_CHARS.length() - 1));
- buffer.append(rndchar);
- }
- return buffer.toString();
- }
-
- public static int generateCount(final int max) {
- return RND.nextInt(max - 1) + 1;
- }
-
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Documented
+@Target({ElementType.TYPE})
+@Retention(RetentionPolicy.RUNTIME)
+public @interface Pipelined {
}
Copied: httpcomponents/httpcore/trunk/httpcore-nio/src/main/java/org/apache/http/nio/protocol/PipeliningClientExchangeHandler.java (from r1590623, httpcomponents/httpcore/trunk/httpcore-nio/src/main/java/org/apache/http/nio/protocol/BasicAsyncClientExchangeHandler.java)
URL: http://svn.apache.org/viewvc/httpcomponents/httpcore/trunk/httpcore-nio/src/main/java/org/apache/http/nio/protocol/PipeliningClientExchangeHandler.java?p2=httpcomponents/httpcore/trunk/httpcore-nio/src/main/java/org/apache/http/nio/protocol/PipeliningClientExchangeHandler.java&p1=httpcomponents/httpcore/trunk/httpcore-nio/src/main/java/org/apache/http/nio/protocol/BasicAsyncClientExchangeHandler.java&r1=1590623&r2=1593793&rev=1593793&view=diff
==============================================================================
--- httpcomponents/httpcore/trunk/httpcore-nio/src/main/java/org/apache/http/nio/protocol/BasicAsyncClientExchangeHandler.java (original)
+++ httpcomponents/httpcore/trunk/httpcore-nio/src/main/java/org/apache/http/nio/protocol/PipeliningClientExchangeHandler.java Sun May 11 14:09:30 2014
@@ -27,7 +27,12 @@
package org.apache.http.nio.protocol;
+import java.io.Closeable;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Future;
import org.apache.http.ConnectionClosedException;
@@ -46,26 +51,32 @@ import org.apache.http.protocol.HttpCont
import org.apache.http.protocol.HttpCoreContext;
import org.apache.http.protocol.HttpProcessor;
import org.apache.http.util.Args;
+import org.apache.http.util.Asserts;
/**
- * Basic implementation of {@link HttpAsyncClientExchangeHandler} that executes
+ * Basic implementation of {@link org.apache.http.nio.protocol.HttpAsyncClientExchangeHandler} that executes
* a single HTTP request / response exchange.
*
* @param <T> the result type of request execution.
- * @since 4.3
+ * @since 4.4
*/
-public class BasicAsyncClientExchangeHandler<T> implements HttpAsyncClientExchangeHandler {
+@Pipelined()
+public class PipeliningClientExchangeHandler<T> implements HttpAsyncClientExchangeHandler {
- private final HttpAsyncRequestProducer requestProducer;
- private final HttpAsyncResponseConsumer<T> responseConsumer;
- private final BasicFuture<T> future;
+ private final Queue<HttpAsyncRequestProducer> requestProducerQueue;
+ private final Queue<HttpAsyncResponseConsumer<T>> responseConsumerQueue;
+ private final Queue<HttpRequest> requestQueue;
+ private final Queue<T> resultQueue;
+ private final BasicFuture<List<T>> future;
private final HttpContext localContext;
private final NHttpClientConnection conn;
private final HttpProcessor httppocessor;
private final ConnectionReuseStrategy connReuseStrategy;
- private volatile boolean requestSent;
+ private volatile HttpAsyncRequestProducer requestProducer;
+ private volatile HttpAsyncResponseConsumer<T> responseConsumer;
private volatile boolean keepAlive;
+ private volatile boolean done;
/**
* Creates new instance of BasicAsyncRequestExecutionHandler.
@@ -78,23 +89,30 @@ public class BasicAsyncClientExchangeHan
* @param httppocessor the HTTP protocol processor.
* @param connReuseStrategy the connection re-use strategy.
*/
- public BasicAsyncClientExchangeHandler(
- final HttpAsyncRequestProducer requestProducer,
- final HttpAsyncResponseConsumer<T> responseConsumer,
- final FutureCallback<T> callback,
+ public PipeliningClientExchangeHandler(
+ final List<HttpAsyncRequestProducer> requestProducers,
+ final List<HttpAsyncResponseConsumer<T>> responseConsumers,
+ final FutureCallback<List<T>> callback,
final HttpContext localContext,
final NHttpClientConnection conn,
final HttpProcessor httppocessor,
final ConnectionReuseStrategy connReuseStrategy) {
super();
- this.requestProducer = Args.notNull(requestProducer, "Request producer");
- this.responseConsumer = Args.notNull(responseConsumer, "Response consumer");
- this.future = new BasicFuture<T>(callback);
+ Args.notEmpty(requestProducers, "Request producer list");
+ Args.notEmpty(responseConsumers, "Response consumer list");
+ Args.check(requestProducers.size() == responseConsumers.size(),
+ "Number of request producers does not match that of response consumers");
+ this.requestProducerQueue = new ConcurrentLinkedQueue<HttpAsyncRequestProducer>(requestProducers);
+ this.responseConsumerQueue = new ConcurrentLinkedQueue<HttpAsyncResponseConsumer<T>>(responseConsumers);
+ this.requestQueue = new ConcurrentLinkedQueue<HttpRequest>();
+ this.resultQueue = new ConcurrentLinkedQueue<T>();
+ this.future = new BasicFuture<List<T>>(callback);
this.localContext = Args.notNull(localContext, "HTTP context");
this.conn = Args.notNull(conn, "HTTP connection");
this.httppocessor = Args.notNull(httppocessor, "HTTP processor");
this.connReuseStrategy = connReuseStrategy != null ? connReuseStrategy :
DefaultConnectionReuseStrategy.INSTANCE;
+ this.localContext.setAttribute(HttpCoreContext.HTTP_CONNECTION, this.conn);
}
/**
@@ -106,28 +124,41 @@ public class BasicAsyncClientExchangeHan
* @param conn the actual connection.
* @param httppocessor the HTTP protocol processor.
*/
- public BasicAsyncClientExchangeHandler(
- final HttpAsyncRequestProducer requestProducer,
- final HttpAsyncResponseConsumer<T> responseConsumer,
+ public PipeliningClientExchangeHandler(
+ final List<HttpAsyncRequestProducer> requestProducers,
+ final List<HttpAsyncResponseConsumer<T>> responseConsumers,
final HttpContext localContext,
final NHttpClientConnection conn,
final HttpProcessor httppocessor) {
- this(requestProducer, responseConsumer, null, localContext, conn, httppocessor, null);
+ this(requestProducers, responseConsumers, null, localContext, conn, httppocessor, null);
}
- public Future<T> getFuture() {
+ public Future<List<T>> getFuture() {
return this.future;
}
+ private static void closeQuietly(final Closeable closeable) {
+ if (closeable != null) {
+ try {
+ closeable.close();
+ } catch (final IOException ex) {
+ }
+ }
+ }
+
private void releaseResources() {
- try {
- this.responseConsumer.close();
- } catch (final IOException ex) {
+ closeQuietly(this.requestProducer);
+ this.requestProducer = null;
+ closeQuietly(this.responseConsumer);
+ this.responseConsumer = null;
+ while (!this.requestProducerQueue.isEmpty()) {
+ closeQuietly(this.requestProducerQueue.remove());
}
- try {
- this.requestProducer.close();
- } catch (final IOException ex) {
+ while (!this.responseConsumerQueue.isEmpty()) {
+ closeQuietly(this.responseConsumerQueue.remove());
}
+ this.requestQueue.clear();
+ this.resultQueue.clear();
}
@Override
@@ -140,29 +171,45 @@ public class BasicAsyncClientExchangeHan
@Override
public HttpRequest generateRequest() throws IOException, HttpException {
+ Asserts.check(this.requestProducer == null, "Inconsistent state: request producer is not null");
+ this.requestProducer = this.requestProducerQueue.poll();
+ if (this.requestProducer == null) {
+ return null;
+ }
final HttpRequest request = this.requestProducer.generateRequest();
- this.localContext.setAttribute(HttpCoreContext.HTTP_REQUEST, request);
- this.localContext.setAttribute(HttpCoreContext.HTTP_CONNECTION, this.conn);
this.httppocessor.process(request, this.localContext);
+ this.requestQueue.add(request);
return request;
}
@Override
public void produceContent(
final ContentEncoder encoder, final IOControl ioctrl) throws IOException {
+ Asserts.check(this.requestProducer != null, "Inconsistent state: request producer is null");
this.requestProducer.produceContent(encoder, ioctrl);
}
@Override
public void requestCompleted() {
+ Asserts.check(this.requestProducer != null, "Inconsistent state: request producer is null");
this.requestProducer.requestCompleted(this.localContext);
- this.requestSent = true;
+ this.requestProducer = null;
}
@Override
public void responseReceived(final HttpResponse response) throws IOException, HttpException {
+ Asserts.check(this.responseConsumer == null, "Inconsistent state: response consumer is not null");
+
+ this.responseConsumer = this.responseConsumerQueue.poll();
+ Asserts.check(this.responseConsumer != null, "Inconsistent state: response consumer queue is empty");
+
+ final HttpRequest request = this.requestQueue.poll();
+ Asserts.check(request != null, "Inconsistent state: request queue is empty");
+
+ this.localContext.setAttribute(HttpCoreContext.HTTP_REQUEST, request);
this.localContext.setAttribute(HttpCoreContext.HTTP_RESPONSE, response);
this.httppocessor.process(response, this.localContext);
+
this.responseConsumer.responseReceived(response);
this.keepAlive = this.connReuseStrategy.keepAlive(response, this.localContext);
}
@@ -170,11 +217,13 @@ public class BasicAsyncClientExchangeHan
@Override
public void consumeContent(
final ContentDecoder decoder, final IOControl ioctrl) throws IOException {
+ Asserts.check(this.responseConsumer != null, "Inconsistent state: response consumer is null");
this.responseConsumer.consumeContent(decoder, ioctrl);
}
@Override
public void responseCompleted() throws IOException {
+ Asserts.check(this.responseConsumer != null, "Inconsistent state: response consumer is null");
try {
if (!this.keepAlive) {
this.conn.close();
@@ -182,12 +231,21 @@ public class BasicAsyncClientExchangeHan
this.responseConsumer.responseCompleted(this.localContext);
final T result = this.responseConsumer.getResult();
final Exception ex = this.responseConsumer.getException();
- if (ex == null) {
- this.future.completed(result);
+ this.responseConsumer = null;
+ if (result != null) {
+ this.resultQueue.add(result);
} else {
this.future.failed(ex);
+ this.conn.shutdown();
+ }
+ if (!conn.isOpen()) {
+ this.done = true;
+ releaseResources();
+ }
+ if (!this.future.isDone() && this.responseConsumerQueue.isEmpty()) {
+ this.future.completed(new ArrayList<T>(this.resultQueue));
+ this.resultQueue.clear();
}
- releaseResources();
} catch (final RuntimeException ex) {
failed(ex);
throw ex;
@@ -201,11 +259,14 @@ public class BasicAsyncClientExchangeHan
@Override
public void failed(final Exception ex) {
+ this.done = true;
try {
- if (!this.requestSent) {
+ if (this.requestProducer != null) {
this.requestProducer.failed(ex);
}
- this.responseConsumer.failed(ex);
+ if (this.responseConsumer != null) {
+ this.responseConsumer.failed(ex);
+ }
} finally {
try {
this.future.failed(ex);
@@ -217,6 +278,7 @@ public class BasicAsyncClientExchangeHan
@Override
public boolean cancel() {
+ this.done = true;
try {
final boolean cancelled = this.responseConsumer.cancel();
this.future.cancel();
@@ -230,7 +292,7 @@ public class BasicAsyncClientExchangeHan
@Override
public boolean isDone() {
- return this.responseConsumer.isDone();
+ return this.done;
}
}
Modified: httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/RndTestPatternGenerator.java
URL: http://svn.apache.org/viewvc/httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/RndTestPatternGenerator.java?rev=1593793&r1=1593792&r2=1593793&view=diff
==============================================================================
--- httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/RndTestPatternGenerator.java (original)
+++ httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/RndTestPatternGenerator.java Sun May 11 14:09:30 2014
@@ -47,4 +47,8 @@ final class RndTestPatternGenerator {
return RND.nextInt(max - 1) + 1;
}
+ public static boolean generateBoolean() {
+ return RND.nextBoolean();
+ }
+
}
Modified: httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/SimpleRequestHandler.java
URL: http://svn.apache.org/viewvc/httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/SimpleRequestHandler.java?rev=1593793&r1=1593792&r2=1593793&view=diff
==============================================================================
--- httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/SimpleRequestHandler.java (original)
+++ httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/SimpleRequestHandler.java Sun May 11 14:09:30 2014
@@ -60,24 +60,25 @@ final class SimpleRequestHandler impleme
final HttpResponse response,
final HttpContext context) throws HttpException, IOException {
- String content = null;
+ final String s = request.getRequestLine().getUri();
+ final int idx = s.indexOf('x');
+ if (idx == -1) {
+ throw new HttpException("Unexpected request-URI format");
+ }
+ final String pattern = s.substring(0, idx);
+ final int count = Integer.parseInt(s.substring(idx + 1, s.length()));
+ response.addHeader("Pattern", pattern);
+
+ final String content;
if (request instanceof HttpEntityEnclosingRequest) {
final HttpEntity entity = ((HttpEntityEnclosingRequest) request).getEntity();
if (entity != null) {
content = EntityUtils.toString(entity);
} else {
response.setStatusCode(HttpStatus.SC_BAD_REQUEST);
- content = "Request entity not avaialble";
+ content = "Request entity not available";
}
} else {
- final String s = request.getRequestLine().getUri();
- final int idx = s.indexOf('x');
- if (idx == -1) {
- throw new HttpException("Unexpected request-URI format");
- }
- final String pattern = s.substring(0, idx);
- final int count = Integer.parseInt(s.substring(idx + 1, s.length()));
-
final StringBuilder buffer = new StringBuilder();
for (int i = 0; i < count; i++) {
buffer.append(pattern);
Added: httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/TestHttpAsyncHandlersPipelining.java
URL: http://svn.apache.org/viewvc/httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/TestHttpAsyncHandlersPipelining.java?rev=1593793&view=auto
==============================================================================
--- httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/TestHttpAsyncHandlersPipelining.java (added)
+++ httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/TestHttpAsyncHandlersPipelining.java Sun May 11 14:09:30 2014
@@ -0,0 +1,356 @@
+/*
+ * ====================================================================
+ * 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.http.nio.integration;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.Future;
+
+import org.apache.http.HttpEntityEnclosingRequest;
+import org.apache.http.HttpException;
+import org.apache.http.HttpHost;
+import org.apache.http.HttpRequest;
+import org.apache.http.HttpResponse;
+import org.apache.http.HttpStatus;
+import org.apache.http.HttpVersion;
+import org.apache.http.entity.ContentType;
+import org.apache.http.impl.nio.DefaultNHttpClientConnection;
+import org.apache.http.impl.nio.DefaultNHttpServerConnection;
+import org.apache.http.message.BasicHttpEntityEnclosingRequest;
+import org.apache.http.message.BasicHttpRequest;
+import org.apache.http.message.BasicHttpResponse;
+import org.apache.http.nio.NHttpConnectionFactory;
+import org.apache.http.nio.entity.NStringEntity;
+import org.apache.http.nio.protocol.BasicAsyncRequestConsumer;
+import org.apache.http.nio.protocol.BasicAsyncRequestHandler;
+import org.apache.http.nio.protocol.BasicAsyncResponseProducer;
+import org.apache.http.nio.protocol.HttpAsyncExchange;
+import org.apache.http.nio.protocol.HttpAsyncExpectationVerifier;
+import org.apache.http.nio.protocol.HttpAsyncRequestConsumer;
+import org.apache.http.nio.protocol.HttpAsyncRequestHandler;
+import org.apache.http.nio.protocol.HttpAsyncRequestHandlerMapper;
+import org.apache.http.nio.protocol.UriHttpAsyncRequestHandlerMapper;
+import org.apache.http.nio.reactor.IOReactorStatus;
+import org.apache.http.nio.reactor.ListenerEndpoint;
+import org.apache.http.nio.testserver.HttpCoreNIOTestBase;
+import org.apache.http.nio.testserver.LoggingClientConnectionFactory;
+import org.apache.http.nio.testserver.LoggingServerConnectionFactory;
+import org.apache.http.protocol.HttpContext;
+import org.apache.http.protocol.HttpProcessor;
+import org.apache.http.protocol.ImmutableHttpProcessor;
+import org.apache.http.protocol.RequestConnControl;
+import org.apache.http.protocol.RequestContent;
+import org.apache.http.protocol.RequestTargetHost;
+import org.apache.http.protocol.RequestUserAgent;
+import org.apache.http.util.EntityUtils;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * HttpCore NIO integration tests for pipelined request processing.
+ */
+public class TestHttpAsyncHandlersPipelining extends HttpCoreNIOTestBase {
+
+ public static final HttpProcessor DEFAULT_HTTP_PROC = new ImmutableHttpProcessor(
+ new RequestContent(),
+ new RequestTargetHost(),
+ new RequestConnControl(),
+ new RequestUserAgent("TEST-CLIENT/1.1"));
+
+ @Before
+ public void setUp() throws Exception {
+ initServer();
+ initClient();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ shutDownClient();
+ shutDownServer();
+ }
+
+ @Override
+ protected NHttpConnectionFactory<DefaultNHttpServerConnection> createServerConnectionFactory() throws Exception {
+ return new LoggingServerConnectionFactory();
+ }
+
+ @Override
+ protected NHttpConnectionFactory<DefaultNHttpClientConnection> createClientConnectionFactory() throws Exception {
+ return new LoggingClientConnectionFactory();
+ }
+
+ private InetSocketAddress start(
+ final HttpProcessor clientProtocolProcessor,
+ final HttpProcessor serverProtocolProcessor,
+ final HttpAsyncRequestHandlerMapper requestHandlerResolver,
+ final HttpAsyncExpectationVerifier expectationVerifier) throws Exception {
+ this.server.start(serverProtocolProcessor, requestHandlerResolver, expectationVerifier);
+ this.client.start(clientProtocolProcessor);
+
+ final ListenerEndpoint endpoint = this.server.getListenerEndpoint();
+ endpoint.waitFor();
+
+ Assert.assertEquals("Test server status", IOReactorStatus.ACTIVE, this.server.getStatus());
+ return (InetSocketAddress) endpoint.getAddress();
+ }
+
+ private InetSocketAddress start(
+ final HttpAsyncRequestHandlerMapper requestHandlerResolver,
+ final HttpAsyncExpectationVerifier expectationVerifier) throws Exception {
+ return start(DEFAULT_HTTP_PROC, null, requestHandlerResolver, expectationVerifier);
+ }
+
+ private static String createRequestUri(final String pattern, final int count) {
+ return pattern + "x" + count;
+ }
+
+ private static String createExpectedString(final String pattern, final int count) {
+ final StringBuilder buffer = new StringBuilder();
+ for (int i = 0; i < count; i++) {
+ buffer.append(pattern);
+ }
+ return buffer.toString();
+ }
+
+ @Test
+ public void testHttpGets() throws Exception {
+ final UriHttpAsyncRequestHandlerMapper registry = new UriHttpAsyncRequestHandlerMapper();
+ registry.register("*", new BasicAsyncRequestHandler(new SimpleRequestHandler()));
+ final InetSocketAddress address = start(registry, null);
+
+ this.client.setMaxPerRoute(3);
+ this.client.setMaxTotal(3);
+
+ final String pattern = RndTestPatternGenerator.generateText();
+ final int count = RndTestPatternGenerator.generateCount(1000);
+
+ final HttpHost target = new HttpHost("localhost", address.getPort());
+ final String expectedPattern = createExpectedString(pattern, count);
+
+ final Queue<Future<List<HttpResponse>>> queue = new ConcurrentLinkedQueue<Future<List<HttpResponse>>>();
+ for (int i = 0; i < 10; i++) {
+ final String requestUri = createRequestUri(pattern, count);
+ final Future<List<HttpResponse>> future = this.client.executePipelined(target,
+ new BasicHttpRequest("GET", requestUri),
+ new BasicHttpRequest("GET", requestUri),
+ new BasicHttpRequest("GET", requestUri));
+ queue.add(future);
+ }
+
+ while (!queue.isEmpty()) {
+ final Future<List<HttpResponse>> future = queue.remove();
+ final List<HttpResponse> responses = future.get();
+ Assert.assertNotNull(responses);
+ Assert.assertEquals(3, responses.size());
+ for (HttpResponse response: responses) {
+ Assert.assertEquals(HttpStatus.SC_OK, response.getStatusLine().getStatusCode());
+ Assert.assertEquals(expectedPattern, EntityUtils.toString(response.getEntity()));
+ }
+ }
+ }
+
+ @Test
+ public void testHttpHeads() throws Exception {
+ final UriHttpAsyncRequestHandlerMapper registry = new UriHttpAsyncRequestHandlerMapper();
+ registry.register("*", new BasicAsyncRequestHandler(new SimpleRequestHandler()));
+ final InetSocketAddress address = start(registry, null);
+
+ this.client.setMaxPerRoute(3);
+ this.client.setMaxTotal(3);
+
+ final String pattern = RndTestPatternGenerator.generateText();
+ final int count = RndTestPatternGenerator.generateCount(1000);
+
+ final HttpHost target = new HttpHost("localhost", address.getPort());
+
+ final Queue<Future<List<HttpResponse>>> queue = new ConcurrentLinkedQueue<Future<List<HttpResponse>>>();
+ for (int i = 0; i < 10; i++) {
+ final String requestUri = createRequestUri(pattern, count);
+ final Future<List<HttpResponse>> future = this.client.executePipelined(target,
+ new BasicHttpRequest("GET", requestUri),
+ new BasicHttpRequest("GET", requestUri),
+ new BasicHttpRequest("GET", requestUri));
+ queue.add(future);
+ }
+
+ while (!queue.isEmpty()) {
+ final Future<List<HttpResponse>> future = queue.remove();
+ final List<HttpResponse> responses = future.get();
+ Assert.assertNotNull(responses);
+ Assert.assertEquals(3, responses.size());
+ for (HttpResponse response: responses) {
+ Assert.assertEquals(HttpStatus.SC_OK, response.getStatusLine().getStatusCode());
+ }
+ }
+ }
+
+ @Test
+ public void testHttpPosts() throws Exception {
+ final UriHttpAsyncRequestHandlerMapper registry = new UriHttpAsyncRequestHandlerMapper();
+ registry.register("*", new BasicAsyncRequestHandler(new SimpleRequestHandler()));
+ final InetSocketAddress address = start(registry, null);
+
+ this.client.setMaxPerRoute(3);
+ this.client.setMaxTotal(3);
+
+ final String pattern = RndTestPatternGenerator.generateText();
+ final int count = RndTestPatternGenerator.generateCount(1000);
+
+ final HttpHost target = new HttpHost("localhost", address.getPort());
+ final String expectedPattern = createExpectedString(pattern, count);
+
+ final Queue<Future<List<HttpResponse>>> queue = new ConcurrentLinkedQueue<Future<List<HttpResponse>>>();
+ for (int i = 0; i < 10; i++) {
+ final String requestUri = createRequestUri(pattern, count);
+ final HttpEntityEnclosingRequest request1 = new BasicHttpEntityEnclosingRequest("POST", requestUri);
+ final NStringEntity entity1 = new NStringEntity(expectedPattern, ContentType.DEFAULT_TEXT);
+ entity1.setChunked(RndTestPatternGenerator.generateBoolean());
+ request1.setEntity(entity1);
+ final HttpEntityEnclosingRequest request2 = new BasicHttpEntityEnclosingRequest("POST", requestUri);
+ final NStringEntity entity2 = new NStringEntity(expectedPattern, ContentType.DEFAULT_TEXT);
+ entity2.setChunked(RndTestPatternGenerator.generateBoolean());
+ request2.setEntity(entity2);
+ final HttpEntityEnclosingRequest request3 = new BasicHttpEntityEnclosingRequest("POST", requestUri);
+ final NStringEntity entity3 = new NStringEntity(expectedPattern, ContentType.DEFAULT_TEXT);
+ entity3.setChunked(RndTestPatternGenerator.generateBoolean());
+ request3.setEntity(entity3);
+ final Future<List<HttpResponse>> future = this.client.executePipelined(target,
+ request1, request2, request3);
+ queue.add(future);
+ }
+
+ while (!queue.isEmpty()) {
+ final Future<List<HttpResponse>> future = queue.remove();
+ final List<HttpResponse> responses = future.get();
+ Assert.assertNotNull(responses);
+ Assert.assertEquals(3, responses.size());
+ for (HttpResponse response: responses) {
+ Assert.assertEquals(HttpStatus.SC_OK, response.getStatusLine().getStatusCode());
+ Assert.assertEquals(expectedPattern, EntityUtils.toString(response.getEntity()));
+ }
+ }
+ }
+
+ @Test
+ public void testHttpDelayedResponse() throws Exception {
+
+ class DelayedRequestHandler implements HttpAsyncRequestHandler<HttpRequest> {
+
+ private final SimpleRequestHandler requestHandler;
+
+ public DelayedRequestHandler() {
+ super();
+ this.requestHandler = new SimpleRequestHandler();
+ }
+
+ @Override
+ public HttpAsyncRequestConsumer<HttpRequest> processRequest(
+ final HttpRequest request,
+ final HttpContext context) {
+ return new BasicAsyncRequestConsumer();
+ }
+
+ @Override
+ public void handle(
+ final HttpRequest request,
+ final HttpAsyncExchange httpexchange,
+ final HttpContext context) throws HttpException, IOException {
+ final BasicHttpResponse response = new BasicHttpResponse(HttpVersion.HTTP_1_1, HttpStatus.SC_OK, "OK");
+ new Thread() {
+ @Override
+ public void run() {
+ // Wait a bit, to make sure this is delayed.
+ try { Thread.sleep(100); } catch(final InterruptedException ie) {}
+ // Set the entity after delaying...
+ try {
+ requestHandler.handle(request, response, context);
+ } catch (final Exception ex) {
+ response.setStatusCode(HttpStatus.SC_INTERNAL_SERVER_ERROR);
+ }
+ httpexchange.submitResponse(new BasicAsyncResponseProducer(response));
+ }
+ }.start();
+ }
+
+ }
+
+ final UriHttpAsyncRequestHandlerMapper registry = new UriHttpAsyncRequestHandlerMapper();
+ registry.register("*", new DelayedRequestHandler());
+ final InetSocketAddress address = start(registry, null);
+
+ this.client.setMaxPerRoute(3);
+ this.client.setMaxTotal(3);
+
+ final String pattern1 = RndTestPatternGenerator.generateText();
+ final String pattern2 = RndTestPatternGenerator.generateText();
+ final String pattern3 = RndTestPatternGenerator.generateText();
+ final int count = RndTestPatternGenerator.generateCount(1000);
+
+ final HttpHost target = new HttpHost("localhost", address.getPort());
+ final String expectedPattern1 = createExpectedString(pattern1, count);
+ final String expectedPattern2 = createExpectedString(pattern2, count);
+ final String expectedPattern3 = createExpectedString(pattern3, count);
+
+ final Queue<Future<List<HttpResponse>>> queue = new ConcurrentLinkedQueue<Future<List<HttpResponse>>>();
+ for (int i = 0; i < 1; i++) {
+ final HttpRequest request1 = new BasicHttpRequest("GET", createRequestUri(pattern1, count));
+ final HttpEntityEnclosingRequest request2 = new BasicHttpEntityEnclosingRequest("POST",
+ createRequestUri(pattern2, count));
+ final NStringEntity entity2 = new NStringEntity(expectedPattern2, ContentType.DEFAULT_TEXT);
+ entity2.setChunked(RndTestPatternGenerator.generateBoolean());
+ request2.setEntity(entity2);
+ final HttpEntityEnclosingRequest request3 = new BasicHttpEntityEnclosingRequest("POST",
+ createRequestUri(pattern3, count));
+ final NStringEntity entity3 = new NStringEntity(expectedPattern3, ContentType.DEFAULT_TEXT);
+ entity3.setChunked(RndTestPatternGenerator.generateBoolean());
+ request3.setEntity(entity3);
+ final Future<List<HttpResponse>> future = this.client.executePipelined(target,
+ request1, request2, request3);
+ queue.add(future);
+ }
+
+ while (!queue.isEmpty()) {
+ final Future<List<HttpResponse>> future = queue.remove();
+ final List<HttpResponse> responses = future.get();
+ Assert.assertNotNull(responses);
+ Assert.assertEquals(3, responses.size());
+ for (HttpResponse response: responses) {
+ Assert.assertEquals(HttpStatus.SC_OK, response.getStatusLine().getStatusCode());
+ }
+ Assert.assertEquals(expectedPattern1, EntityUtils.toString(responses.get(0).getEntity()));
+ Assert.assertEquals(expectedPattern2, EntityUtils.toString(responses.get(1).getEntity()));
+ Assert.assertEquals(expectedPattern3, EntityUtils.toString(responses.get(2).getEntity()));
+ }
+ }
+
+}
Propchange: httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/TestHttpAsyncHandlersPipelining.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/TestHttpAsyncHandlersPipelining.java
------------------------------------------------------------------------------
svn:keywords = Date Revision
Propchange: httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/TestHttpAsyncHandlersPipelining.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Copied: httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/TestHttpsAsyncHandlersPipelining.java (from r1590623, httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/RndTestPatternGenerator.java)
URL: http://svn.apache.org/viewvc/httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/TestHttpsAsyncHandlersPipelining.java?p2=httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/TestHttpsAsyncHandlersPipelining.java&p1=httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/RndTestPatternGenerator.java&r1=1590623&r2=1593793&rev=1593793&view=diff
==============================================================================
--- httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/RndTestPatternGenerator.java (original)
+++ httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/TestHttpsAsyncHandlersPipelining.java Sun May 11 14:09:30 2014
@@ -27,24 +27,26 @@
package org.apache.http.nio.integration;
-import java.util.Random;
-
-final class RndTestPatternGenerator {
-
- private static final Random RND = new Random();
- private static final String TEST_CHARS = "0123456789ABCDEF";
-
- public static String generateText() {
- final StringBuilder buffer = new StringBuilder();
- for (int i = 0; i < 5; i++) {
- final char rndchar = TEST_CHARS.charAt(RND.nextInt(TEST_CHARS.length() - 1));
- buffer.append(rndchar);
- }
- return buffer.toString();
+import org.apache.http.impl.nio.DefaultNHttpClientConnection;
+import org.apache.http.impl.nio.DefaultNHttpServerConnection;
+import org.apache.http.nio.NHttpConnectionFactory;
+import org.apache.http.nio.testserver.LoggingSSLClientConnectionFactory;
+import org.apache.http.nio.testserver.LoggingSSLServerConnectionFactory;
+import org.apache.http.nio.testserver.SSLTestContexts;
+
+/**
+ * HttpCore NIO integration tests for pipelined request processing using SSL.
+ */
+public class TestHttpsAsyncHandlersPipelining extends TestHttpAsyncHandlersPipelining {
+
+ @Override
+ protected NHttpConnectionFactory<DefaultNHttpServerConnection> createServerConnectionFactory() throws Exception {
+ return new LoggingSSLServerConnectionFactory(SSLTestContexts.createServerSSLContext());
}
- public static int generateCount(final int max) {
- return RND.nextInt(max - 1) + 1;
+ @Override
+ protected NHttpConnectionFactory<DefaultNHttpClientConnection> createClientConnectionFactory() throws Exception {
+ return new LoggingSSLClientConnectionFactory(SSLTestContexts.createClientSSLContext());
}
}
Copied: httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/TestServerSidePipelining.java (from r1590623, httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/TestPipelining.java)
URL: http://svn.apache.org/viewvc/httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/TestServerSidePipelining.java?p2=httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/TestServerSidePipelining.java&p1=httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/TestPipelining.java&r1=1590623&r2=1593793&rev=1593793&view=diff
==============================================================================
--- httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/TestPipelining.java (original)
+++ httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/TestServerSidePipelining.java Sun May 11 14:09:30 2014
@@ -73,7 +73,7 @@ import org.junit.Test;
/**
* Tests for handling pipelined requests.
*/
-public class TestPipelining extends HttpCoreNIOTestBase {
+public class TestServerSidePipelining extends HttpCoreNIOTestBase {
protected HttpProcessor serverHttpProc;
protected HttpAsyncService serviceHandler;
Propchange: httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/TestServerSidePipelining.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/TestServerSidePipelining.java
------------------------------------------------------------------------------
svn:keywords = Date Revision
Propchange: httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/integration/TestServerSidePipelining.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Modified: httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/protocol/TestHttpAsyncRequestExecutor.java
URL: http://svn.apache.org/viewvc/httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/protocol/TestHttpAsyncRequestExecutor.java?rev=1593793&r1=1593792&r2=1593793&view=diff
==============================================================================
--- httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/protocol/TestHttpAsyncRequestExecutor.java (original)
+++ httpcomponents/httpcore/trunk/httpcore-nio/src/test/java/org/apache/http/nio/protocol/TestHttpAsyncRequestExecutor.java Sun May 11 14:09:30 2014
@@ -238,6 +238,7 @@ public class TestHttpAsyncRequestExecuto
@Test
public void testRequestContentOutput() throws Exception {
final State state = new HttpAsyncRequestExecutor.State();
+ state.setRequestState(MessageState.BODY_STREAM);
this.connContext.setAttribute(HttpAsyncRequestExecutor.HTTP_EXCHANGE_STATE, state);
this.connContext.setAttribute(HttpAsyncRequestExecutor.HTTP_HANDLER, this.exchangeHandler);
Mockito.when(this.encoder.isCompleted()).thenReturn(Boolean.FALSE);
@@ -251,6 +252,7 @@ public class TestHttpAsyncRequestExecuto
@Test
public void testRequestContentOutputCompleted() throws Exception {
final State state = new HttpAsyncRequestExecutor.State();
+ state.setRequestState(MessageState.BODY_STREAM);
this.connContext.setAttribute(HttpAsyncRequestExecutor.HTTP_EXCHANGE_STATE, state);
this.connContext.setAttribute(HttpAsyncRequestExecutor.HTTP_HANDLER, this.exchangeHandler);
Mockito.when(this.encoder.isCompleted()).thenReturn(Boolean.TRUE);
@@ -308,7 +310,7 @@ public class TestHttpAsyncRequestExecuto
this.protocolHandler.responseReceived(this.conn);
Assert.assertNull(state.getResponse());
- Assert.assertEquals(MessageState.ACK, state.getRequestState());
+ Assert.assertEquals(MessageState.BODY_STREAM, state.getRequestState());
Assert.assertEquals(MessageState.READY, state.getResponseState());
Mockito.verify(this.conn).setSocketTimeout(1000);
Mockito.verify(this.conn).requestOutput();
@@ -463,6 +465,7 @@ public class TestHttpAsyncRequestExecuto
@Test
public void testResponseContentInput() throws Exception {
final State state = new HttpAsyncRequestExecutor.State();
+ state.setResponseState(MessageState.BODY_STREAM);
this.connContext.setAttribute(HttpAsyncRequestExecutor.HTTP_EXCHANGE_STATE, state);
this.connContext.setAttribute(HttpAsyncRequestExecutor.HTTP_HANDLER, this.exchangeHandler);
Mockito.when(this.decoder.isCompleted()).thenReturn(Boolean.FALSE);
@@ -478,6 +481,7 @@ public class TestHttpAsyncRequestExecuto
final State state = new HttpAsyncRequestExecutor.State();
final HttpRequest request = new BasicHttpRequest("GET", "/");
state.setRequest(request);
+ state.setResponseState(MessageState.BODY_STREAM);
final BasicHttpResponse response = new BasicHttpResponse(HttpVersion.HTTP_1_1, 200, "OK");
state.setResponse(response);
Mockito.when(this.exchangeHandler.isDone()).thenReturn(Boolean.TRUE);
@@ -497,6 +501,7 @@ public class TestHttpAsyncRequestExecuto
@Test
public void testResponseContentOutputCompletedHandlerNotDone() throws Exception {
final State state = new HttpAsyncRequestExecutor.State();
+ state.setResponseState(MessageState.BODY_STREAM);
final HttpRequest request = new BasicHttpRequest("GET", "/");
state.setRequest(request);
final BasicHttpResponse response = new BasicHttpResponse(HttpVersion.HTTP_1_1, 200, "OK");
@@ -521,6 +526,7 @@ public class TestHttpAsyncRequestExecuto
final State state = new HttpAsyncRequestExecutor.State();
final HttpRequest request = new BasicHttpRequest("GET", "/");
state.setRequest(request);
+ state.setResponseState(MessageState.BODY_STREAM);
final BasicHttpResponse response = new BasicHttpResponse(HttpVersion.HTTP_1_1, 200, "OK");
state.setResponse(response);
Mockito.when(this.exchangeHandler.isDone()).thenReturn(Boolean.FALSE);
@@ -543,6 +549,7 @@ public class TestHttpAsyncRequestExecuto
final State state = new HttpAsyncRequestExecutor.State();
final HttpRequest request = new BasicHttpRequest("GET", "/");
state.setRequest(request);
+ state.setResponseState(MessageState.BODY_STREAM);
final BasicHttpResponse response = new BasicHttpResponse(HttpVersion.HTTP_1_1, 200, "OK");
state.setResponse(response);
state.invalidate();