Improvements to HttpSender. (#12111)
* Changed ContentSender demand from iterate()+IDLE to succeeded()+SCHEDULED. This ensures that there is no re-iteration in case a 100 Continue response arrives. This, in turn, avoids that the demand is performed multiple times, causing ISE to be thrown. * Changed the 100 Continue action of the proxy Servlet/Handler, that provides the request content, to be executed by the HttpSender, rather than by the HttpReceiver. Signed-off-by: Simone Bordet <simone.bordet@gmail.com>
This commit is contained in:
parent
58cfe7709f
commit
fa143fa62a
|
@ -52,8 +52,9 @@ public class ContinueProtocolHandler implements ProtocolHandler
|
|||
return new ContinueListener();
|
||||
}
|
||||
|
||||
protected void onContinue(Request request)
|
||||
protected Runnable onContinue(Request request)
|
||||
{
|
||||
return null;
|
||||
}
|
||||
|
||||
protected class ContinueListener extends BufferingResponseListener
|
||||
|
@ -79,8 +80,10 @@ public class ContinueProtocolHandler implements ProtocolHandler
|
|||
{
|
||||
// All good, continue.
|
||||
exchange.resetResponse();
|
||||
exchange.proceed(null);
|
||||
onContinue(request);
|
||||
Runnable proceedAction = onContinue(request);
|
||||
// Pass the proceed action to be executed
|
||||
// by the sender, not here by the receiver.
|
||||
exchange.proceed(proceedAction, null);
|
||||
}
|
||||
else
|
||||
{
|
||||
|
@ -90,7 +93,7 @@ public class ContinueProtocolHandler implements ProtocolHandler
|
|||
ResponseListeners listeners = exchange.getResponseListeners();
|
||||
HttpContentResponse contentResponse = new HttpContentResponse(response, getContent(), getMediaType(), getEncoding());
|
||||
listeners.emitSuccess(contentResponse);
|
||||
exchange.proceed(new HttpRequestException("Expectation failed", request));
|
||||
exchange.proceed(null, new HttpRequestException("Expectation failed", request));
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
@ -146,9 +146,9 @@ public abstract class HttpChannel implements CyclicTimeouts.Expirable
|
|||
|
||||
public abstract void release();
|
||||
|
||||
public void proceed(HttpExchange exchange, Throwable failure)
|
||||
public void proceed(HttpExchange exchange, Runnable proceedAction, Throwable failure)
|
||||
{
|
||||
getHttpSender().proceed(exchange, failure);
|
||||
getHttpSender().proceed(exchange, proceedAction, failure);
|
||||
}
|
||||
|
||||
public void abort(HttpExchange exchange, Throwable requestFailure, Throwable responseFailure, Promise<Boolean> promise)
|
||||
|
|
|
@ -317,11 +317,11 @@ public class HttpExchange implements CyclicTimeouts.Expirable
|
|||
}
|
||||
}
|
||||
|
||||
public void proceed(Throwable failure)
|
||||
public void proceed(Runnable proceedAction, Throwable failure)
|
||||
{
|
||||
HttpChannel channel = getHttpChannel();
|
||||
if (channel != null)
|
||||
channel.proceed(this, failure);
|
||||
channel.proceed(this, proceedAction, failure);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
|
@ -317,12 +317,15 @@ public abstract class HttpSender
|
|||
{
|
||||
}
|
||||
|
||||
public void proceed(HttpExchange exchange, Throwable failure)
|
||||
public void proceed(HttpExchange exchange, Runnable proceedAction, Throwable failure)
|
||||
{
|
||||
// Received a 100 Continue, although Expect header was not sent.
|
||||
// Received a 100 Continue, although the Expect header was not sent.
|
||||
if (!contentSender.expect100)
|
||||
return;
|
||||
|
||||
// Write the fields in this order, since the reader of
|
||||
// these fields will read them in the opposite order.
|
||||
contentSender.proceedAction = proceedAction;
|
||||
contentSender.expect100 = false;
|
||||
if (failure == null)
|
||||
{
|
||||
|
@ -462,32 +465,39 @@ public abstract class HttpSender
|
|||
|
||||
private class ContentSender extends IteratingCallback
|
||||
{
|
||||
private HttpExchange exchange;
|
||||
// Fields that are set externally.
|
||||
private volatile HttpExchange exchange;
|
||||
private volatile Runnable proceedAction;
|
||||
private volatile boolean expect100;
|
||||
// Fields only used internally.
|
||||
private Content.Chunk chunk;
|
||||
private ByteBuffer contentBuffer;
|
||||
private boolean expect100;
|
||||
private boolean committed;
|
||||
private boolean success;
|
||||
private boolean complete;
|
||||
private Promise<Boolean> abort;
|
||||
private boolean demanded;
|
||||
|
||||
@Override
|
||||
public boolean reset()
|
||||
{
|
||||
exchange = null;
|
||||
proceedAction = null;
|
||||
expect100 = false;
|
||||
chunk = null;
|
||||
contentBuffer = null;
|
||||
expect100 = false;
|
||||
committed = false;
|
||||
success = false;
|
||||
complete = false;
|
||||
abort = null;
|
||||
demanded = false;
|
||||
return super.reset();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected Action process() throws Throwable
|
||||
{
|
||||
HttpExchange exchange = this.exchange;
|
||||
if (complete)
|
||||
{
|
||||
if (success)
|
||||
|
@ -498,15 +508,26 @@ public abstract class HttpSender
|
|||
HttpRequest request = exchange.getRequest();
|
||||
Content.Source content = request.getBody();
|
||||
|
||||
boolean expect100 = this.expect100;
|
||||
if (expect100)
|
||||
{
|
||||
// If the request was sent already, wait for
|
||||
// the 100 response before sending the content.
|
||||
if (committed)
|
||||
return Action.IDLE;
|
||||
else
|
||||
chunk = null;
|
||||
// Do not send any content yet.
|
||||
chunk = null;
|
||||
}
|
||||
else
|
||||
{
|
||||
// Run the proceed action first, which likely will provide
|
||||
// content after having received the 100 Continue response.
|
||||
Runnable action = proceedAction;
|
||||
proceedAction = null;
|
||||
if (action != null)
|
||||
action.run();
|
||||
|
||||
// Read the request content.
|
||||
chunk = content.read();
|
||||
}
|
||||
if (LOG.isDebugEnabled())
|
||||
|
@ -516,11 +537,14 @@ public abstract class HttpSender
|
|||
{
|
||||
if (committed)
|
||||
{
|
||||
content.demand(this::iterate);
|
||||
return Action.IDLE;
|
||||
// No content after the headers, demand.
|
||||
demanded = true;
|
||||
content.demand(this::succeeded);
|
||||
return Action.SCHEDULED;
|
||||
}
|
||||
else
|
||||
{
|
||||
// Normalize to avoid null checks.
|
||||
chunk = Content.Chunk.EMPTY;
|
||||
}
|
||||
}
|
||||
|
@ -545,48 +569,49 @@ public abstract class HttpSender
|
|||
@Override
|
||||
protected void onSuccess()
|
||||
{
|
||||
boolean proceed = true;
|
||||
if (committed)
|
||||
if (demanded)
|
||||
{
|
||||
if (contentBuffer.hasRemaining())
|
||||
proceed = someToContent(exchange, contentBuffer);
|
||||
// Content is now available, reset
|
||||
// the demand and iterate again.
|
||||
demanded = false;
|
||||
}
|
||||
else
|
||||
{
|
||||
committed = true;
|
||||
if (headersToCommit(exchange))
|
||||
boolean proceed = true;
|
||||
if (committed)
|
||||
{
|
||||
// Was any content sent while committing?
|
||||
if (contentBuffer.hasRemaining())
|
||||
proceed = someToContent(exchange, contentBuffer);
|
||||
}
|
||||
else
|
||||
{
|
||||
proceed = false;
|
||||
committed = true;
|
||||
proceed = headersToCommit(exchange);
|
||||
if (proceed)
|
||||
{
|
||||
// Was any content sent while committing?
|
||||
if (contentBuffer.hasRemaining())
|
||||
proceed = someToContent(exchange, contentBuffer);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
boolean last = chunk.isLast();
|
||||
chunk.release();
|
||||
chunk = null;
|
||||
boolean last = chunk.isLast();
|
||||
chunk.release();
|
||||
chunk = null;
|
||||
|
||||
if (proceed)
|
||||
{
|
||||
if (last)
|
||||
if (proceed)
|
||||
{
|
||||
success = true;
|
||||
if (last)
|
||||
{
|
||||
success = true;
|
||||
complete = true;
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
// There was some concurrent error, terminate.
|
||||
complete = true;
|
||||
}
|
||||
else if (expect100)
|
||||
{
|
||||
if (LOG.isDebugEnabled())
|
||||
LOG.debug("Expecting 100 Continue for {}", exchange.getRequest());
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
// There was some concurrent error, terminate.
|
||||
complete = true;
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
@ -444,12 +444,11 @@ public abstract class ProxyHandler extends Handler.Abstract
|
|||
Response.writeError(clientToProxyRequest, proxyToClientResponse, callback, status);
|
||||
}
|
||||
|
||||
protected void onServerToProxyResponse100Continue(Request clientToProxyRequest, org.eclipse.jetty.client.Request proxyToServerRequest)
|
||||
protected Runnable onServerToProxyResponse100Continue(Request clientToProxyRequest, org.eclipse.jetty.client.Request proxyToServerRequest)
|
||||
{
|
||||
if (LOG.isDebugEnabled())
|
||||
LOG.debug("{} P2C 100 continue response", requestId(clientToProxyRequest));
|
||||
Runnable action = (Runnable)proxyToServerRequest.getAttributes().get(PROXY_TO_SERVER_CONTINUE_ATTRIBUTE);
|
||||
action.run();
|
||||
return (Runnable)proxyToServerRequest.getAttributes().get(PROXY_TO_SERVER_CONTINUE_ATTRIBUTE);
|
||||
}
|
||||
|
||||
protected void onServerToProxyResponse102Processing(Request clientToProxyRequest, org.eclipse.jetty.client.Request proxyToServerRequest, HttpFields serverToProxyResponseHeaders, Response proxyToClientResponse)
|
||||
|
@ -776,13 +775,12 @@ public abstract class ProxyHandler extends Handler.Abstract
|
|||
private class ProxyContinueProtocolHandler extends ContinueProtocolHandler
|
||||
{
|
||||
@Override
|
||||
protected void onContinue(org.eclipse.jetty.client.Request proxyToServerRequest)
|
||||
protected Runnable onContinue(org.eclipse.jetty.client.Request proxyToServerRequest)
|
||||
{
|
||||
super.onContinue(proxyToServerRequest);
|
||||
var clientToProxyRequest = (Request)proxyToServerRequest.getAttributes().get(CLIENT_TO_PROXY_REQUEST_ATTRIBUTE);
|
||||
if (LOG.isDebugEnabled())
|
||||
LOG.debug("{} S2P received 100 Continue", requestId(clientToProxyRequest));
|
||||
onServerToProxyResponse100Continue(clientToProxyRequest, proxyToServerRequest);
|
||||
return onServerToProxyResponse100Continue(clientToProxyRequest, proxyToServerRequest);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
@ -763,10 +763,9 @@ public abstract class AbstractProxyServlet extends HttpServlet
|
|||
}
|
||||
}
|
||||
|
||||
protected void onContinue(HttpServletRequest clientRequest, Request proxyRequest)
|
||||
protected Runnable onContinue(HttpServletRequest clientRequest, Request proxyRequest)
|
||||
{
|
||||
if (_log.isDebugEnabled())
|
||||
_log.debug("{} handling 100 Continue", getRequestId(clientRequest));
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
|
@ -851,10 +850,10 @@ public abstract class AbstractProxyServlet extends HttpServlet
|
|||
class ProxyContinueProtocolHandler extends ContinueProtocolHandler
|
||||
{
|
||||
@Override
|
||||
protected void onContinue(Request request)
|
||||
protected Runnable onContinue(Request request)
|
||||
{
|
||||
HttpServletRequest clientRequest = (HttpServletRequest)request.getAttributes().get(CLIENT_REQUEST_ATTRIBUTE);
|
||||
AbstractProxyServlet.this.onContinue(clientRequest, request);
|
||||
return AbstractProxyServlet.this.onContinue(clientRequest, request);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
@ -171,11 +171,11 @@ public class AsyncMiddleManServlet extends AbstractProxyServlet
|
|||
}
|
||||
|
||||
@Override
|
||||
protected void onContinue(HttpServletRequest clientRequest, Request proxyRequest)
|
||||
protected Runnable onContinue(HttpServletRequest clientRequest, Request proxyRequest)
|
||||
{
|
||||
super.onContinue(clientRequest, proxyRequest);
|
||||
Runnable action = (Runnable)proxyRequest.getAttributes().get(CONTINUE_ACTION_ATTRIBUTE);
|
||||
action.run();
|
||||
if (_log.isDebugEnabled())
|
||||
_log.debug("{} handling 100 Continue", getRequestId(clientRequest));
|
||||
return (Runnable)proxyRequest.getAttributes().get(CONTINUE_ACTION_ATTRIBUTE);
|
||||
}
|
||||
|
||||
private void transform(ContentTransformer transformer, ByteBuffer input, boolean finished, List<ByteBuffer> output) throws IOException
|
||||
|
|
|
@ -16,7 +16,6 @@ package org.eclipse.jetty.ee10.proxy;
|
|||
import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.util.concurrent.Executor;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import jakarta.servlet.AsyncContext;
|
||||
|
@ -144,12 +143,11 @@ public class ProxyServlet extends AbstractProxyServlet
|
|||
}
|
||||
|
||||
@Override
|
||||
protected void onContinue(HttpServletRequest clientRequest, Request proxyRequest)
|
||||
protected Runnable onContinue(HttpServletRequest clientRequest, Request proxyRequest)
|
||||
{
|
||||
super.onContinue(clientRequest, proxyRequest);
|
||||
Runnable action = (Runnable)proxyRequest.getAttributes().get(CONTINUE_ACTION_ATTRIBUTE);
|
||||
Executor executor = getHttpClient().getExecutor();
|
||||
executor.execute(action);
|
||||
if (_log.isDebugEnabled())
|
||||
_log.debug("{} handling 100 Continue", getRequestId(clientRequest));
|
||||
return (Runnable)proxyRequest.getAttributes().get(CONTINUE_ACTION_ATTRIBUTE);
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
@ -768,10 +768,17 @@ public abstract class AbstractProxyServlet extends HttpServlet
|
|||
}
|
||||
}
|
||||
|
||||
protected void onContinue(HttpServletRequest clientRequest, Request proxyRequest)
|
||||
/**
|
||||
* <p>Returns the action to perform when the proxy receives
|
||||
* a 100 Continue response from the server.</p>
|
||||
*
|
||||
* @param clientRequest the client request
|
||||
* @param proxyRequest the request being proxied
|
||||
* @return the 100 Continue action to run
|
||||
*/
|
||||
protected Runnable onContinue(HttpServletRequest clientRequest, Request proxyRequest)
|
||||
{
|
||||
if (_log.isDebugEnabled())
|
||||
_log.debug("{} handling 100 Continue", getRequestId(clientRequest));
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
|
@ -856,10 +863,10 @@ public abstract class AbstractProxyServlet extends HttpServlet
|
|||
class ProxyContinueProtocolHandler extends ContinueProtocolHandler
|
||||
{
|
||||
@Override
|
||||
protected void onContinue(Request request)
|
||||
protected Runnable onContinue(Request request)
|
||||
{
|
||||
HttpServletRequest clientRequest = (HttpServletRequest)request.getAttributes().get(CLIENT_REQUEST_ATTRIBUTE);
|
||||
AbstractProxyServlet.this.onContinue(clientRequest, request);
|
||||
return AbstractProxyServlet.this.onContinue(clientRequest, request);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
@ -171,11 +171,11 @@ public class AsyncMiddleManServlet extends AbstractProxyServlet
|
|||
}
|
||||
|
||||
@Override
|
||||
protected void onContinue(HttpServletRequest clientRequest, Request proxyRequest)
|
||||
protected Runnable onContinue(HttpServletRequest clientRequest, Request proxyRequest)
|
||||
{
|
||||
super.onContinue(clientRequest, proxyRequest);
|
||||
Runnable action = (Runnable)proxyRequest.getAttributes().get(CONTINUE_ACTION_ATTRIBUTE);
|
||||
action.run();
|
||||
if (_log.isDebugEnabled())
|
||||
_log.debug("{} handling 100 Continue", getRequestId(clientRequest));
|
||||
return (Runnable)proxyRequest.getAttributes().get(CONTINUE_ACTION_ATTRIBUTE);
|
||||
}
|
||||
|
||||
private void transform(ContentTransformer transformer, ByteBuffer input, boolean finished, List<ByteBuffer> output) throws IOException
|
||||
|
|
|
@ -16,7 +16,6 @@ package org.eclipse.jetty.ee9.proxy;
|
|||
import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.util.concurrent.Executor;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import jakarta.servlet.AsyncContext;
|
||||
|
@ -144,12 +143,11 @@ public class ProxyServlet extends AbstractProxyServlet
|
|||
}
|
||||
|
||||
@Override
|
||||
protected void onContinue(HttpServletRequest clientRequest, Request proxyRequest)
|
||||
protected Runnable onContinue(HttpServletRequest clientRequest, Request proxyRequest)
|
||||
{
|
||||
super.onContinue(clientRequest, proxyRequest);
|
||||
Runnable action = (Runnable)proxyRequest.getAttributes().get(CONTINUE_ACTION_ATTRIBUTE);
|
||||
Executor executor = getHttpClient().getExecutor();
|
||||
executor.execute(action);
|
||||
if (_log.isDebugEnabled())
|
||||
_log.debug("{} handling 100 Continue", getRequestId(clientRequest));
|
||||
return (Runnable)proxyRequest.getAttributes().get(CONTINUE_ACTION_ATTRIBUTE);
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
Loading…
Reference in New Issue