diff options
14 files changed, 284 insertions, 94 deletions
diff --git a/jetty-client/src/main/java/org/eclipse/jetty/client/AuthenticationProtocolHandler.java b/jetty-client/src/main/java/org/eclipse/jetty/client/AuthenticationProtocolHandler.java index 007d0e2b46..ccb15cf642 100644 --- a/jetty-client/src/main/java/org/eclipse/jetty/client/AuthenticationProtocolHandler.java +++ b/jetty-client/src/main/java/org/eclipse/jetty/client/AuthenticationProtocolHandler.java @@ -80,14 +80,12 @@ public class AuthenticationProtocolHandler implements ProtocolHandler public void onComplete(Result result) { Request request = result.getRequest(); - HttpConversation conversation = client.getConversation(request.getConversationID(), false); - List<Response.ResponseListener> listeners = conversation.getExchanges().peekFirst().getResponseListeners(); ContentResponse response = new HttpContentResponse(result.getResponse(), getContent(), getEncoding()); if (result.isFailed()) { Throwable failure = result.getFailure(); LOG.debug("Authentication challenge failed {}", failure); - notifier.forwardFailureComplete(listeners, request, result.getRequestFailure(), response, result.getResponseFailure()); + forwardFailureComplete(request, result.getRequestFailure(), response, result.getResponseFailure()); return; } @@ -95,7 +93,7 @@ public class AuthenticationProtocolHandler implements ProtocolHandler if (wwwAuthenticates.isEmpty()) { LOG.debug("Authentication challenge without WWW-Authenticate header"); - notifier.forwardFailureComplete(listeners, request, null, response, new HttpResponseException("HTTP protocol violation: 401 without WWW-Authenticate header", response)); + forwardFailureComplete(request, null, response, new HttpResponseException("HTTP protocol violation: 401 without WWW-Authenticate header", response)); return; } @@ -114,15 +112,16 @@ public class AuthenticationProtocolHandler implements ProtocolHandler if (authentication == null) { LOG.debug("No authentication available for {}", request); - notifier.forwardSuccessComplete(listeners, request, response); + forwardSuccessComplete(request, response); return; } + HttpConversation conversation = client.getConversation(request.getConversationID(), false); final Authentication.Result authnResult = authentication.authenticate(request, response, wwwAuthenticate.value, conversation); LOG.debug("Authentication result {}", authnResult); if (authnResult == null) { - notifier.forwardSuccessComplete(listeners, request, response); + forwardSuccessComplete(request, response); return; } @@ -138,6 +137,20 @@ public class AuthenticationProtocolHandler implements ProtocolHandler }).send(null); } + private void forwardSuccessComplete(Request request, Response response) + { + HttpConversation conversation = client.getConversation(request.getConversationID(), false); + conversation.updateResponseListeners(null); + notifier.forwardSuccessComplete(conversation.getResponseListeners(), request, response); + } + + private void forwardFailureComplete(Request request, Throwable requestFailure, Response response, Throwable responseFailure) + { + HttpConversation conversation = client.getConversation(request.getConversationID(), false); + conversation.updateResponseListeners(null); + notifier.forwardFailureComplete(conversation.getResponseListeners(), request, requestFailure, response, responseFailure); + } + private List<WWWAuthenticate> parseWWWAuthenticate(Response response) { // TODO: these should be ordered by strength diff --git a/jetty-client/src/main/java/org/eclipse/jetty/client/ContinueProtocolHandler.java b/jetty-client/src/main/java/org/eclipse/jetty/client/ContinueProtocolHandler.java index e7995c1958..3b17b5229a 100644 --- a/jetty-client/src/main/java/org/eclipse/jetty/client/ContinueProtocolHandler.java +++ b/jetty-client/src/main/java/org/eclipse/jetty/client/ContinueProtocolHandler.java @@ -68,15 +68,16 @@ public class ContinueProtocolHandler implements ProtocolHandler // Mark the 100 Continue response as handled conversation.setAttribute(ATTRIBUTE, Boolean.TRUE); + // Reset the conversation listeners, since we are going to receive another response code + conversation.updateResponseListeners(null); + HttpExchange exchange = conversation.getExchanges().peekLast(); assert exchange.getResponse() == response; - List<Response.ResponseListener> listeners = exchange.getResponseListeners(); switch (response.getStatus()) { case 100: { // All good, continue - conversation.setResponseListener(null); exchange.resetResponse(true); exchange.proceed(true); break; @@ -86,7 +87,7 @@ public class ContinueProtocolHandler implements ProtocolHandler // Server either does not support 100 Continue, // or it does and wants to refuse the request content, // or we got some other HTTP status code like a redirect. - conversation.setResponseListener(null); + List<Response.ResponseListener> listeners = exchange.getResponseListeners(); HttpContentResponse contentResponse = new HttpContentResponse(response, getContent(), getEncoding()); notifier.forwardSuccess(listeners, contentResponse); exchange.proceed(false); @@ -101,6 +102,8 @@ public class ContinueProtocolHandler implements ProtocolHandler HttpConversation conversation = client.getConversation(response.getConversationID(), false); // Mark the 100 Continue response as handled conversation.setAttribute(ATTRIBUTE, Boolean.TRUE); + // Reset the conversation listeners to allow the conversation to be completed + conversation.updateResponseListeners(null); HttpExchange exchange = conversation.getExchanges().peekLast(); assert exchange.getResponse() == response; diff --git a/jetty-client/src/main/java/org/eclipse/jetty/client/HttpConversation.java b/jetty-client/src/main/java/org/eclipse/jetty/client/HttpConversation.java index cca98c54c7..6ab9a3a50d 100644 --- a/jetty-client/src/main/java/org/eclipse/jetty/client/HttpConversation.java +++ b/jetty-client/src/main/java/org/eclipse/jetty/client/HttpConversation.java @@ -19,7 +19,6 @@ package org.eclipse.jetty.client; import java.util.ArrayList; -import java.util.Arrays; import java.util.Deque; import java.util.List; import java.util.concurrent.ConcurrentLinkedDeque; @@ -32,7 +31,8 @@ public class HttpConversation extends AttributesMap private final Deque<HttpExchange> exchanges = new ConcurrentLinkedDeque<>(); private final HttpClient client; private final long id; - private volatile Response.ResponseListener listener; + private volatile boolean complete; + private volatile List<Response.ResponseListener> listeners; public HttpConversation(HttpClient client, long id) { @@ -55,42 +55,42 @@ public class HttpConversation extends AttributesMap * This list changes as the conversation proceeds, as follows: * <ol> * <li> - * request R1 send => conversation.setResponseListener(null) + * request R1 send => conversation.updateResponseListeners(null) * <ul> * <li>exchanges in conversation: E1</li> * <li>listeners to be notified: E1.listeners</li> * </ul> * </li> * <li> - * response R1 arrived, 401 => conversation.setResponseListener(AuthenticationProtocolHandler.listener) + * response R1 arrived, 401 => conversation.updateResponseListeners(AuthenticationProtocolHandler.listener) * <ul> * <li>exchanges in conversation: E1</li> * <li>listeners to be notified: AuthenticationProtocolHandler.listener</li> * </ul> * </li> * <li> - * request R2 send => conversation.setResponseListener(null) + * request R2 send => conversation.updateResponseListeners(null) * <ul> * <li>exchanges in conversation: E1 + E2</li> * <li>listeners to be notified: E2.listeners + E1.listeners</li> * </ul> * </li> * <li> - * response R2 arrived, 302 => conversation.setResponseListener(RedirectProtocolHandler.listener) + * response R2 arrived, 302 => conversation.updateResponseListeners(RedirectProtocolHandler.listener) * <ul> * <li>exchanges in conversation: E1 + E2</li> * <li>listeners to be notified: E2.listeners + RedirectProtocolHandler.listener</li> * </ul> * </li> * <li> - * request R3 send => conversation.setResponseListener(null) + * request R3 send => conversation.updateResponseListeners(null) * <ul> * <li>exchanges in conversation: E1 + E2 + E3</li> * <li>listeners to be notified: E3.listeners + E1.listeners</li> * </ul> * </li> * <li> - * response R3 arrived, 200 => conversation.setResponseListener(null) + * response R3 arrived, 200 => conversation.updateResponseListeners(null) * <ul> * <li>exchanges in conversation: E1 + E2 + E3</li> * <li>listeners to be notified: E3.listeners + E1.listeners</li> @@ -110,45 +110,52 @@ public class HttpConversation extends AttributesMap */ public List<Response.ResponseListener> getResponseListeners() { + return listeners; + } + + /** + * Requests to update the response listener, eventually using the given override response listener, + * that must be notified instead of the first exchange response listeners. + * This works in conjunction with {@link #getResponseListeners()}, returning the appropriate response + * listeners that needs to be notified of response events. + * + * @param overrideListener the override response listener + */ + public void updateResponseListeners(Response.ResponseListener overrideListener) + { + // If we have no override listener, then the + // conversation may be completed at a later time + complete = overrideListener == null; + + // Create a new instance to avoid that iterating over the listeners + // will notify a listener that may send a new request and trigger + // another call to this method which will build different listeners + // which may be iterated over when the iteration continues. + listeners = new ArrayList<>(); + HttpExchange firstExchange = exchanges.peekFirst(); HttpExchange lastExchange = exchanges.peekLast(); if (firstExchange == lastExchange) { - if (listener != null) - return Arrays.asList(listener); + if (overrideListener != null) + listeners.add(overrideListener); else - return firstExchange.getResponseListeners(); + listeners.addAll(firstExchange.getResponseListeners()); } else { // Order is important, we want to notify the last exchange first - List<Response.ResponseListener> result = new ArrayList<>(lastExchange.getResponseListeners()); - if (listener != null) - result.add(listener); + listeners.addAll(lastExchange.getResponseListeners()); + if (overrideListener != null) + listeners.add(overrideListener); else - result.addAll(firstExchange.getResponseListeners()); - return result; + listeners.addAll(firstExchange.getResponseListeners()); } } - /** - * Sets an override response listener that must be notified instead of the first exchange response listeners. - * This works in conjunction with {@link #getResponseListeners()}, returning the appropriate response - * listeners that needs to be notified of response events. - * - * @param listener the override response listener - */ - public void setResponseListener(Response.ResponseListener listener) - { - this.listener = listener; - } - public void complete() { - // The conversation is really terminated only - // when there is no conversation listener that - // may have continued the conversation. - if (listener == null) + if (complete) client.removeConversation(this); } diff --git a/jetty-client/src/main/java/org/eclipse/jetty/client/HttpExchange.java b/jetty-client/src/main/java/org/eclipse/jetty/client/HttpExchange.java index 56bc5bc2a9..615542141e 100644 --- a/jetty-client/src/main/java/org/eclipse/jetty/client/HttpExchange.java +++ b/jetty-client/src/main/java/org/eclipse/jetty/client/HttpExchange.java @@ -52,7 +52,7 @@ public class HttpExchange this.listeners = listeners; this.response = new HttpResponse(request, listeners); conversation.getExchanges().offer(this); - conversation.setResponseListener(null); + conversation.updateResponseListeners(null); } public HttpConversation getConversation() diff --git a/jetty-client/src/main/java/org/eclipse/jetty/client/HttpReceiver.java b/jetty-client/src/main/java/org/eclipse/jetty/client/HttpReceiver.java index 8f4e593cae..1af0caf143 100644 --- a/jetty-client/src/main/java/org/eclipse/jetty/client/HttpReceiver.java +++ b/jetty-client/src/main/java/org/eclipse/jetty/client/HttpReceiver.java @@ -145,8 +145,13 @@ public class HttpReceiver implements HttpParser.ResponseHandler<ByteBuffer> // Probe the protocol handlers HttpClient client = connection.getHttpClient(); ProtocolHandler protocolHandler = client.findProtocolHandler(exchange.getRequest(), response); - Response.Listener handlerListener = protocolHandler == null ? null : protocolHandler.getResponseListener(); - exchange.getConversation().setResponseListener(handlerListener); + Response.Listener handlerListener = null; + if (protocolHandler != null) + { + handlerListener = protocolHandler.getResponseListener(); + LOG.debug("Found protocol handler {}", protocolHandler); + } + exchange.getConversation().updateResponseListeners(handlerListener); LOG.debug("Receiving {}", response); ResponseNotifier notifier = connection.getDestination().getResponseNotifier(); diff --git a/jetty-client/src/main/java/org/eclipse/jetty/client/HttpRequest.java b/jetty-client/src/main/java/org/eclipse/jetty/client/HttpRequest.java index ae27d407ac..a6646be194 100644 --- a/jetty-client/src/main/java/org/eclipse/jetty/client/HttpRequest.java +++ b/jetty-client/src/main/java/org/eclipse/jetty/client/HttpRequest.java @@ -293,6 +293,13 @@ public class HttpRequest implements Request } @Override + public Request onRequestContent(ContentListener listener) + { + this.requestListeners.add(listener); + return this; + } + + @Override public Request onRequestSuccess(SuccessListener listener) { this.requestListeners.add(listener); diff --git a/jetty-client/src/main/java/org/eclipse/jetty/client/HttpSender.java b/jetty-client/src/main/java/org/eclipse/jetty/client/HttpSender.java index 0c0e064355..ba579a5b68 100644 --- a/jetty-client/src/main/java/org/eclipse/jetty/client/HttpSender.java +++ b/jetty-client/src/main/java/org/eclipse/jetty/client/HttpSender.java @@ -162,13 +162,16 @@ public class HttpSender implements AsyncContentProvider.Listener while (true) { - HttpGenerator.Result result = generator.generateRequest(requestInfo, header, chunk, contentChunk.content, contentChunk.lastContent); + ByteBuffer content = contentChunk.content; + final ByteBuffer contentBuffer = content == null ? null : content.slice(); + + HttpGenerator.Result result = generator.generateRequest(requestInfo, header, chunk, content, contentChunk.lastContent); switch (result) { case NEED_INFO: { - ContentProvider content = request.getContent(); - long contentLength = content == null ? -1 : content.getLength(); + ContentProvider requestContent = request.getContent(); + long contentLength = requestContent == null ? -1 : requestContent.getLength(); requestInfo = new HttpGenerator.RequestInfo(request.getVersion(), request.getHeaders(), contentLength, request.getMethod().asString(), request.getPath()); break; } @@ -224,15 +227,8 @@ public class HttpSender implements AsyncContentProvider.Listener { LOG.debug("Write succeeded for {}", request); - if (!commit(request)) - return; - - if (expecting100ContinueResponse) - { - LOG.debug("Expecting 100 Continue for {}", request); - continueContentChunk.signal(); + if (!processWrite(request, contentBuffer, expecting100ContinueResponse)) return; - } send(); } @@ -250,7 +246,7 @@ public class HttpSender implements AsyncContentProvider.Listener continueContentChunk = new ContinueContentChunk(contentChunk); } - write(callback, header, chunk, expecting100ContinueResponse ? null : contentChunk.content); + write(callback, header, chunk, expecting100ContinueResponse ? null : content); if (callback.process()) { @@ -260,15 +256,8 @@ public class HttpSender implements AsyncContentProvider.Listener if (callback.isSucceeded()) { - if (!commit(request)) - return; - - if (expecting100ContinueResponse) - { - LOG.debug("Expecting 100 Continue for {}", request); - continueContentChunk.signal(); + if (!processWrite(request, contentBuffer, expecting100ContinueResponse)) return; - } // Send further content contentChunk = new ContentChunk(contentIterator); @@ -363,6 +352,27 @@ public class HttpSender implements AsyncContentProvider.Listener } } + private boolean processWrite(Request request, ByteBuffer content, boolean expecting100ContinueResponse) + { + if (!commit(request)) + return false; + + if (content != null) + { + RequestNotifier notifier = connection.getDestination().getRequestNotifier(); + notifier.notifyContent(request, content); + } + + if (expecting100ContinueResponse) + { + LOG.debug("Expecting 100 Continue for {}", request); + continueContentChunk.signal(); + return false; + } + + return true; + } + public void proceed(boolean proceed) { ContinueContentChunk contentChunk = continueContentChunk; diff --git a/jetty-client/src/main/java/org/eclipse/jetty/client/RedirectProtocolHandler.java b/jetty-client/src/main/java/org/eclipse/jetty/client/RedirectProtocolHandler.java index 0d5fa02840..c158b29481 100644 --- a/jetty-client/src/main/java/org/eclipse/jetty/client/RedirectProtocolHandler.java +++ b/jetty-client/src/main/java/org/eclipse/jetty/client/RedirectProtocolHandler.java @@ -212,7 +212,8 @@ public class RedirectProtocolHandler extends Response.Listener.Empty implements Request request = result.getRequest(); Response response = result.getResponse(); HttpConversation conversation = client.getConversation(request.getConversationID(), false); - List<Response.ResponseListener> listeners = conversation.getExchanges().peekFirst().getResponseListeners(); + conversation.updateResponseListeners(null); + List<Response.ResponseListener> listeners = conversation.getResponseListeners(); notifier.notifyFailure(listeners, response, failure); notifier.notifyComplete(listeners, new Result(request, response, failure)); } diff --git a/jetty-client/src/main/java/org/eclipse/jetty/client/RequestNotifier.java b/jetty-client/src/main/java/org/eclipse/jetty/client/RequestNotifier.java index 4e32e54a1f..5c0c47945e 100644 --- a/jetty-client/src/main/java/org/eclipse/jetty/client/RequestNotifier.java +++ b/jetty-client/src/main/java/org/eclipse/jetty/client/RequestNotifier.java @@ -18,6 +18,7 @@ package org.eclipse.jetty.client; +import java.nio.ByteBuffer; import java.util.List; import org.eclipse.jetty.client.api.Request; @@ -155,6 +156,36 @@ public class RequestNotifier } } + public void notifyContent(Request request, ByteBuffer content) + { + // Optimized to avoid allocations of iterator instances + List<Request.RequestListener> requestListeners = request.getRequestListeners(null); + for (int i = 0; i < requestListeners.size(); ++i) + { + Request.RequestListener listener = requestListeners.get(i); + if (listener instanceof Request.ContentListener) + notifyContent((Request.ContentListener)listener, request, content); + } + List<Request.Listener> listeners = client.getRequestListeners(); + for (int i = 0; i < listeners.size(); ++i) + { + Request.Listener listener = listeners.get(i); + notifyContent(listener, request, content); + } + } + + private void notifyContent(Request.ContentListener listener, Request request, ByteBuffer content) + { + try + { + listener.onContent(request, content); + } + catch (Exception x) + { + LOG.info("Exception while notifying listener " + listener, x); + } + } + public void notifySuccess(Request request) { // Optimized to avoid allocations of iterator instances diff --git a/jetty-client/src/main/java/org/eclipse/jetty/client/api/Request.java b/jetty-client/src/main/java/org/eclipse/jetty/client/api/Request.java index 78f68a419b..f5c6705d54 100644 --- a/jetty-client/src/main/java/org/eclipse/jetty/client/api/Request.java +++ b/jetty-client/src/main/java/org/eclipse/jetty/client/api/Request.java @@ -20,6 +20,7 @@ package org.eclipse.jetty.client.api; import java.io.IOException; import java.net.URI; +import java.nio.ByteBuffer; import java.nio.file.Path; import java.util.EventListener; import java.util.List; @@ -266,6 +267,12 @@ public interface Request Request onRequestCommit(CommitListener listener); /** + * @param listener a listener for request content events + * @return this request object + */ + Request onRequestContent(ContentListener listener); + + /** * @param listener a listener for request success event * @return this request object */ @@ -417,6 +424,19 @@ public interface Request } /** + * Listener for the request content event. + */ + public interface ContentListener extends RequestListener + { + /** + * Callback method invoked when a chunk of request content has been sent successfully. + * Changes to bytes in the given buffer have no effect, as the content has already been sent. + * @param request the request that has been committed + */ + public void onContent(Request request, ByteBuffer content); + } + + /** * Listener for the request succeeded event. */ public interface SuccessListener extends RequestListener @@ -445,7 +465,7 @@ public interface Request /** * Listener for all request events. */ - public interface Listener extends QueuedListener, BeginListener, HeadersListener, CommitListener, SuccessListener, FailureListener + public interface Listener extends QueuedListener, BeginListener, HeadersListener, CommitListener, ContentListener, SuccessListener, FailureListener { /** * An empty implementation of {@link Listener} @@ -473,6 +493,11 @@ public interface Request } @Override + public void onContent(Request request, ByteBuffer content) + { + } + + @Override public void onSuccess(Request request) { } diff --git a/jetty-client/src/test/java/org/eclipse/jetty/client/HostnameVerificationTest.java b/jetty-client/src/test/java/org/eclipse/jetty/client/HostnameVerificationTest.java index cf0d813ed9..7fe669c423 100644 --- a/jetty-client/src/test/java/org/eclipse/jetty/client/HostnameVerificationTest.java +++ b/jetty-client/src/test/java/org/eclipse/jetty/client/HostnameVerificationTest.java @@ -19,6 +19,8 @@ package org.eclipse.jetty.client; import java.io.IOException; +import java.nio.channels.ClosedChannelException; +import java.security.cert.CertificateException; import java.util.concurrent.ExecutionException; import javax.net.ssl.SSLHandshakeException; import javax.servlet.ServletException; @@ -37,12 +39,12 @@ import org.junit.Before; import org.junit.Test; import static junit.framework.Assert.fail; -import static org.hamcrest.CoreMatchers.instanceOf; -import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.instanceOf; +import static org.junit.Assert.assertThat; /** - * This test class runs tests to make sure that hostname verification (http://www.ietf.org/rfc/rfc2818.txt section 3 - * .1) is configurable in SslContextFactory and works as expected. + * This test class runs tests to make sure that hostname verification (http://www.ietf.org/rfc/rfc2818.txt + * section 3.1) is configurable in SslContextFactory and works as expected. */ public class HostnameVerificationTest { @@ -107,10 +109,20 @@ public class HostnameVerificationTest client.GET(uri); fail("sending request to client should have failed with an Exception!"); } - catch (ExecutionException e) + catch (ExecutionException x) { - assertThat("We got a SSLHandshakeException as localhost doesn't match the hostname of the certificate", - e.getCause().getCause(), instanceOf(SSLHandshakeException.class)); + // The test may fail in 2 ways, since the CertificateException thrown because of the hostname + // verification failure is not rethrown immediately by the JDK SSL implementation, but only + // rethrown on the next read or write. + // Therefore this test may catch a SSLHandshakeException, or a ClosedChannelException. + // If it is the former, we verify that its cause is a CertificateException. + + // ExecutionException wraps an EofException that wraps the SSLHandshakeException + Throwable cause = x.getCause().getCause(); + if (cause instanceof SSLHandshakeException) + assertThat(cause.getCause().getCause(), instanceOf(CertificateException.class)); + else + assertThat(cause, instanceOf(ClosedChannelException.class)); } } diff --git a/jetty-client/src/test/java/org/eclipse/jetty/client/HttpClientAuthenticationTest.java b/jetty-client/src/test/java/org/eclipse/jetty/client/HttpClientAuthenticationTest.java index 8247c49123..80c3d0b930 100644 --- a/jetty-client/src/test/java/org/eclipse/jetty/client/HttpClientAuthenticationTest.java +++ b/jetty-client/src/test/java/org/eclipse/jetty/client/HttpClientAuthenticationTest.java @@ -49,7 +49,6 @@ import org.eclipse.jetty.toolchain.test.MavenTestingUtils; import org.eclipse.jetty.util.security.Constraint; import org.eclipse.jetty.util.ssl.SslContextFactory; import org.junit.Assert; -import org.junit.Ignore; import org.junit.Test; public class HttpClientAuthenticationTest extends AbstractHttpClientServerTest @@ -104,7 +103,6 @@ public class HttpClientAuthenticationTest extends AbstractHttpClientServerTest test_Authentication(new BasicAuthentication(uri, realm, "basic", "basic")); } - @Ignore @Test public void test_DigestAuthentication() throws Exception { @@ -135,6 +133,7 @@ public class HttpClientAuthenticationTest extends AbstractHttpClientServerTest Assert.assertEquals(401, response.getStatus()); Assert.assertTrue(requests.get().await(5, TimeUnit.SECONDS)); client.getRequestListeners().remove(requestListener); + Assert.assertNull(client.getConversation(request.getConversationID(), false)); authenticationStore.addAuthentication(authentication); diff --git a/jetty-client/src/test/java/org/eclipse/jetty/client/HttpClientTest.java b/jetty-client/src/test/java/org/eclipse/jetty/client/HttpClientTest.java index bb9bea6d97..225078050b 100644 --- a/jetty-client/src/test/java/org/eclipse/jetty/client/HttpClientTest.java +++ b/jetty-client/src/test/java/org/eclipse/jetty/client/HttpClientTest.java @@ -28,6 +28,7 @@ import java.nio.channels.UnresolvedAddressException; import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; +import java.util.Arrays; import java.util.Iterator; import java.util.List; import java.util.NoSuchElementException; @@ -35,6 +36,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; import java.util.zip.GZIPOutputStream; import javax.servlet.ServletException; @@ -286,6 +288,59 @@ public class HttpClientTest extends AbstractHttpClientServerTest } @Test + public void test_POST_WithContent_NotifiesRequestContentListener() throws Exception + { + final byte[] content = {0, 1, 2, 3}; + start(new EmptyServerHandler()); + + ContentResponse response = client.POST(scheme + "://localhost:" + connector.getLocalPort()) + .onRequestContent(new Request.ContentListener() + { + @Override + public void onContent(Request request, ByteBuffer buffer) + { + byte[] bytes = new byte[buffer.remaining()]; + buffer.get(bytes); + if (!Arrays.equals(content, bytes)) + request.abort(new Exception()); + } + }) + .content(new BytesContentProvider(content)) + .timeout(5, TimeUnit.SECONDS) + .send(); + + Assert.assertNotNull(response); + Assert.assertEquals(200, response.getStatus()); + } + + @Test + public void test_POST_WithContent_TracksProgress() throws Exception + { + start(new EmptyServerHandler()); + + final AtomicInteger progress = new AtomicInteger(); + ContentResponse response = client.POST(scheme + "://localhost:" + connector.getLocalPort()) + .onRequestContent(new Request.ContentListener() + { + @Override + public void onContent(Request request, ByteBuffer buffer) + { + byte[] bytes = new byte[buffer.remaining()]; + Assert.assertEquals(1, bytes.length); + buffer.get(bytes); + Assert.assertEquals(bytes[0], progress.getAndIncrement()); + } + }) + .content(new BytesContentProvider(new byte[]{0}, new byte[]{1}, new byte[]{2}, new byte[]{3}, new byte[]{4})) + .timeout(5, TimeUnit.SECONDS) + .send(); + + Assert.assertNotNull(response); + Assert.assertEquals(200, response.getStatus()); + Assert.assertEquals(5, progress.get()); + } + + @Test public void test_QueuedRequest_IsSent_WhenPreviousRequestSucceeded() throws Exception { start(new EmptyServerHandler()); diff --git a/jetty-client/src/test/java/org/eclipse/jetty/client/HttpRequestAbortTest.java b/jetty-client/src/test/java/org/eclipse/jetty/client/HttpRequestAbortTest.java index 293e772c38..5d1a2b1279 100644 --- a/jetty-client/src/test/java/org/eclipse/jetty/client/HttpRequestAbortTest.java +++ b/jetty-client/src/test/java/org/eclipse/jetty/client/HttpRequestAbortTest.java @@ -18,7 +18,6 @@ package org.eclipse.jetty.client; -import java.io.EOFException; import java.io.IOException; import java.nio.ByteBuffer; import java.util.concurrent.CountDownLatch; @@ -26,7 +25,6 @@ import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; - import javax.servlet.ServletException; import javax.servlet.http.HttpServletRequest; import javax.servlet.http.HttpServletResponse; @@ -36,11 +34,9 @@ import org.eclipse.jetty.client.api.Request; import org.eclipse.jetty.client.api.Response; import org.eclipse.jetty.client.api.Result; import org.eclipse.jetty.client.util.ByteBufferContentProvider; -import org.eclipse.jetty.server.HttpChannel; import org.eclipse.jetty.server.handler.AbstractHandler; import org.eclipse.jetty.toolchain.test.annotation.Slow; import org.eclipse.jetty.util.IO; -import org.eclipse.jetty.util.log.StdErrLog; import org.eclipse.jetty.util.ssl.SslContextFactory; import org.junit.Assert; import org.junit.Test; @@ -226,7 +222,6 @@ public class HttpRequestAbortTest extends AbstractHttpClientServerTest } }); - StdErrLog.getLogger(HttpChannel.class).setHideStacks(true); final Throwable cause = new Exception(); try { @@ -254,24 +249,51 @@ public class HttpRequestAbortTest extends AbstractHttpClientServerTest } catch (ExecutionException x) { - Throwable abort = x.getCause(); - if (abort instanceof EOFException) - { - // Server closed abruptly - System.err.println("C"); - } - else if (abort == cause) - { - // Expected - } - else + Assert.assertSame(cause, x.getCause()); + } + } + + @Test + public void testAbortOnContent() throws Exception + { + start(new EmptyServerHandler() + { + @Override + public void handle(String target, org.eclipse.jetty.server.Request baseRequest, HttpServletRequest request, HttpServletResponse response) throws IOException, ServletException { - throw x; + super.handle(target, baseRequest, request, response); + IO.copy(request.getInputStream(), response.getOutputStream()); } + }); + + final Throwable cause = new Exception(); + try + { + client.newRequest("localhost", connector.getLocalPort()) + .scheme(scheme) + .onRequestContent(new Request.ContentListener() + { + @Override + public void onContent(Request request, ByteBuffer content) + { + request.abort(cause); + } + }) + .content(new ByteBufferContentProvider(ByteBuffer.wrap(new byte[]{0}), ByteBuffer.wrap(new byte[]{1})) + { + @Override + public long getLength() + { + return -1; + } + }) + .timeout(5, TimeUnit.SECONDS) + .send(); + Assert.fail(); } - finally + catch (ExecutionException x) { - StdErrLog.getLogger(HttpChannel.class).setHideStacks(false); + Assert.assertSame(cause, x.getCause()); } } |