Eclipse
/
jetty.project
mirror of https://github.com/jetty/jetty.project.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #4556 from lorban/http-input 

HttpInput refactoring
This commit is contained in:
Simone Bordet 2020-11-03 18:54:23 +01:00 committed by GitHub
parent 167eded4cf 814dc69803
commit ebea687b5b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
31 changed files with 3477 additions and 2682 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
jetty-client/src/test/java/org/eclipse/jetty/client/ssl/SslBytesServerTest.java
View File

@ -1377,9 +1377,12 @@ public class SslBytesServerTest extends SslBytesTest
// Check that we did not spin
TimeUnit.MILLISECONDS.sleep(500);
assertThat(sslFills.get(), Matchers.lessThan(100));
// The new HttpInput impl tends to call fill and parse more often than the previous one
// b/c HttpChannel.needContent() does a fill and parse before doing a fill interested;
// this runs the parser an goes to the OS more often but requires less rescheduling.
assertThat(sslFills.get(), Matchers.lessThan(150));
assertThat(sslFlushes.get(), Matchers.lessThan(50));
assertThat(httpParses.get(), Matchers.lessThan(100));
assertThat(httpParses.get(), Matchers.lessThan(150));
assertNull(request.get(5, TimeUnit.SECONDS));
@ -1399,9 +1402,12 @@ public class SslBytesServerTest extends SslBytesTest
// Check that we did not spin
TimeUnit.MILLISECONDS.sleep(500);
assertThat(sslFills.get(), Matchers.lessThan(100));
// The new HttpInput impl tends to call fill and parse more often than the previous one
// b/c HttpChannel.needContent() does a fill and parse before doing a fill interested;
// this runs the parser an goes to the OS more often but requires less rescheduling.
assertThat(sslFills.get(), Matchers.lessThan(150));
assertThat(sslFlushes.get(), Matchers.lessThan(50));
assertThat(httpParses.get(), Matchers.lessThan(100));
assertThat(httpParses.get(), Matchers.lessThan(150));
closeClient(client);
}
@ -1596,9 +1602,12 @@ public class SslBytesServerTest extends SslBytesTest
// Check that we did not spin
TimeUnit.MILLISECONDS.sleep(500);
assertThat(sslFills.get(), Matchers.lessThan(50));
// The new HttpInput impl tends to call fill and parse more often than the previous one
// b/c HttpChannel.needContent() does a fill and parse before doing a fill interested;
// this runs the parser and goes to the OS more often but requires less rescheduling.
assertThat(sslFills.get(), Matchers.lessThan(70));
assertThat(sslFlushes.get(), Matchers.lessThan(20));
assertThat(httpParses.get(), Matchers.lessThan(50));
assertThat(httpParses.get(), Matchers.lessThan(70));
closeClient(client);
}
@ -1743,9 +1752,12 @@ public class SslBytesServerTest extends SslBytesTest
// Check that we did not spin
TimeUnit.MILLISECONDS.sleep(500);
assertThat(sslFills.get(), Matchers.lessThan(50));
// The new HttpInput impl tends to call fill and parse more often than the previous one
// b/c HttpChannel.needContent() does a fill and parse before doing a fill interested;
// this runs the parser and goes to the OS more often but requires less rescheduling.
assertThat(sslFills.get(), Matchers.lessThan(80));
assertThat(sslFlushes.get(), Matchers.lessThan(20));
assertThat(httpParses.get(), Matchers.lessThan(100));
assertThat(httpParses.get(), Matchers.lessThan(120));
closeClient(client);
}

140
jetty-fcgi/fcgi-server/src/main/java/org/eclipse/jetty/fcgi/server/HttpChannelOverFCGI.java
View File

@ -18,7 +18,11 @@
package org.eclipse.jetty.fcgi.server;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Locale;
import java.util.Queue;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicReference;
@ -34,8 +38,10 @@ import org.eclipse.jetty.io.EndPoint;
import org.eclipse.jetty.server.Connector;
import org.eclipse.jetty.server.HttpChannel;
import org.eclipse.jetty.server.HttpConfiguration;
import org.eclipse.jetty.server.HttpInput;
import org.eclipse.jetty.server.HttpTransport;
import org.eclipse.jetty.util.StringUtil;
import org.eclipse.jetty.util.thread.AutoLock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -43,6 +49,9 @@ public class HttpChannelOverFCGI extends HttpChannel
{
private static final Logger LOG = LoggerFactory.getLogger(HttpChannelOverFCGI.class);
private final Queue<HttpInput.Content> _contentQueue = new LinkedList<>();
private final AutoLock _lock = new AutoLock();
private HttpInput.Content _specialContent;
private final HttpFields.Mutable fields = HttpFields.build();
private final Dispatcher dispatcher;
private String method;
@ -57,6 +66,101 @@ public class HttpChannelOverFCGI extends HttpChannel
this.dispatcher = new Dispatcher(connector.getServer().getThreadPool(), this);
}
@Override
public boolean onContent(HttpInput.Content content)
{
boolean b = super.onContent(content);
Throwable failure;
try (AutoLock l = _lock.lock())
{
failure = _specialContent == null ? null : _specialContent.getError();
if (failure == null)
_contentQueue.offer(content);
}
if (failure != null)
content.failed(failure);
return b;
}
@Override
public boolean needContent()
{
try (AutoLock l = _lock.lock())
{
boolean hasContent = _specialContent != null || !_contentQueue.isEmpty();
if (LOG.isDebugEnabled())
LOG.debug("needContent has content? {}", hasContent);
return hasContent;
}
}
@Override
public HttpInput.Content produceContent()
{
HttpInput.Content content;
try (AutoLock l = _lock.lock())
{
content = _contentQueue.poll();
if (content == null)
content = _specialContent;
}
if (LOG.isDebugEnabled())
LOG.debug("produceContent has produced {}", content);
return content;
}
@Override
public boolean failAllContent(Throwable failure)
{
if (LOG.isDebugEnabled())
LOG.debug("failing all content with {}", (Object)failure);
List<HttpInput.Content> copy;
try (AutoLock l = _lock.lock())
{
copy = new ArrayList<>(_contentQueue);
_contentQueue.clear();
}
copy.forEach(c -> c.failed(failure));
HttpInput.Content lastContent = copy.isEmpty() ? null : copy.get(copy.size() - 1);
boolean atEof = lastContent != null && lastContent.isEof();
if (LOG.isDebugEnabled())
LOG.debug("failed all content, EOF = {}", atEof);
return atEof;
}
@Override
public boolean failed(Throwable x)
{
if (LOG.isDebugEnabled())
LOG.debug("failed " + x);
try (AutoLock l = _lock.lock())
{
Throwable error = _specialContent == null ? null : _specialContent.getError();
if (error != null && error != x)
error.addSuppressed(x);
else
_specialContent = new HttpInput.ErrorContent(x);
}
return getRequest().getHttpInput().onContentProducible();
}
@Override
protected boolean eof()
{
if (LOG.isDebugEnabled())
LOG.debug("received EOF");
try (AutoLock l = _lock.lock())
{
_specialContent = new HttpInput.EofContent();
}
return getRequest().getHttpInput().onContentProducible();
}
protected void header(HttpField field)
{
String name = field.getName();
@ -127,12 +231,46 @@ public class HttpChannelOverFCGI extends HttpChannel
public boolean onIdleTimeout(Throwable timeout)
{
boolean handle = getRequest().getHttpInput().onIdleTimeout(timeout);
boolean handle = doOnIdleTimeout(timeout);
if (handle)
execute(this);
return !handle;
}
private boolean doOnIdleTimeout(Throwable x)
{
boolean neverDispatched = getState().isIdle();
boolean waitingForContent;
HttpInput.Content specialContent;
try (AutoLock l = _lock.lock())
{
waitingForContent = _contentQueue.isEmpty() || _contentQueue.peek().remaining() == 0;
specialContent = _specialContent;
}
if ((waitingForContent || neverDispatched) && specialContent == null)
{
x.addSuppressed(new Throwable("HttpInput idle timeout"));
try (AutoLock l = _lock.lock())
{
_specialContent = new HttpInput.ErrorContent(x);
}
return getRequest().getHttpInput().onContentProducible();
}
return false;
}
@Override
public void recycle()
{
try (AutoLock l = _lock.lock())
{
if (!_contentQueue.isEmpty())
throw new AssertionError("unconsumed content: " + _contentQueue);
_specialContent = null;
}
super.recycle();
}
private static class Dispatcher implements Runnable
{
private final AtomicReference<State> state = new AtomicReference<>(State.IDLE);

16
jetty-http2/http2-common/src/main/java/org/eclipse/jetty/http2/HTTP2Stream.java
View File

@ -22,6 +22,7 @@ import java.io.EOFException;
import java.io.IOException;
import java.nio.channels.WritePendingException;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
@ -235,6 +236,21 @@ public class HTTP2Stream extends IdleTimeout implements IStream, Callback, Dumpa
return state == CloseState.REMOTELY_CLOSED || state == CloseState.CLOSING || state == CloseState.CLOSED;
}
@Override
public boolean failAllData(Throwable x)
{
List<DataEntry> copy;
try (AutoLock l = lock.lock())
{
dataDemand = 0;
copy = new ArrayList<>(dataQueue);
dataQueue.clear();
}
copy.forEach(dataEntry -> dataEntry.callback.failed(x));
DataEntry lastDataEntry = copy.isEmpty() ? null : copy.get(copy.size() - 1);
return lastDataEntry != null && lastDataEntry.frame.isEndStream();
}
public boolean isLocallyClosed()
{
return closeState.get() == CloseState.LOCALLY_CLOSED;

3
jetty-http2/http2-common/src/main/java/org/eclipse/jetty/http2/HTTP2StreamEndPoint.java
View File

@ -216,6 +216,9 @@ public abstract class HTTP2StreamEndPoint implements EndPoint
else
{
entry.succeed();
// WebSocket does not have a backpressure API so you must always demand
// the next frame after succeeding the previous one.
stream.demand(1);
}
return length;
}

8
jetty-http2/http2-common/src/main/java/org/eclipse/jetty/http2/IStream.java
View File

@ -119,6 +119,14 @@ public interface IStream extends Stream, Attachable, Closeable
*/
boolean isRemotelyClosed();
/**
* Fail all data queued in the stream and reset
* demand to 0.
* @param x the exception to fail the data with.
* @return true if the end of the stream was reached, false otherwise.
*/
boolean failAllData(Throwable x);
/**
* @return whether this stream has been reset (locally or remotely) or has been failed
* @see #isReset()

5
jetty-http2/http2-common/src/main/java/org/eclipse/jetty/http2/api/Stream.java
View File

@ -242,7 +242,10 @@ public interface Stream
* @param callback the callback to complete when the bytes of the DATA frame have been consumed
* @see #onDataDemanded(Stream, DataFrame, Callback)
*/
public void onData(Stream stream, DataFrame frame, Callback callback);
public default void onData(Stream stream, DataFrame frame, Callback callback)
{
callback.succeeded();
}
/**
* <p>Callback method invoked when a DATA frame has been demanded.</p>

26
jetty-http2/http2-server/src/main/java/org/eclipse/jetty/http2/server/ContentDemander_state.puml Normal file
View File

@ -0,0 +1,26 @@
@startuml
null:
content:
DEMANDING:
EOF:
[*] --> null
null --> DEMANDING : demand()
null --> EOF : eof()
null -left-> null : onTimeout()
DEMANDING --> DEMANDING : demand()
DEMANDING --> content : onContent()\n onTimeout()
DEMANDING --> EOF : eof()
EOF --> EOF : eof()\n onTimeout()
note bottom of content: content1 -> content2 is only\nvalid if content1 is special
note top of content: content -> null only happens\nwhen content is not special
content --> content : onContent()\n onTimeout()
content --> null: take()
content --> EOF: eof()
@enduml

2
jetty-http2/http2-server/src/main/java/org/eclipse/jetty/http2/server/HTTP2ServerConnectionFactory.java
View File

@ -157,7 +157,7 @@ public class HTTP2ServerConnectionFactory extends AbstractHTTP2ServerConnectionF
}
@Override
public void onData(Stream stream, DataFrame frame, Callback callback)
public void onDataDemanded(Stream stream, DataFrame frame, Callback callback)
{
getConnection().onData((IStream)stream, frame, callback);
}

395
jetty-http2/http2-server/src/main/java/org/eclipse/jetty/http2/server/HttpChannelOverHTTP2.java
View File

@ -21,6 +21,7 @@ package org.eclipse.jetty.http2.server;
import java.io.Closeable;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import org.eclipse.jetty.http.BadMessageException;
@ -57,10 +58,12 @@ public class HttpChannelOverHTTP2 extends HttpChannel implements Closeable, Writ
private boolean _expect100Continue;
private boolean _delayedUntilContent;
private boolean _useOutputDirectByteBuffers;
private final ContentDemander _contentDemander;
public HttpChannelOverHTTP2(Connector connector, HttpConfiguration configuration, EndPoint endPoint, HttpTransportOverHTTP2 transport)
{
super(connector, configuration, endPoint, transport);
_contentDemander = new ContentDemander();
}
protected IStream getStream()
@ -131,9 +134,18 @@ public class HttpChannelOverHTTP2 extends HttpChannel implements Closeable, Writ
_delayedUntilContent = getHttpConfiguration().isDelayDispatchUntilContent() &&
!endStream && !_expect100Continue && !connect;
// Delay the demand of DATA frames for CONNECT with :protocol.
if (!connect || request.getProtocol() == null)
getStream().demand(1);
// Delay the demand of DATA frames for CONNECT with :protocol
// or for normal requests expecting 100 continue.
if (connect)
{
if (request.getProtocol() == null)
_contentDemander.demand(false);
}
else
{
if (_delayedUntilContent)
_contentDemander.demand(false);
}
if (LOG.isDebugEnabled())
{
@ -204,6 +216,7 @@ public class HttpChannelOverHTTP2 extends HttpChannel implements Closeable, Writ
{
_expect100Continue = false;
_delayedUntilContent = false;
_contentDemander.recycle();
super.recycle();
getHttpTransport().recycle();
}
@ -224,26 +237,16 @@ public class HttpChannelOverHTTP2 extends HttpChannel implements Closeable, Writ
@Override
public Runnable onData(DataFrame frame, Callback callback)
{
return onRequestContent(frame, callback);
}
public Runnable onRequestContent(DataFrame frame, final Callback callback)
{
Stream stream = getStream();
if (stream.isReset())
{
// Consume previously queued content to
// enlarge the session flow control window.
consumeInput();
// Consume immediately this content.
callback.succeeded();
return null;
}
ByteBuffer buffer = frame.getData();
int length = buffer.remaining();
boolean handle = onContent(new HttpInput.Content(buffer)
HttpInput.Content content = new HttpInput.Content(buffer)
{
@Override
public boolean isEof()
{
return frame.isEndStream();
}
@Override
public void succeeded()
{
@ -261,23 +264,31 @@ public class HttpChannelOverHTTP2 extends HttpChannel implements Closeable, Writ
{
return callback.getInvocationType();
}
});
};
boolean needed = _contentDemander.onContent(content);
boolean handle = onContent(content);
boolean endStream = frame.isEndStream();
if (endStream)
{
boolean handleContent = onContentComplete();
// This will generate EOF -> must happen before onContentProducible.
boolean handleRequest = onRequestComplete();
handle |= handleContent | handleRequest;
}
boolean woken = needed && getRequest().getHttpInput().onContentProducible();
handle |= woken;
if (LOG.isDebugEnabled())
{
LOG.debug("HTTP2 Request #{}/{}: {} bytes of {} content, handle: {}",
Stream stream = getStream();
LOG.debug("HTTP2 Request #{}/{}: {} bytes of {} content, woken: {}, needed: {}, handle: {}",
stream.getId(),
Integer.toHexString(stream.getSession().hashCode()),
length,
endStream ? "last" : "some",
woken,
needed,
handle);
}
@ -286,6 +297,326 @@ public class HttpChannelOverHTTP2 extends HttpChannel implements Closeable, Writ
return handle || wasDelayed ? this : null;
}
/**
* Demanding content is a marker content that is used to remember that a demand was
* registered into the stream. The {@code needed} flag indicates if the demand originated
* from a call to {@link #produceContent()} when false or {@link #needContent()}
* when true, as {@link HttpInput#onContentProducible()} must only be called
* only when {@link #needContent()} was called.
* Instances of this class must never escape the scope of this channel impl,
* so {@link #produceContent()} must never return one.
*/
private static final class DemandingContent extends HttpInput.SpecialContent
{
private final boolean needed;
private DemandingContent(boolean needed)
{
this.needed = needed;
}
}
private static final HttpInput.Content EOF = new HttpInput.EofContent();
private static final HttpInput.Content DEMANDING_NEEDED = new DemandingContent(true);
private static final HttpInput.Content DEMANDING_NOT_NEEDED = new DemandingContent(false);
private class ContentDemander
{
private final AtomicReference<HttpInput.Content> _content = new AtomicReference<>();
public void recycle()
{
if (LOG.isDebugEnabled())
LOG.debug("recycle {}", this);
HttpInput.Content c = _content.getAndSet(null);
if (c != null && !c.isSpecial())
throw new AssertionError("unconsumed content: " + c);
}
public HttpInput.Content poll()
{
while (true)
{
HttpInput.Content c = _content.get();
if (LOG.isDebugEnabled())
LOG.debug("poll, content = {}", c);
if (c == null || c.isSpecial() || _content.compareAndSet(c, c.isEof() ? EOF : null))
{
if (LOG.isDebugEnabled())
LOG.debug("returning current content");
return c;
}
}
}
public boolean demand(boolean needed)
{
while (true)
{
HttpInput.Content c = _content.get();
if (LOG.isDebugEnabled())
LOG.debug("demand({}), content = {}", needed, c);
if (c instanceof DemandingContent)
{
if (needed && !((DemandingContent)c).needed)
{
if (!_content.compareAndSet(c, DEMANDING_NEEDED))
{
if (LOG.isDebugEnabled())
LOG.debug("already demanding but switched needed flag to true");
continue;
}
}
if (LOG.isDebugEnabled())
LOG.debug("already demanding, returning false");
return false;
}
if (c != null)
{
if (LOG.isDebugEnabled())
LOG.debug("content available, returning true");
return true;
}
if (_content.compareAndSet(null, needed ? DEMANDING_NEEDED : DEMANDING_NOT_NEEDED))
{
IStream stream = getStream();
if (stream == null)
{
_content.set(null);
if (LOG.isDebugEnabled())
LOG.debug("no content available, switched to demanding but stream is now null");
return false;
}
if (LOG.isDebugEnabled())
LOG.debug("no content available, demanding stream {}", stream);
stream.demand(1);
c = _content.get();
boolean hasContent = !(c instanceof DemandingContent) && c != null;
if (LOG.isDebugEnabled())
LOG.debug("has content now? {}", hasContent);
return hasContent;
}
}
}
public boolean onContent(HttpInput.Content content)
{
while (true)
{
HttpInput.Content c = _content.get();
if (LOG.isDebugEnabled())
LOG.debug("content delivered by stream: {}, current content: {}", content, c);
if (c instanceof DemandingContent)
{
if (_content.compareAndSet(c, content))
{
boolean needed = ((DemandingContent)c).needed;
if (LOG.isDebugEnabled())
LOG.debug("replacing demand content with {} succeeded; returning {}", content, needed);
return needed;
}
}
else if (c == null)
{
if (!content.isSpecial())
{
// This should never happen, consider as a bug.
content.failed(new IllegalStateException("Non special content without demand : " + content));
return false;
}
if (_content.compareAndSet(null, content))
{
if (LOG.isDebugEnabled())
LOG.debug("replacing null content with {} succeeded", content);
return false;
}
}
else if (c.isEof() && content.isEof() && content.isEmpty())
{
content.succeeded();
return true;
}
else if (content.getError() != null)
{
if (c.getError() != null)
{
if (c.getError() != content.getError())
c.getError().addSuppressed(content.getError());
return true;
}
if (_content.compareAndSet(c, content))
{
c.failed(content.getError());
if (LOG.isDebugEnabled())
LOG.debug("replacing current content with {} succeeded", content);
return true;
}
}
else if (c.getError() != null && content.remaining() == 0)
{
content.succeeded();
return true;
}
else
{
// This should never happen, consider as a bug.
content.failed(new IllegalStateException("Cannot overwrite exiting content " + c + " with " + content));
return false;
}
}
}
public boolean onTimeout(Throwable failure)
{
while (true)
{
HttpInput.Content c = _content.get();
if (LOG.isDebugEnabled())
LOG.debug("onTimeout with current content: {} and failure = {}", c, failure);
if (!(c instanceof DemandingContent))
return false;
if (_content.compareAndSet(c, new HttpInput.ErrorContent(failure)))
{
if (LOG.isDebugEnabled())
LOG.debug("replacing current content with error succeeded");
return true;
}
}
}
public void eof()
{
while (true)
{
HttpInput.Content c = _content.get();
if (LOG.isDebugEnabled())
LOG.debug("eof with current content: {}", c);
if (c instanceof DemandingContent)
{
if (_content.compareAndSet(c, EOF))
{
if (LOG.isDebugEnabled())
LOG.debug("replacing current content with special EOF succeeded");
return;
}
}
else if (c == null)
{
if (_content.compareAndSet(null, EOF))
{
if (LOG.isDebugEnabled())
LOG.debug("replacing null content with special EOF succeeded");
return;
}
}
else if (c.isEof())
{
if (LOG.isDebugEnabled())
LOG.debug("current content already is EOF");
return;
}
else if (c.remaining() == 0)
{
if (_content.compareAndSet(c, EOF))
{
if (LOG.isDebugEnabled())
LOG.debug("replacing current content with special EOF succeeded");
return;
}
}
else
{
// EOF may arrive with HEADERS frame (e.g. a trailer) that is not flow controlled, so we need to wrap the existing content.
// Covered by HttpTrailersTest.testRequestTrailersWithContent.
HttpInput.Content content = new HttpInput.WrappingContent(c, true);
if (_content.compareAndSet(c, content))
{
if (LOG.isDebugEnabled())
LOG.debug("replacing current content with {} succeeded", content);
return;
}
}
}
}
public boolean failContent(Throwable failure)
{
while (true)
{
HttpInput.Content c = _content.get();
if (LOG.isDebugEnabled())
LOG.debug("failing current content: {} with failure: {}", c, failure);
if (c == null)
return false;
if (c.isSpecial())
return c.isEof();
if (_content.compareAndSet(c, null))
{
c.failed(failure);
if (LOG.isDebugEnabled())
LOG.debug("replacing current content with null succeeded");
return false;
}
}
}
@Override
public String toString()
{
return getClass().getSimpleName() + "@" + hashCode() + " _content=" + _content;
}
}
@Override
public boolean needContent()
{
boolean hasContent = _contentDemander.demand(true);
if (LOG.isDebugEnabled())
LOG.debug("needContent has content? {}", hasContent);
return hasContent;
}
@Override
public HttpInput.Content produceContent()
{
HttpInput.Content content = null;
if (_contentDemander.demand(false))
content = _contentDemander.poll();
if (LOG.isDebugEnabled())
LOG.debug("produceContent produced {}", content);
return content;
}
@Override
public boolean failAllContent(Throwable failure)
{
if (LOG.isDebugEnabled())
LOG.debug("failing all content with {}", (Object)failure);
boolean atEof = getStream().failAllData(failure);
atEof |= _contentDemander.failContent(failure);
if (LOG.isDebugEnabled())
LOG.debug("failed all content, reached EOF? {}", atEof);
return atEof;
}
@Override
public boolean failed(Throwable x)
{
if (LOG.isDebugEnabled())
LOG.debug("failed " + x);
_contentDemander.onContent(new HttpInput.ErrorContent(x));
return getRequest().getHttpInput().onContentProducible();
}
@Override
protected boolean eof()
{
_contentDemander.eof();
return false;
}
@Override
public Runnable onTrailer(HeadersFrame frame)
{
@ -301,7 +632,10 @@ public class HttpChannelOverHTTP2 extends HttpChannel implements Closeable, Writ
System.lineSeparator(), trailers);
}
// This will generate EOF -> need to call onContentProducible.
boolean handle = onRequestComplete();
boolean woken = getRequest().getHttpInput().onContentProducible();
handle |= woken;
boolean wasDelayed = _delayedUntilContent;
_delayedUntilContent = false;
@ -320,25 +654,30 @@ public class HttpChannelOverHTTP2 extends HttpChannel implements Closeable, Writ
final boolean delayed = _delayedUntilContent;
_delayedUntilContent = false;
boolean result = isIdle();
if (result)
boolean reset = isIdle();
if (reset)
consumeInput();
getHttpTransport().onStreamTimeout(failure);
if (getRequest().getHttpInput().onIdleTimeout(failure) || delayed)
failure.addSuppressed(new Throwable("HttpInput idle timeout"));
_contentDemander.onTimeout(failure);
boolean needed = getRequest().getHttpInput().onContentProducible();
if (needed || delayed)
{
consumer.accept(this::handleWithContext);
result = false;
reset = false;
}
return result;
return reset;
}
@Override
public Runnable onFailure(Throwable failure, Callback callback)
{
getHttpTransport().onStreamFailure(failure);
boolean handle = getRequest().getHttpInput().failed(failure);
boolean handle = failed(failure);
consumeInput();
return new FailureTask(failure, callback, handle);
}

61
jetty-security/src/test/java/org/eclipse/jetty/security/authentication/SpnegoAuthenticatorTest.java
View File

@ -30,6 +30,7 @@ import org.eclipse.jetty.server.Authentication;
import org.eclipse.jetty.server.HttpChannel;
import org.eclipse.jetty.server.HttpChannelState;
import org.eclipse.jetty.server.HttpConfiguration;
import org.eclipse.jetty.server.HttpInput;
import org.eclipse.jetty.server.HttpOutput;
import org.eclipse.jetty.server.Request;
import org.eclipse.jetty.server.Response;
@ -62,6 +63,36 @@ public class SpnegoAuthenticatorTest
return null;
}
@Override
public boolean failed(Throwable x)
{
return false;
}
@Override
protected boolean eof()
{
return false;
}
@Override
public boolean needContent()
{
return false;
}
@Override
public HttpInput.Content produceContent()
{
return null;
}
@Override
public boolean failAllContent(Throwable failure)
{
return false;
}
@Override
protected HttpOutput newHttpOutput()
{
@ -97,6 +128,36 @@ public class SpnegoAuthenticatorTest
return null;
}
@Override
public boolean failed(Throwable x)
{
return false;
}
@Override
protected boolean eof()
{
return false;
}
@Override
public boolean needContent()
{
return false;
}
@Override
public HttpInput.Content produceContent()
{
return null;
}
@Override
public boolean failAllContent(Throwable failure)
{
return false;
}
@Override
protected HttpOutput newHttpOutput()
{

354
jetty-server/src/main/java/org/eclipse/jetty/server/AsyncContentProducer.java Normal file
View File

@ -0,0 +1,354 @@
//
// ========================================================================
// Copyright (c) 1995-2020 Mort Bay Consulting Pty Ltd and others.
//
// This program and the accompanying materials are made available under
// the terms of the Eclipse Public License 2.0 which is available at
// https://www.eclipse.org/legal/epl-2.0
//
// This Source Code may also be made available under the following
// Secondary Licenses when the conditions for such availability set
// forth in the Eclipse Public License, v. 2.0 are satisfied:
// the Apache License v2.0 which is available at
// https://www.apache.org/licenses/LICENSE-2.0
//
// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
// ========================================================================
//
package org.eclipse.jetty.server;
import java.util.concurrent.TimeUnit;
import org.eclipse.jetty.http.BadMessageException;
import org.eclipse.jetty.http.HttpStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Non-blocking {@link ContentProducer} implementation. Calling {@link #nextContent()} will never block
* but will return null when there is no available content.
*/
class AsyncContentProducer implements ContentProducer
{
private static final Logger LOG = LoggerFactory.getLogger(AsyncContentProducer.class);
private final HttpChannel _httpChannel;
private HttpInput.Interceptor _interceptor;
private HttpInput.Content _rawContent;
private HttpInput.Content _transformedContent;
private boolean _error;
private long _firstByteTimeStamp = Long.MIN_VALUE;
private long _rawContentArrived;
AsyncContentProducer(HttpChannel httpChannel)
{
_httpChannel = httpChannel;
}
@Override
public void recycle()
{
if (LOG.isDebugEnabled())
LOG.debug("recycling {}", this);
_interceptor = null;
_rawContent = null;
_transformedContent = null;
_error = false;
_firstByteTimeStamp = Long.MIN_VALUE;
_rawContentArrived = 0L;
}
@Override
public HttpInput.Interceptor getInterceptor()
{
return _interceptor;
}
@Override
public void setInterceptor(HttpInput.Interceptor interceptor)
{
this._interceptor = interceptor;
}
@Override
public int available()
{
HttpInput.Content content = nextTransformedContent();
int available = content == null ? 0 : content.remaining();
if (LOG.isDebugEnabled())
LOG.debug("available = {}", available);
return available;
}
@Override
public boolean hasContent()
{
boolean hasContent = _rawContent != null;
if (LOG.isDebugEnabled())
LOG.debug("hasContent = {}", hasContent);
return hasContent;
}
@Override
public boolean isError()
{
if (LOG.isDebugEnabled())
LOG.debug("isError = {}", _error);
return _error;
}
@Override
public void checkMinDataRate()
{
long minRequestDataRate = _httpChannel.getHttpConfiguration().getMinRequestDataRate();
if (LOG.isDebugEnabled())
LOG.debug("checkMinDataRate [m={},t={}]", minRequestDataRate, _firstByteTimeStamp);
if (minRequestDataRate > 0 && _firstByteTimeStamp != Long.MIN_VALUE)
{
long period = System.nanoTime() - _firstByteTimeStamp;
if (period > 0)
{
long minimumData = minRequestDataRate * TimeUnit.NANOSECONDS.toMillis(period) / TimeUnit.SECONDS.toMillis(1);
if (getRawContentArrived() < minimumData)
{
if (LOG.isDebugEnabled())
LOG.debug("checkMinDataRate check failed");
BadMessageException bad = new BadMessageException(HttpStatus.REQUEST_TIMEOUT_408,
String.format("Request content data rate < %d B/s", minRequestDataRate));
if (_httpChannel.getState().isResponseCommitted())
{
if (LOG.isDebugEnabled())
LOG.debug("checkMinDataRate aborting channel");
_httpChannel.abort(bad);
}
failCurrentContent(bad);
throw bad;
}
}
}
}
@Override
public long getRawContentArrived()
{
if (LOG.isDebugEnabled())
LOG.debug("getRawContentArrived = {}", _rawContentArrived);
return _rawContentArrived;
}
@Override
public boolean consumeAll(Throwable x)
{
if (LOG.isDebugEnabled())
LOG.debug("consumeAll [e={}]", (Object)x);
failCurrentContent(x);
// A specific HttpChannel mechanism must be used as the following code
// does not guarantee that the channel will synchronously deliver all
// content it already contains:
// while (true)
// {
// HttpInput.Content content = _httpChannel.produceContent();
// ...
// }
// as the HttpChannel's produceContent() contract makes no such promise;
// for instance the H2 implementation calls Stream.demand() that may
// deliver the content asynchronously. Tests in StreamResetTest cover this.
boolean atEof = _httpChannel.failAllContent(x);
if (LOG.isDebugEnabled())
LOG.debug("failed all content of http channel; at EOF? {}", atEof);
return atEof;
}
private void failCurrentContent(Throwable x)
{
if (LOG.isDebugEnabled())
LOG.debug("failing currently held content [r={},t={}]", _rawContent, _transformedContent, x);
if (_transformedContent != null && !_transformedContent.isSpecial())
{
if (_transformedContent != _rawContent)
{
_transformedContent.skip(_transformedContent.remaining());
_transformedContent.failed(x);
}
_transformedContent = null;
}
if (_rawContent != null && !_rawContent.isSpecial())
{
_rawContent.skip(_rawContent.remaining());
_rawContent.failed(x);
_rawContent = null;
}
}
@Override
public boolean onContentProducible()
{
if (LOG.isDebugEnabled())
LOG.debug("onContentProducible");
return _httpChannel.getState().onReadReady();
}
@Override
public HttpInput.Content nextContent()
{
HttpInput.Content content = nextTransformedContent();
if (LOG.isDebugEnabled())
LOG.debug("nextContent = {}", content);
if (content != null)
_httpChannel.getState().onReadIdle();
return content;
}
@Override
public void reclaim(HttpInput.Content content)
{
if (LOG.isDebugEnabled())
LOG.debug("reclaim {} [t={}]", content, _transformedContent);
if (_transformedContent == content)
{
content.succeeded();
if (_transformedContent == _rawContent)
_rawContent = null;
_transformedContent = null;
}
}
@Override
public boolean isReady()
{
HttpInput.Content content = nextTransformedContent();
if (content == null)
{
_httpChannel.getState().onReadUnready();
if (_httpChannel.needContent())
{
content = nextTransformedContent();
if (LOG.isDebugEnabled())
LOG.debug("isReady got transformed content after needContent retry {}", content);
if (content != null)
_httpChannel.getState().onContentAdded();
}
else
{
if (LOG.isDebugEnabled())
LOG.debug("isReady has no transformed content after needContent");
}
}
else
{
if (LOG.isDebugEnabled())
LOG.debug("isReady got transformed content {}", content);
_httpChannel.getState().onContentAdded();
}
boolean ready = content != null;
if (LOG.isDebugEnabled())
LOG.debug("isReady = {}", ready);
return ready;
}
private HttpInput.Content nextTransformedContent()
{
if (LOG.isDebugEnabled())
LOG.debug("nextTransformedContent [r={},t={}]", _rawContent, _transformedContent);
if (_rawContent == null)
{
_rawContent = produceRawContent();
if (_rawContent == null)
return null;
}
if (_transformedContent != null && _transformedContent.isEmpty())
{
if (_transformedContent != _rawContent)
_transformedContent.succeeded();
if (LOG.isDebugEnabled())
LOG.debug("nulling depleted transformed content");
_transformedContent = null;
}
while (_transformedContent == null)
{
if (_rawContent.isSpecial())
{
// TODO does EOF need to be passed to the interceptors?
_error = _rawContent.getError() != null;
if (LOG.isDebugEnabled())
LOG.debug("raw content is special (with error = {}), returning it", _error);
return _rawContent;
}
if (_interceptor != null)
{
if (LOG.isDebugEnabled())
LOG.debug("using interceptor {} to transform raw content", _interceptor);
_transformedContent = _interceptor.readFrom(_rawContent);
}
else
{
if (LOG.isDebugEnabled())
LOG.debug("null interceptor, transformed content = raw content");
_transformedContent = _rawContent;
}
if (_transformedContent != null && _transformedContent.isEmpty())
{
if (_transformedContent != _rawContent)
_transformedContent.succeeded();
if (LOG.isDebugEnabled())
LOG.debug("nulling depleted transformed content");
_transformedContent = null;
}
if (_transformedContent == null)
{
if (_rawContent.isEmpty())
{
_rawContent.succeeded();
_rawContent = null;
if (LOG.isDebugEnabled())
LOG.debug("nulling depleted raw content");
_rawContent = produceRawContent();
if (_rawContent == null)
{
if (LOG.isDebugEnabled())
LOG.debug("produced null raw content, returning null");
return null;
}
}
else
{
if (LOG.isDebugEnabled())
LOG.debug("raw content is not empty");
}
}
else
{
if (LOG.isDebugEnabled())
LOG.debug("transformed content is not empty");
}
}
if (LOG.isDebugEnabled())
LOG.debug("returning transformed content {}", _transformedContent);
return _transformedContent;
}
private HttpInput.Content produceRawContent()
{
HttpInput.Content content = _httpChannel.produceContent();
if (content != null)
{
_rawContentArrived += content.remaining();
if (_firstByteTimeStamp == Long.MIN_VALUE)
_firstByteTimeStamp = System.nanoTime();
if (LOG.isDebugEnabled())
LOG.debug("produceRawContent updated rawContentArrived to {} and firstByteTimeStamp to {}", _rawContentArrived, _firstByteTimeStamp);
}
if (LOG.isDebugEnabled())
LOG.debug("produceRawContent produced {}", content);
return content;
}
}

164
jetty-server/src/main/java/org/eclipse/jetty/server/BlockingContentProducer.java Normal file
View File

@ -0,0 +1,164 @@
//
// ========================================================================
// Copyright (c) 1995-2020 Mort Bay Consulting Pty Ltd and others.
//
// This program and the accompanying materials are made available under
// the terms of the Eclipse Public License 2.0 which is available at
// https://www.eclipse.org/legal/epl-2.0
//
// This Source Code may also be made available under the following
// Secondary Licenses when the conditions for such availability set
// forth in the Eclipse Public License, v. 2.0 are satisfied:
// the Apache License v2.0 which is available at
// https://www.apache.org/licenses/LICENSE-2.0
//
// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
// ========================================================================
//
package org.eclipse.jetty.server;
import java.util.concurrent.Semaphore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Blocking implementation of {@link ContentProducer}. Calling {@link #nextContent()} will block when
* there is no available content but will never return null.
*/
class BlockingContentProducer implements ContentProducer
{
private static final Logger LOG = LoggerFactory.getLogger(BlockingContentProducer.class);
private final Semaphore _semaphore = new Semaphore(0);
private final AsyncContentProducer _asyncContentProducer;
BlockingContentProducer(AsyncContentProducer delegate)
{
_asyncContentProducer = delegate;
}
@Override
public void recycle()
{
if (LOG.isDebugEnabled())
LOG.debug("recycling {}", this);
_asyncContentProducer.recycle();
_semaphore.drainPermits();
}
@Override
public int available()
{
return _asyncContentProducer.available();
}
@Override
public boolean hasContent()
{
return _asyncContentProducer.hasContent();
}
@Override
public boolean isError()
{
return _asyncContentProducer.isError();
}
@Override
public void checkMinDataRate()
{
_asyncContentProducer.checkMinDataRate();
}
@Override
public long getRawContentArrived()
{
return _asyncContentProducer.getRawContentArrived();
}
@Override
public boolean consumeAll(Throwable x)
{
return _asyncContentProducer.consumeAll(x);
}
@Override
public HttpInput.Content nextContent()
{
while (true)
{
HttpInput.Content content = _asyncContentProducer.nextContent();
if (LOG.isDebugEnabled())
LOG.debug("nextContent async producer returned {}", content);
if (content != null)
return content;
// IFF isReady() returns false then HttpChannel.needContent() has been called,
// thus we know that eventually a call to onContentProducible will come.
if (_asyncContentProducer.isReady())
{
if (LOG.isDebugEnabled())
LOG.debug("nextContent async producer is ready, retrying");
continue;
}
if (LOG.isDebugEnabled())
LOG.debug("nextContent async producer is not ready, waiting on semaphore {}", _semaphore);
try
{
_semaphore.acquire();
}
catch (InterruptedException e)
{
return new HttpInput.ErrorContent(e);
}
}
}
@Override
public void reclaim(HttpInput.Content content)
{
_asyncContentProducer.reclaim(content);
}
@Override
public boolean isReady()
{
boolean ready = available() > 0;
if (LOG.isDebugEnabled())
LOG.debug("isReady = {}", ready);
return ready;
}
@Override
public HttpInput.Interceptor getInterceptor()
{
return _asyncContentProducer.getInterceptor();
}
@Override
public void setInterceptor(HttpInput.Interceptor interceptor)
{
_asyncContentProducer.setInterceptor(interceptor);
}
@Override
public boolean onContentProducible()
{
// In blocking mode, the dispatched thread normally does not have to be rescheduled as it is normally in state
// DISPATCHED blocked on the semaphore that just needs to be released for the dispatched thread to resume. This is why
// this method always returns false.
// But async errors can occur while the dispatched thread is NOT blocked reading (i.e.: in state WAITING),
// so the WAITING to WOKEN transition must be done by the error-notifying thread which then has to reschedule the
// dispatched thread after HttpChannelState.asyncError() is called.
// Calling _asyncContentProducer.wakeup() changes the channel state from WAITING to WOKEN which would prevent the
// subsequent call to HttpChannelState.asyncError() from rescheduling the thread.
// AsyncServletTest.testStartAsyncThenClientStreamIdleTimeout() tests this.
if (LOG.isDebugEnabled())
LOG.debug("onContentProducible releasing semaphore {}", _semaphore);
_semaphore.release();
return false;
}
}

141
jetty-server/src/main/java/org/eclipse/jetty/server/ContentProducer.java Normal file
View File

@ -0,0 +1,141 @@
//
// ========================================================================
// Copyright (c) 1995-2020 Mort Bay Consulting Pty Ltd and others.
//
// This program and the accompanying materials are made available under
// the terms of the Eclipse Public License 2.0 which is available at
// https://www.eclipse.org/legal/epl-2.0
//
// This Source Code may also be made available under the following
// Secondary Licenses when the conditions for such availability set
// forth in the Eclipse Public License, v. 2.0 are satisfied:
// the Apache License v2.0 which is available at
// https://www.apache.org/licenses/LICENSE-2.0
//
// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
// ========================================================================
//
package org.eclipse.jetty.server;
/**
* ContentProducer is the bridge between {@link HttpInput} and {@link HttpChannel}.
* It wraps a {@link HttpChannel} and uses the {@link HttpChannel#needContent()},
* {@link HttpChannel#produceContent()} and {@link HttpChannel#failAllContent(Throwable)}
* methods, tracks the current state of the channel's input by updating the
* {@link HttpChannelState} and provides the necessary mechanism to unblock
* the reader thread when using a blocking implementation or to know if the reader thread
* has to be rescheduled when using an async implementation.
*/
public interface ContentProducer
{
/**
* Reset all internal state and clear any held resources.
*/
void recycle();
/**
* Fail all content currently available in this {@link ContentProducer} instance
* as well as in the underlying {@link HttpChannel}.
*
* This call is always non-blocking.
* Doesn't change state.
* @return true if EOF was reached.
*/
boolean consumeAll(Throwable x);
/**
* Check if the current data rate consumption is above the minimal rate.
* Abort the channel, fail the content currently available and throw a
* BadMessageException(REQUEST_TIMEOUT_408) if the check fails.
*/
void checkMinDataRate();
/**
* Get the byte count produced by the underlying {@link HttpChannel}.
*
* This call is always non-blocking.
* Doesn't change state.
* @return the byte count produced by the underlying {@link HttpChannel}.
*/
long getRawContentArrived();
/**
* Get the byte count that can immediately be read from this
* {@link ContentProducer} instance or the underlying {@link HttpChannel}.
*
* This call is always non-blocking.
* Doesn't change state.
* @return the available byte count.
*/
int available();
/**
* Check if this {@link ContentProducer} instance contains some
* content without querying the underlying {@link HttpChannel}.
*
* This call is always non-blocking.
* Doesn't change state.
* Doesn't query the HttpChannel.
* @return true if this {@link ContentProducer} instance contains content, false otherwise.
*/
boolean hasContent();
/**
* Check if the underlying {@link HttpChannel} reached an error content.
* This call is always non-blocking.
* Doesn't change state.
* Doesn't query the HttpChannel.
* @return true if the underlying {@link HttpChannel} reached an error content, false otherwise.
*/
boolean isError();
/**
* Get the next content that can be read from or that describes the special condition
* that was reached (error, eof).
* This call may or may not block until some content is available, depending on the implementation.
* The returned content is decoded by the interceptor set with {@link #setInterceptor(HttpInput.Interceptor)}
* or left as-is if no intercept is set.
* After this call, state can be either of UNREADY or IDLE.
* @return the next content that can be read from or null if the implementation does not block
* and has no available content.
*/
HttpInput.Content nextContent();
/**
* Free up the content by calling {@link HttpInput.Content#succeeded()} on it
* and updating this instance' internal state.
*/
void reclaim(HttpInput.Content content);
/**
* Check if this {@link ContentProducer} instance has some content that can be read without blocking.
* If there is some, the next call to {@link #nextContent()} will not block.
* If there isn't any and the implementation does not block, this method will trigger a
* {@link javax.servlet.ReadListener} callback once some content is available.
* This call is always non-blocking.
* After this call, state can be either of UNREADY or READY.
* @return true if some content is immediately available, false otherwise.
*/
boolean isReady();
/**
* Get the {@link org.eclipse.jetty.server.HttpInput.Interceptor}.
* @return The {@link org.eclipse.jetty.server.HttpInput.Interceptor}, or null if none set.
*/
HttpInput.Interceptor getInterceptor();
/**
* Set the interceptor.
* @param interceptor The interceptor to use.
*/
void setInterceptor(HttpInput.Interceptor interceptor);
/**
* Wake up the thread that is waiting for the next content.
* After this call, state can be READY.
* @return true if the thread has to be rescheduled, false otherwise.
*/
boolean onContentProducible();
}

80
jetty-server/src/main/java/org/eclipse/jetty/server/HttpChannel.java
View File

@ -64,7 +64,7 @@ import org.slf4j.LoggerFactory;
* HttpParser.RequestHandler callbacks. The completion of the active phase is signalled by a call to
* HttpTransport.completed().
*/
public class HttpChannel implements Runnable, HttpOutput.Interceptor
public abstract class HttpChannel implements Runnable, HttpOutput.Interceptor
{
public static Listener NOOP_LISTENER = new Listener() {};
private static final Logger LOG = LoggerFactory.getLogger(HttpChannel.class);
@ -119,11 +119,53 @@ public class HttpChannel implements Runnable, HttpOutput.Interceptor
return _state.isSendError();
}
protected HttpInput newHttpInput(HttpChannelState state)
private HttpInput newHttpInput(HttpChannelState state)
{
return new HttpInput(state);
}
/**
* Notify the channel that content is needed. If some content is immediately available, true is returned and
* {@link #produceContent()} has to be called and will return a non-null object.
* If no content is immediately available, {@link HttpInput#onContentProducible()} is called once some content arrives
* and {@link #produceContent()} can be called without returning null.
* If a failure happens, then {@link HttpInput#onContentProducible()} will be called and an error content will return the
* error on the next call to {@link #produceContent()}.
* @return true if content is immediately available.
*/
public abstract boolean needContent();
/**
* Produce a {@link HttpInput.Content} object with data currently stored within the channel. The produced content
* can be special (meaning calling {@link HttpInput.Content#isSpecial()} returns true) if the channel reached a special
* state, like EOF or an error.
* Once a special content has been returned, all subsequent calls to this method will always return a special content
* of the same kind and {@link #needContent()} will always return true.
* The returned content is "raw", i.e.: not decoded.
* @return a {@link HttpInput.Content} object if one is immediately available without blocking, null otherwise.
*/
public abstract HttpInput.Content produceContent();
/**
* Fail all content that is currently stored within the channel.
* @param failure the failure to fail the content with.
* @return true if EOF was reached while failing all content, false otherwise.
*/
public abstract boolean failAllContent(Throwable failure);
/**
* Fail the channel's input.
* @param failure the failure.
* @return true if the channel needs to be rescheduled.
*/
public abstract boolean failed(Throwable failure);
/**
* Mark the channel's input as EOF.
* @return true if the channel needs to be rescheduled.
*/
protected abstract boolean eof();
protected HttpOutput newHttpOutput()
{
return new HttpOutput(this);
@ -303,19 +345,6 @@ public class HttpChannel implements Runnable, HttpOutput.Interceptor
_transientListeners.clear();
}
public void onAsyncWaitForContent()
{
}
public void onBlockWaitForContent()
{
}
public void onBlockWaitForContentFailure(Throwable failure)
{
getRequest().getHttpInput().failed(failure);
}
@Override
public void run()
{
@ -445,18 +474,6 @@ public class HttpChannel implements Runnable, HttpOutput.Interceptor
throw _state.getAsyncContextEvent().getThrowable();
}
case READ_REGISTER:
{
onAsyncWaitForContent();
break;
}
case READ_PRODUCE:
{
_request.getHttpInput().asyncReadProduce();
break;
}
case READ_CALLBACK:
{
ContextHandler handler = _state.getContextHandler();
@ -706,7 +723,7 @@ public class HttpChannel implements Runnable, HttpOutput.Interceptor
if (LOG.isDebugEnabled())
LOG.debug("onContent {} {}", this, content);
_combinedListener.onRequestContent(_request, content.getByteBuffer());
return _request.getHttpInput().addContent(content);
return false;
}
public boolean onContentComplete()
@ -729,7 +746,7 @@ public class HttpChannel implements Runnable, HttpOutput.Interceptor
{
if (LOG.isDebugEnabled())
LOG.debug("onRequestComplete {}", this);
boolean result = _request.getHttpInput().eof();
boolean result = eof();
_combinedListener.onRequestEnd(_request);
return result;
}
@ -765,11 +782,6 @@ public class HttpChannel implements Runnable, HttpOutput.Interceptor
_transport.onCompleted();
}
public boolean onEarlyEOF()
{
return _request.getHttpInput().earlyEOF();
}
public void onBadMessage(BadMessageException failure)
{
int status = failure.getCode();

203
jetty-server/src/main/java/org/eclipse/jetty/server/HttpChannelOverHttp.java
View File

@ -40,6 +40,7 @@ import org.eclipse.jetty.http.HttpVersion;
import org.eclipse.jetty.http.MetaData;
import org.eclipse.jetty.io.Connection;
import org.eclipse.jetty.io.EndPoint;
import org.eclipse.jetty.io.EofException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -50,6 +51,7 @@ public class HttpChannelOverHttp extends HttpChannel implements HttpParser.Reque
{
private static final Logger LOG = LoggerFactory.getLogger(HttpChannelOverHttp.class);
private static final HttpField PREAMBLE_UPGRADE_H2C = new HttpField(HttpHeader.UPGRADE, "h2c");
private static final HttpInput.Content EOF = new HttpInput.EofContent();
private final HttpConnection _httpConnection;
private final RequestBuilder _requestBuilder = new RequestBuilder();
private MetaData.Request _metadata;
@ -61,6 +63,14 @@ public class HttpChannelOverHttp extends HttpChannel implements HttpParser.Reque
private boolean _expect102Processing = false;
private List<String> _complianceViolations;
private HttpFields.Mutable _trailers;
// Field _content doesn't need to be volatile nor protected by a lock
// as it is always accessed by the same thread, i.e.: we get notified by onFillable
// that the socket contains new bytes and either schedule an onDataAvailable
// call that is going to read the socket or release the blocking semaphore to wake up
// the blocked reader and make it read the socket. The same logic is true for async
// events like timeout: we get notified and either schedule onError or release the
// blocking semaphore.
private HttpInput.Content _content;
public HttpChannelOverHttp(HttpConnection httpConnection, Connector connector, HttpConfiguration config, EndPoint endPoint, HttpTransport transport)
{
@ -75,6 +85,79 @@ public class HttpChannelOverHttp extends HttpChannel implements HttpParser.Reque
_httpConnection.getGenerator().setPersistent(false);
}
@Override
public boolean needContent()
{
if (_content != null)
{
if (LOG.isDebugEnabled())
LOG.debug("needContent has content immediately available: {}", _content);
return true;
}
_httpConnection.parseAndFillForContent();
if (_content != null)
{
if (LOG.isDebugEnabled())
LOG.debug("needContent has content after parseAndFillForContent: {}", _content);
return true;
}
if (LOG.isDebugEnabled())
LOG.debug("needContent has no content");
_httpConnection.asyncReadFillInterested();
return false;
}
@Override
public HttpInput.Content produceContent()
{
if (_content == null)
{
if (LOG.isDebugEnabled())
LOG.debug("produceContent has no content, parsing and filling");
_httpConnection.parseAndFillForContent();
}
HttpInput.Content result = _content;
if (result != null && !result.isSpecial())
_content = result.isEof() ? EOF : null;
if (LOG.isDebugEnabled())
LOG.debug("produceContent produced {}", result);
return result;
}
@Override
public boolean failAllContent(Throwable failure)
{
if (LOG.isDebugEnabled())
LOG.debug("failing all content with {}", (Object)failure);
if (_content != null && !_content.isSpecial())
{
_content.failed(failure);
_content = _content.isEof() ? EOF : null;
if (_content == EOF)
return true;
}
while (true)
{
HttpInput.Content c = produceContent();
if (c == null)
{
if (LOG.isDebugEnabled())
LOG.debug("failed all content, EOF was not reached");
return false;
}
c.skip(c.remaining());
c.failed(failure);
if (c.isSpecial())
{
boolean atEof = c.isEof();
if (LOG.isDebugEnabled())
LOG.debug("failed all content, EOF = {}", atEof);
return atEof;
}
}
}
@Override
public void badMessage(BadMessageException failure)
{
@ -85,7 +168,7 @@ public class HttpChannelOverHttp extends HttpChannel implements HttpParser.Reque
if (_metadata == null)
_metadata = _requestBuilder.build();
onRequest(_metadata);
getRequest().getHttpInput().earlyEOF();
markEarlyEOF();
}
catch (Exception e)
{
@ -96,12 +179,23 @@ public class HttpChannelOverHttp extends HttpChannel implements HttpParser.Reque
}
@Override
public boolean content(ByteBuffer content)
public boolean content(ByteBuffer buffer)
{
HttpInput.Content c = _httpConnection.newContent(content);
boolean handle = onContent(c) || _delayedForContent;
HttpInput.Content content = _httpConnection.newContent(buffer);
if (_content != null)
{
if (_content.isSpecial())
content.failed(_content.getError());
else
throw new AssertionError("Cannot overwrite exiting content " + _content + " with " + content);
}
else
{
_content = content;
onContent(_content);
_delayedForContent = false;
return handle;
}
return true;
}
@Override
@ -147,13 +241,70 @@ public class HttpChannelOverHttp extends HttpChannel implements HttpParser.Reque
_httpConnection.getGenerator().setPersistent(false);
// If we have no request yet, just close
if (_metadata == null)
{
_httpConnection.close();
else if (onEarlyEOF() || _delayedForContent)
}
else
{
markEarlyEOF();
if (_delayedForContent)
{
_delayedForContent = false;
handle();
}
}
}
private void markEarlyEOF()
{
if (LOG.isDebugEnabled())
LOG.debug("received early EOF, content = {}", _content);
EofException failure = new EofException("Early EOF");
if (_content != null)
_content.failed(failure);
_content = new HttpInput.ErrorContent(failure);
}
@Override
protected boolean eof()
{
if (LOG.isDebugEnabled())
LOG.debug("received EOF, content = {}", _content);
if (_content == null)
{
_content = EOF;
}
else
{
HttpInput.Content c = _content;
_content = new HttpInput.WrappingContent(c, true);
}
return false;
}
@Override
public boolean failed(Throwable x)
{
if (LOG.isDebugEnabled())
LOG.debug("failed {}, content = {}", x, _content);
Throwable error = null;
if (_content != null && _content.isSpecial())
error = _content.getError();
if (error != null && error != x)
{
error.addSuppressed(x);
}
else
{
if (_content != null)
_content.failed(x);
_content = new HttpInput.ErrorContent(x);
}
return getRequest().getHttpInput().onContentProducible();
}
@Override
public EndPoint getTunnellingEndPoint()
@ -309,24 +460,6 @@ public class HttpChannelOverHttp extends HttpChannel implements HttpParser.Reque
return onRequestComplete();
}
@Override
public void onAsyncWaitForContent()
{
_httpConnection.asyncReadFillInterested();
}
@Override
public void onBlockWaitForContent()
{
_httpConnection.blockingReadFillInterested();
}
@Override
public void onBlockWaitForContentFailure(Throwable failure)
{
_httpConnection.blockingReadFailure(failure);
}
@Override
public void onComplianceViolation(ComplianceViolation.Mode mode, ComplianceViolation violation, String details)
{
@ -434,6 +567,9 @@ public class HttpChannelOverHttp extends HttpChannel implements HttpParser.Reque
_upgrade = null;
_trailers = null;
_metadata = null;
if (_content != null && !_content.isSpecial())
throw new AssertionError("unconsumed content: " + _content);
_content = null;
}
@Override
@ -459,12 +595,6 @@ public class HttpChannelOverHttp extends HttpChannel implements HttpParser.Reque
super.handleException(x);
}
@Override
protected HttpInput newHttpInput(HttpChannelState state)
{
return new HttpInputOverHTTP(state);
}
/**
* <p>Attempts to perform an HTTP/1.1 upgrade.</p>
* <p>The upgrade looks up a {@link ConnectionFactory.Upgrading} from the connector
@ -534,13 +664,24 @@ public class HttpChannelOverHttp extends HttpChannel implements HttpParser.Reque
if (_delayedForContent)
{
_delayedForContent = false;
getRequest().getHttpInput().onIdleTimeout(timeout);
doOnIdleTimeout(timeout);
execute(this);
return false;
}
return true;
}
private void doOnIdleTimeout(Throwable x)
{
boolean neverDispatched = getState().isIdle();
boolean waitingForContent = _content == null || _content.remaining() == 0;
if ((waitingForContent || neverDispatched) && (_content == null || !_content.isSpecial()))
{
x.addSuppressed(new Throwable("HttpInput idle timeout"));
_content = new HttpInput.ErrorContent(x);
}
}
private static class RequestBuilder
{
private final HttpFields.Mutable _fieldsBuilder = HttpFields.build();

237
jetty-server/src/main/java/org/eclipse/jetty/server/HttpChannelState.java
View File

@ -108,11 +108,8 @@ public class HttpChannelState
private enum InputState
{
IDLE, // No isReady; No data
REGISTER, // isReady()==false handling; No data
REGISTERED, // isReady()==false !handling; No data
POSSIBLE, // isReady()==false async read callback called (http/1 only)
PRODUCING, // isReady()==false READ_PRODUCE action is being handled (http/1 only)
READY // isReady() was false, onContentAdded has been called
UNREADY, // isReady()==false; No data
READY // isReady() was false; data is available
}
/*
@ -137,8 +134,6 @@ public class HttpChannelState
ASYNC_ERROR, // handle an async error
ASYNC_TIMEOUT, // call asyncContext onTimeout
WRITE_CALLBACK, // handle an IO write callback
READ_REGISTER, // Register for fill interest
READ_PRODUCE, // Check is a read is possible by parsing/filling
READ_CALLBACK, // handle an IO read callback
COMPLETE, // Complete the response by closing output
TERMINATED, // No further actions
@ -465,19 +460,12 @@ public class HttpChannelState
case ASYNC:
switch (_inputState)
{
case POSSIBLE:
_inputState = InputState.PRODUCING;
return Action.READ_PRODUCE;
case IDLE:
case UNREADY:
break;
case READY:
_inputState = InputState.IDLE;
return Action.READ_CALLBACK;
case REGISTER:
case PRODUCING:
_inputState = InputState.REGISTERED;
return Action.READ_REGISTER;
case IDLE:
case REGISTERED:
break;
default:
throw new IllegalStateException(getStatusStringLocked());
@ -1222,99 +1210,8 @@ public class HttpChannelState
_channel.getRequest().setAttribute(name, attribute);
}
/**
* Called to signal async read isReady() has returned false.
* This indicates that there is no content available to be consumed
* and that once the channel enters the ASYNC_WAIT state it will
* register for read interest by calling {@link HttpChannel#onAsyncWaitForContent()}
* either from this method or from a subsequent call to {@link #unhandle()}.
*/
public void onReadUnready()
{
boolean interested = false;
try (AutoLock l = lock())
{
if (LOG.isDebugEnabled())
LOG.debug("onReadUnready {}", toStringLocked());
switch (_inputState)
{
case IDLE:
case READY:
if (_state == State.WAITING)
{
interested = true;
_inputState = InputState.REGISTERED;
}
else
{
_inputState = InputState.REGISTER;
}
break;
case REGISTER:
case REGISTERED:
case POSSIBLE:
case PRODUCING:
break;
default:
throw new IllegalStateException(toStringLocked());
}
}
if (interested)
_channel.onAsyncWaitForContent();
}
/**
* Called to signal that content is now available to read.
* If the channel is in ASYNC_WAIT state and unready (ie isReady() has
* returned false), then the state is changed to ASYNC_WOKEN and true
* is returned.
*
* @return True IFF the channel was unready and in ASYNC_WAIT state
*/
public boolean onContentAdded()
{
boolean woken = false;
try (AutoLock l = lock())
{
if (LOG.isDebugEnabled())
LOG.debug("onContentAdded {}", toStringLocked());
switch (_inputState)
{
case IDLE:
case READY:
break;
case PRODUCING:
_inputState = InputState.READY;
break;
case REGISTER:
case REGISTERED:
_inputState = InputState.READY;
if (_state == State.WAITING)
{
woken = true;
_state = State.WOKEN;
}
break;
default:
throw new IllegalStateException(toStringLocked());
}
}
return woken;
}
/**
* Called to signal that the channel is ready for a callback.
* This is similar to calling {@link #onReadUnready()} followed by
* {@link #onContentAdded()}, except that as content is already
* available, read interest is never set.
*
* @return true if woken
*/
@ -1328,7 +1225,11 @@ public class HttpChannelState
switch (_inputState)
{
case READY:
_inputState = InputState.READY;
break;
case IDLE:
case UNREADY:
_inputState = InputState.READY;
if (_state == State.WAITING)
{
@ -1344,64 +1245,102 @@ public class HttpChannelState
return woken;
}
/**
* Called to indicate that more content may be available,
* but that a handling thread may need to produce (fill/parse)
* it. Typically called by the async read success callback.
*
* @return {@code true} if more content may be available
*/
public boolean onReadPossible()
{
boolean woken = false;
try (AutoLock l = lock())
{
if (LOG.isDebugEnabled())
LOG.debug("onReadPossible {}", toStringLocked());
switch (_inputState)
{
case REGISTERED:
_inputState = InputState.POSSIBLE;
if (_state == State.WAITING)
{
woken = true;
_state = State.WOKEN;
}
break;
default:
throw new IllegalStateException(toStringLocked());
}
}
return woken;
}
/**
* Called to signal that a read has read -1.
* Will wake if the read was called while in ASYNC_WAIT state
*
* @return {@code true} if woken
*/
public boolean onReadEof()
{
boolean woken = false;
try (AutoLock l = lock())
{
if (LOG.isDebugEnabled())
LOG.debug("onEof {}", toStringLocked());
LOG.debug("onReadEof {}", toStringLocked());
// Force read ready so onAllDataRead can be called
switch (_inputState)
{
case IDLE:
case READY:
case UNREADY:
_inputState = InputState.READY;
if (_state == State.WAITING)
{
woken = true;
_state = State.WOKEN;
}
break;
default:
throw new IllegalStateException(toStringLocked());
}
}
return woken;
}
public void onContentAdded()
{
try (AutoLock l = lock())
{
if (LOG.isDebugEnabled())
LOG.debug("onContentAdded {}", toStringLocked());
switch (_inputState)
{
case IDLE:
case UNREADY:
case READY:
_inputState = InputState.READY;
break;
default:
throw new IllegalStateException(toStringLocked());
}
}
}
public void onReadIdle()
{
try (AutoLock l = lock())
{
if (LOG.isDebugEnabled())
LOG.debug("onReadIdle {}", toStringLocked());
switch (_inputState)
{
case UNREADY:
case READY:
case IDLE:
_inputState = InputState.IDLE;
break;
default:
throw new IllegalStateException(toStringLocked());
}
}
}
/**
* Called to indicate that more content may be available,
* but that a handling thread may need to produce (fill/parse)
* it. Typically called by the async read success callback.
*/
public void onReadUnready()
{
try (AutoLock l = lock())
{
if (LOG.isDebugEnabled())
LOG.debug("onReadUnready {}", toStringLocked());
switch (_inputState)
{
case IDLE:
case UNREADY:
case READY: // READY->UNREADY is needed by AsyncServletIOTest.testStolenAsyncRead
_inputState = InputState.UNREADY;
break;
default:
throw new IllegalStateException(toStringLocked());
}
}
}
public boolean onWritePossible()
{
boolean wake = false;

84
jetty-server/src/main/java/org/eclipse/jetty/server/HttpChannelState_input.puml Normal file
View File

@ -0,0 +1,84 @@
@startuml
title HttpChannelState
note top of onReadReady_inputState: onReadReady
state "input state" as onReadReady_inputState {
state "IDLE" as onReadReady_IDLE
state "UNREADY" as onReadReady_UNREADY
state "READY" as onReadReady_READY
state "channel state" as onReadReady_channelState {
state "WAITING" as onReadReady_WAITING
state "WOKEN" as onReadReady_WOKEN
onReadReady_WAITING --> onReadReady_WOKEN
}
onReadReady_IDLE --> onReadReady_channelState
onReadReady_UNREADY --> onReadReady_channelState
onReadReady_channelState --> onReadReady_READY
onReadReady_READY --> onReadReady_READY
}
note top of onReadEof_inputState: onReadEof
state "input state" as onReadEof_inputState {
state "IDLE" as onReadEof_IDLE
state "UNREADY" as onReadEof_UNREADY
state "READY" as onReadEof_READY
state "channel state" as onReadEof_channelState {
state "WAITING" as onReadEof_WAITING
state "WOKEN" as onReadEof_WOKEN
onReadEof_WAITING --> onReadEof_WOKEN
}
onReadEof_IDLE --> onReadEof_channelState
onReadEof_UNREADY --> onReadEof_channelState
onReadEof_READY --> onReadEof_channelState
onReadEof_channelState --> onReadEof_READY
}
note top of onReadIdle_inputState: onReadIdle
state "input state" as onReadIdle_inputState {
state "IDLE" as onReadIdle_IDLE
state "UNREADY" as onReadIdle_UNREADY
state "READY" as onReadIdle_READY
onReadIdle_IDLE --> onReadIdle_IDLE
onReadIdle_UNREADY --> onReadIdle_IDLE
onReadIdle_READY --> onReadIdle_IDLE
}
note top of onReadUnready_inputState: onReadUnready
state "input state" as onReadUnready_inputState {
state "IDLE" as onReadUnready_IDLE
state "UNREADY" as onReadUnready_UNREADY
state "READY" as onReadUnready_READY
onReadUnready_IDLE --> onReadUnready_UNREADY
onReadUnready_UNREADY --> onReadUnready_UNREADY
onReadUnready_READY --> onReadUnready_UNREADY
}
note top of onContentAdded_inputState: onContentAdded
state "input state" as onContentAdded_inputState {
state "IDLE" as onContentAdded_IDLE
state "UNREADY" as onContentAdded_UNREADY
state "READY" as onContentAdded_READY
onContentAdded_IDLE --> onContentAdded_READY
onContentAdded_UNREADY --> onContentAdded_READY
onContentAdded_READY --> onContentAdded_READY
}
@enduml

84
jetty-server/src/main/java/org/eclipse/jetty/server/HttpConnection.java
View File

@ -33,7 +33,6 @@ import org.eclipse.jetty.http.HttpHeader;
import org.eclipse.jetty.http.HttpHeaderValue;
import org.eclipse.jetty.http.HttpMethod;
import org.eclipse.jetty.http.HttpParser;
import org.eclipse.jetty.http.HttpParser.RequestHandler;
import org.eclipse.jetty.http.MetaData;
import org.eclipse.jetty.http.PreEncodedHttpField;
import org.eclipse.jetty.io.AbstractConnection;
@ -68,7 +67,6 @@ public class HttpConnection extends AbstractConnection implements Runnable, Http
private final HttpParser _parser;
private final AtomicInteger _contentBufferReferences = new AtomicInteger();
private volatile ByteBuffer _requestBuffer = null;
private final BlockingReadCallback _blockingReadCallback = new BlockingReadCallback();
private final AsyncReadCallback _asyncReadCallback = new AsyncReadCallback();
private final SendCallback _sendCallback = new SendCallback();
private final boolean _recordHttpComplianceViolations;
@ -316,21 +314,20 @@ public class HttpConnection extends AbstractConnection implements Runnable, Http
}
/**
* Fill and parse data looking for content
*
* @return true if an {@link RequestHandler} method was called and it returned true;
* Parse and fill data, looking for content
*/
protected boolean fillAndParseForContent()
void parseAndFillForContent()
{
boolean handled = false;
// When fillRequestBuffer() is called, it must always be followed by a parseRequestBuffer() call otherwise this method
// doesn't trigger EOF/earlyEOF which breaks AsyncRequestReadTest.testPartialReadThenShutdown()
int filled = Integer.MAX_VALUE;
while (_parser.inContentState())
{
int filled = fillRequestBuffer();
handled = parseRequestBuffer();
if (handled || filled <= 0 || _input.hasContent())
boolean handled = parseRequestBuffer();
if (handled || filled <= 0)
break;
filled = fillRequestBuffer();
}
return handled;
}
private int fillRequestBuffer()
@ -600,25 +597,7 @@ public class HttpConnection extends AbstractConnection implements Runnable, Http
public void asyncReadFillInterested()
{
getEndPoint().fillInterested(_asyncReadCallback);
}
public void blockingReadFillInterested()
{
// We try fillInterested here because of SSL and
// spurious wakeups. With blocking reads, we read in a loop
// that tries to read/parse content and blocks waiting if there is
// none available. The loop can be woken up by incoming encrypted
// bytes, which due to SSL might not produce any decrypted bytes.
// Thus the loop needs to register fill interest again. However if
// the loop is woken up spuriously, then the register interest again
// can result in a pending read exception, unless we use tryFillInterested.
getEndPoint().tryFillInterested(_blockingReadCallback);
}
public void blockingReadFailure(Throwable e)
{
_blockingReadCallback.failed(e);
getEndPoint().tryFillInterested(_asyncReadCallback);
}
@Override
@ -655,8 +634,15 @@ public class HttpConnection extends AbstractConnection implements Runnable, Http
@Override
public void succeeded()
{
if (_contentBufferReferences.decrementAndGet() == 0)
int counter = _contentBufferReferences.decrementAndGet();
if (counter == 0)
releaseRequestBuffer();
// TODO: this should do something (warn? fail?) if _contentBufferReferences goes below 0
if (counter < 0)
{
LOG.warn("Content reference counting went below zero: {}", counter);
_contentBufferReferences.incrementAndGet();
}
}
@Override
@ -666,44 +652,30 @@ public class HttpConnection extends AbstractConnection implements Runnable, Http
}
}
private class BlockingReadCallback implements Callback
{
@Override
public void succeeded()
{
_input.unblock();
}
@Override
public void failed(Throwable x)
{
_input.failed(x);
}
@Override
public InvocationType getInvocationType()
{
// This callback does not block, rather it wakes up the
// thread that is blocked waiting on the read.
return InvocationType.NON_BLOCKING;
}
}
private class AsyncReadCallback implements Callback
{
@Override
public void succeeded()
{
if (_channel.getState().onReadPossible())
if (_channel.getRequest().getHttpInput().onContentProducible())
_channel.handle();
}
@Override
public void failed(Throwable x)
{
if (_input.failed(x))
if (_channel.failed(x))
_channel.handle();
}
@Override
public InvocationType getInvocationType()
{
// This callback does not block when the HttpInput is in blocking mode,
// rather it wakes up the thread that is blocked waiting on the read;
// but it can if it is in async mode, hence the varying InvocationType.
return _channel.getRequest().getHttpInput().isAsync() ? InvocationType.BLOCKING : InvocationType.NON_BLOCKING;
}
}
private class SendCallback extends IteratingCallback

1492
jetty-server/src/main/java/org/eclipse/jetty/server/HttpInput.java
View File

File diff suppressed because it is too large Load Diff

35
jetty-server/src/main/java/org/eclipse/jetty/server/HttpInputOverHTTP.java
View File

@ -1,35 +0,0 @@
//
// ========================================================================
// Copyright (c) 1995-2020 Mort Bay Consulting Pty Ltd and others.
//
// This program and the accompanying materials are made available under
// the terms of the Eclipse Public License 2.0 which is available at
// https://www.eclipse.org/legal/epl-2.0
//
// This Source Code may also be made available under the following
// Secondary Licenses when the conditions for such availability set
// forth in the Eclipse Public License, v. 2.0 are satisfied:
// the Apache License v2.0 which is available at
// https://www.apache.org/licenses/LICENSE-2.0
//
// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
// ========================================================================
//
package org.eclipse.jetty.server;
import java.io.IOException;
public class HttpInputOverHTTP extends HttpInput
{
public HttpInputOverHTTP(HttpChannelState state)
{
super(state);
}
@Override
protected void produceContent() throws IOException
{
((HttpConnection)getHttpChannelState().getHttpChannel().getEndPoint().getConnection()).fillAndParseForContent();
}
}

16
jetty-server/src/main/java/org/eclipse/jetty/server/HttpInputState.puml Normal file
View File

@ -0,0 +1,16 @@
@startuml
IDLE:
READY:
UNREADY:
[*] --> IDLE
IDLE --> UNREADY : isReady
IDLE -right->READY : isReady
UNREADY -up-> READY : ASYNC onContentProducible
READY -left->IDLE : nextContent
@enduml

114
jetty-server/src/main/java/org/eclipse/jetty/server/HttpInput_async.puml Normal file
View File

@ -0,0 +1,114 @@
@startuml
title "HttpInput"
participant AsyncContentDelivery as "[async\ncontent\ndelivery]"
participant HttpChannel as "Http\nChannel\n"
participant HttpChannelState as "Http\nChannel\nState"
participant HttpInputInterceptor as "Http\nInput.\nInterceptor"
participant AsyncContentProducer as "Async\nContent\nProducer"
participant HttpInput as "Http\nInput\n"
participant Application as "\nApplication\n"
autoactivate on
== Async Read ==
Application->HttpInput: read
activate Application
HttpInput->AsyncContentProducer: nextContent
AsyncContentProducer->AsyncContentProducer: next\nTransformed\nContent
AsyncContentProducer->HttpChannel: produceContent
return raw content or null
alt if raw content is not null
AsyncContentProducer->HttpInputInterceptor: readFrom
return transformed content
end
return
alt if transformed content is not null
AsyncContentProducer->HttpChannelState: onReadIdle
return
end
return content or null
note over HttpInput
throw ISE
if content
is null
end note
HttpInput->AsyncContentProducer: reclaim
return
return
deactivate Application
== isReady ==
Application->HttpInput: isReady
activate Application
HttpInput->AsyncContentProducer: isReady
AsyncContentProducer->AsyncContentProducer: next\nTransformed\nContent
AsyncContentProducer->HttpChannel: produceContent
return raw content or null
alt if raw content is not null
AsyncContentProducer->HttpInputInterceptor: readFrom
return transformed content
end
return
alt if transformed content is not null
AsyncContentProducer->HttpChannelState: onContentAdded
return
else transformed content is null
AsyncContentProducer->HttpChannelState: onReadUnready
return
AsyncContentProducer->HttpChannel: needContent
return
alt if needContent returns true
AsyncContentProducer->AsyncContentProducer: next\nTransformed\nContent
return
alt if transformed content is not null
AsyncContentProducer->HttpChannelState: onContentAdded
return
end
end
end
return boolean\n[transformed\ncontent is not null]
return
deactivate Application
alt if content arrives
AsyncContentDelivery->HttpInput: onContentProducible
HttpInput->AsyncContentProducer: onContentProducible
alt if not at EOF
AsyncContentProducer->HttpChannelState: onReadReady
return true if woken
else if at EOF
AsyncContentProducer->HttpChannelState: onReadEof
return true if woken
end
return true if woken
return true if woken
alt onContentProducible returns true
AsyncContentDelivery->HttpChannel: execute(HttpChannel)
return
end
end
|||
== available ==
Application->HttpInput: available
activate Application
HttpInput->AsyncContentProducer: available
AsyncContentProducer->AsyncContentProducer: next\nTransformed\nContent
AsyncContentProducer->HttpChannel: produceContent
return raw content or null
alt if raw content is not null
AsyncContentProducer->HttpInputInterceptor: readFrom
return transformed content
end
return
return content size or\n0 if content is null
return
deactivate Application
|||
@enduml

64
jetty-server/src/main/java/org/eclipse/jetty/server/HttpInput_blocking.puml Normal file
View File

@ -0,0 +1,64 @@
@startuml
title "HttpInput"
participant AsyncContentDelivery as "[async\ncontent\ndelivery]"
participant HttpChannel as "Http\nChannel\n"
participant HttpChannelState as "Http\nChannel\nState"
participant AsyncContentProducer as "Async\nContent\nProducer"
participant Semaphore as "\nSemaphore\n"
participant BlockingContentProducer as "Blocking\nContent\nProducer"
participant HttpInput as "Http\nInput\n"
participant Application as "\nApplication\n"
autoactivate on
== Blocking Read ==
Application->HttpInput: read
activate Application
HttpInput->BlockingContentProducer: nextContent
loop
BlockingContentProducer->AsyncContentProducer: nextContent
AsyncContentProducer->AsyncContentProducer: nextTransformedContent
AsyncContentProducer->HttpChannel: produceContent
return
return
alt content is not null
AsyncContentProducer->HttpChannelState: onReadIdle
return
end
return content or null
alt content is null
BlockingContentProducer->HttpChannelState: onReadUnready
return
BlockingContentProducer->HttpChannel: needContent
return
alt needContent returns false
BlockingContentProducer->Semaphore: acquire
return
else needContent returns true
note over BlockingContentProducer
continue loop
end note
end
else content is not null
return non-null content
end
end
' return from BlockingContentProducer: nextContent
HttpInput->BlockingContentProducer: reclaim
BlockingContentProducer->AsyncContentProducer: reclaim
return
return
return
deactivate Application
alt if content arrives
AsyncContentDelivery->HttpInput: wakeup
HttpInput->BlockingContentProducer: wakeup
BlockingContentProducer->Semaphore: release
return
return false
return false
end
@enduml

2
jetty-server/src/main/java/org/eclipse/jetty/server/Request.java
View File

@ -749,7 +749,7 @@ public class Request implements HttpServletRequest
public long getContentRead()
{
return _input.getContentConsumed();
return _input.getContentReceived();
}
@Override

340
jetty-server/src/test/java/org/eclipse/jetty/server/AsyncContentProducerTest.java Normal file
View File

@ -0,0 +1,340 @@
//
// ========================================================================
// Copyright (c) 1995-2020 Mort Bay Consulting Pty Ltd and others.
//
// This program and the accompanying materials are made available under
// the terms of the Eclipse Public License 2.0 which is available at
// https://www.eclipse.org/legal/epl-2.0
//
// This Source Code may also be made available under the following
// Secondary Licenses when the conditions for such availability set
// forth in the Eclipse Public License, v. 2.0 are satisfied:
// the Apache License v2.0 which is available at
// https://www.apache.org/licenses/LICENSE-2.0
//
// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
// ========================================================================
//
package org.eclipse.jetty.server;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.zip.GZIPOutputStream;
import org.eclipse.jetty.io.ArrayByteBufferPool;
import org.eclipse.jetty.io.EofException;
import org.eclipse.jetty.server.handler.gzip.GzipHttpInputInterceptor;
import org.eclipse.jetty.util.compression.CompressionPool;
import org.eclipse.jetty.util.compression.InflaterPool;
import org.hamcrest.core.Is;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.notNullValue;
import static org.hamcrest.Matchers.nullValue;
public class AsyncContentProducerTest
{
private ScheduledExecutorService scheduledExecutorService;
private InflaterPool inflaterPool;
@BeforeEach
public void setUp()
{
scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
inflaterPool = new InflaterPool(-1, true);
}
@AfterEach
public void tearDown()
{
scheduledExecutorService.shutdownNow();
}
@Test
public void testAsyncContentProducerNoInterceptor() throws Exception
{
ByteBuffer[] buffers = new ByteBuffer[3];
buffers[0] = ByteBuffer.wrap("1 hello 1".getBytes(StandardCharsets.ISO_8859_1));
buffers[1] = ByteBuffer.wrap("2 howdy 2".getBytes(StandardCharsets.ISO_8859_1));
buffers[2] = ByteBuffer.wrap("3 hey ya 3".getBytes(StandardCharsets.ISO_8859_1));
final int totalContentBytesCount = countRemaining(buffers);
final String originalContentString = asString(buffers);
CyclicBarrier barrier = new CyclicBarrier(2);
ContentProducer contentProducer = new AsyncContentProducer(new ArrayDelayedHttpChannel(buffers, new HttpInput.EofContent(), scheduledExecutorService, barrier));
Throwable error = readAndAssertContent(totalContentBytesCount, originalContentString, contentProducer, (buffers.length + 1) * 2, 0, 4, barrier);
assertThat(error, nullValue());
}
@Test
public void testAsyncContentProducerNoInterceptorWithError() throws Exception
{
ByteBuffer[] buffers = new ByteBuffer[3];
buffers[0] = ByteBuffer.wrap("1 hello 1".getBytes(StandardCharsets.ISO_8859_1));
buffers[1] = ByteBuffer.wrap("2 howdy 2".getBytes(StandardCharsets.ISO_8859_1));
buffers[2] = ByteBuffer.wrap("3 hey ya 3".getBytes(StandardCharsets.ISO_8859_1));
final int totalContentBytesCount = countRemaining(buffers);
final String originalContentString = asString(buffers);
final Throwable expectedError = new EofException("Early EOF");
CyclicBarrier barrier = new CyclicBarrier(2);
ContentProducer contentProducer = new AsyncContentProducer(new ArrayDelayedHttpChannel(buffers, new HttpInput.ErrorContent(expectedError), scheduledExecutorService, barrier));
Throwable error = readAndAssertContent(totalContentBytesCount, originalContentString, contentProducer, (buffers.length + 1) * 2, 0, 4, barrier);
assertThat(error, Is.is(expectedError));
}
@Test
public void testAsyncContentProducerGzipInterceptor() throws Exception
{
ByteBuffer[] uncompressedBuffers = new ByteBuffer[3];
uncompressedBuffers[0] = ByteBuffer.wrap("1 hello 1".getBytes(StandardCharsets.ISO_8859_1));
uncompressedBuffers[1] = ByteBuffer.wrap("2 howdy 2".getBytes(StandardCharsets.ISO_8859_1));
uncompressedBuffers[2] = ByteBuffer.wrap("3 hey ya 3".getBytes(StandardCharsets.ISO_8859_1));
final int totalContentBytesCount = countRemaining(uncompressedBuffers);
final String originalContentString = asString(uncompressedBuffers);
ByteBuffer[] buffers = new ByteBuffer[3];
buffers[0] = gzipByteBuffer(uncompressedBuffers[0]);
buffers[1] = gzipByteBuffer(uncompressedBuffers[1]);
buffers[2] = gzipByteBuffer(uncompressedBuffers[2]);
CyclicBarrier barrier = new CyclicBarrier(2);
ContentProducer contentProducer = new AsyncContentProducer(new ArrayDelayedHttpChannel(buffers, new HttpInput.EofContent(), scheduledExecutorService, barrier));
contentProducer.setInterceptor(new GzipHttpInputInterceptor(inflaterPool, new ArrayByteBufferPool(1, 1, 2), 32));
Throwable error = readAndAssertContent(totalContentBytesCount, originalContentString, contentProducer, (buffers.length + 1) * 2, 0, 4, barrier);
assertThat(error, nullValue());
}
@Test
public void testAsyncContentProducerGzipInterceptorWithTinyBuffers() throws Exception
{
ByteBuffer[] uncompressedBuffers = new ByteBuffer[3];
uncompressedBuffers[0] = ByteBuffer.wrap("1 hello 1".getBytes(StandardCharsets.ISO_8859_1));
uncompressedBuffers[1] = ByteBuffer.wrap("2 howdy 2".getBytes(StandardCharsets.ISO_8859_1));
uncompressedBuffers[2] = ByteBuffer.wrap("3 hey ya 3".getBytes(StandardCharsets.ISO_8859_1));
final int totalContentBytesCount = countRemaining(uncompressedBuffers);
final String originalContentString = asString(uncompressedBuffers);
ByteBuffer[] buffers = new ByteBuffer[3];
buffers[0] = gzipByteBuffer(uncompressedBuffers[0]);
buffers[1] = gzipByteBuffer(uncompressedBuffers[1]);
buffers[2] = gzipByteBuffer(uncompressedBuffers[2]);
CyclicBarrier barrier = new CyclicBarrier(2);
ContentProducer contentProducer = new AsyncContentProducer(new ArrayDelayedHttpChannel(buffers, new HttpInput.EofContent(), scheduledExecutorService, barrier));
contentProducer.setInterceptor(new GzipHttpInputInterceptor(inflaterPool, new ArrayByteBufferPool(1, 1, 2), 1));
Throwable error = readAndAssertContent(totalContentBytesCount, originalContentString, contentProducer, totalContentBytesCount + buffers.length + 2, 25, 4, barrier);
assertThat(error, nullValue());
}
@Test
public void testBlockingContentProducerGzipInterceptorWithError() throws Exception
{
ByteBuffer[] uncompressedBuffers = new ByteBuffer[3];
uncompressedBuffers[0] = ByteBuffer.wrap("1 hello 1".getBytes(StandardCharsets.ISO_8859_1));
uncompressedBuffers[1] = ByteBuffer.wrap("2 howdy 2".getBytes(StandardCharsets.ISO_8859_1));
uncompressedBuffers[2] = ByteBuffer.wrap("3 hey ya 3".getBytes(StandardCharsets.ISO_8859_1));
final int totalContentBytesCount = countRemaining(uncompressedBuffers);
final String originalContentString = asString(uncompressedBuffers);
final Throwable expectedError = new Throwable("HttpInput idle timeout");
ByteBuffer[] buffers = new ByteBuffer[3];
buffers[0] = gzipByteBuffer(uncompressedBuffers[0]);
buffers[1] = gzipByteBuffer(uncompressedBuffers[1]);
buffers[2] = gzipByteBuffer(uncompressedBuffers[2]);
CyclicBarrier barrier = new CyclicBarrier(2);
ContentProducer contentProducer = new AsyncContentProducer(new ArrayDelayedHttpChannel(buffers, new HttpInput.ErrorContent(expectedError), scheduledExecutorService, barrier));
contentProducer.setInterceptor(new GzipHttpInputInterceptor(inflaterPool, new ArrayByteBufferPool(1, 1, 2), 32));
Throwable error = readAndAssertContent(totalContentBytesCount, originalContentString, contentProducer, (buffers.length + 1) * 2, 0, 4, barrier);
assertThat(error, Is.is(expectedError));
}
private Throwable readAndAssertContent(int totalContentBytesCount, String originalContentString, ContentProducer contentProducer, int totalContentCount, int readyCount, int notReadyCount, CyclicBarrier barrier) throws InterruptedException, BrokenBarrierException, TimeoutException
{
int readBytes = 0;
String consumedString = "";
int nextContentCount = 0;
int isReadyFalseCount = 0;
int isReadyTrueCount = 0;
Throwable error = null;
while (true)
{
if (contentProducer.isReady())
isReadyTrueCount++;
else
isReadyFalseCount++;
HttpInput.Content content = contentProducer.nextContent();
nextContentCount++;
if (content == null)
{
barrier.await(5, TimeUnit.SECONDS);
content = contentProducer.nextContent();
nextContentCount++;
}
assertThat(content, notNullValue());
if (content.isSpecial())
{
if (content.isEof())
break;
error = content.getError();
break;
}
byte[] b = new byte[content.remaining()];
readBytes += b.length;
content.getByteBuffer().get(b);
consumedString += new String(b, StandardCharsets.ISO_8859_1);
content.skip(content.remaining());
}
assertThat(nextContentCount, is(totalContentCount));
assertThat(readBytes, is(totalContentBytesCount));
assertThat(consumedString, is(originalContentString));
assertThat(isReadyFalseCount, is(notReadyCount));
assertThat(isReadyTrueCount, is(readyCount));
return error;
}
private static int countRemaining(ByteBuffer[] byteBuffers)
{
int total = 0;
for (ByteBuffer byteBuffer : byteBuffers)
{
total += byteBuffer.remaining();
}
return total;
}
private static String asString(ByteBuffer[] buffers)
{
StringBuilder sb = new StringBuilder();
for (ByteBuffer buffer : buffers)
{
byte[] b = new byte[buffer.remaining()];
buffer.duplicate().get(b);
sb.append(new String(b, StandardCharsets.ISO_8859_1));
}
return sb.toString();
}
private static ByteBuffer gzipByteBuffer(ByteBuffer uncompressedBuffer)
{
try
{
ByteArrayOutputStream baos = new ByteArrayOutputStream();
GZIPOutputStream output = new GZIPOutputStream(baos);
byte[] b = new byte[uncompressedBuffer.remaining()];
uncompressedBuffer.get(b);
output.write(b);
output.close();
return ByteBuffer.wrap(baos.toByteArray());
}
catch (IOException e)
{
throw new RuntimeException(e);
}
}
private static class ArrayDelayedHttpChannel extends HttpChannel
{
private final ByteBuffer[] byteBuffers;
private final HttpInput.Content finalContent;
private final ScheduledExecutorService scheduledExecutorService;
private final CyclicBarrier barrier;
private int counter;
private volatile HttpInput.Content nextContent;
public ArrayDelayedHttpChannel(ByteBuffer[] byteBuffers, HttpInput.Content finalContent, ScheduledExecutorService scheduledExecutorService, CyclicBarrier barrier)
{
super(new MockConnector(), new HttpConfiguration(), null, null);
this.byteBuffers = new ByteBuffer[byteBuffers.length];
this.finalContent = finalContent;
this.scheduledExecutorService = scheduledExecutorService;
this.barrier = barrier;
for (int i = 0; i < byteBuffers.length; i++)
{
this.byteBuffers[i] = byteBuffers[i].duplicate();
}
}
@Override
public boolean needContent()
{
if (nextContent != null)
return true;
scheduledExecutorService.schedule(() ->
{
if (byteBuffers.length > counter)
nextContent = new HttpInput.Content(byteBuffers[counter++]);
else
nextContent = finalContent;
try
{
barrier.await(5, TimeUnit.SECONDS);
}
catch (Exception e)
{
throw new AssertionError(e);
}
}, 50, TimeUnit.MILLISECONDS);
return false;
}
@Override
public HttpInput.Content produceContent()
{
HttpInput.Content result = nextContent;
nextContent = null;
return result;
}
@Override
public boolean failAllContent(Throwable failure)
{
nextContent = null;
counter = byteBuffers.length;
return false;
}
@Override
public boolean failed(Throwable x)
{
return false;
}
@Override
protected boolean eof()
{
return false;
}
}
}

320
jetty-server/src/test/java/org/eclipse/jetty/server/BlockingContentProducerTest.java Normal file
View File

@ -0,0 +1,320 @@
//
// ========================================================================
// Copyright (c) 1995-2020 Mort Bay Consulting Pty Ltd and others.
//
// This program and the accompanying materials are made available under
// the terms of the Eclipse Public License 2.0 which is available at
// https://www.eclipse.org/legal/epl-2.0
//
// This Source Code may also be made available under the following
// Secondary Licenses when the conditions for such availability set
// forth in the Eclipse Public License, v. 2.0 are satisfied:
// the Apache License v2.0 which is available at
// https://www.apache.org/licenses/LICENSE-2.0
//
// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
// ========================================================================
//
package org.eclipse.jetty.server;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.zip.GZIPOutputStream;
import org.eclipse.jetty.io.ArrayByteBufferPool;
import org.eclipse.jetty.io.EofException;
import org.eclipse.jetty.server.handler.gzip.GzipHttpInputInterceptor;
import org.eclipse.jetty.util.compression.InflaterPool;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.nullValue;
import static org.hamcrest.core.Is.is;
public class BlockingContentProducerTest
{
private ScheduledExecutorService scheduledExecutorService;
private InflaterPool inflaterPool;
@BeforeEach
public void setUp()
{
scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
inflaterPool = new InflaterPool(-1, true);
}
@AfterEach
public void tearDown()
{
scheduledExecutorService.shutdownNow();
}
@Test
public void testBlockingContentProducerNoInterceptor()
{
ByteBuffer[] buffers = new ByteBuffer[3];
buffers[0] = ByteBuffer.wrap("1 hello 1".getBytes(StandardCharsets.ISO_8859_1));
buffers[1] = ByteBuffer.wrap("2 howdy 2".getBytes(StandardCharsets.ISO_8859_1));
buffers[2] = ByteBuffer.wrap("3 hey ya 3".getBytes(StandardCharsets.ISO_8859_1));
final int totalContentBytesCount = countRemaining(buffers);
final String originalContentString = asString(buffers);
AtomicReference<ContentProducer> ref = new AtomicReference<>();
ArrayDelayedHttpChannel httpChannel = new ArrayDelayedHttpChannel(buffers, new HttpInput.EofContent(), scheduledExecutorService, () -> ref.get().onContentProducible());
ContentProducer contentProducer = new BlockingContentProducer(new AsyncContentProducer(httpChannel));
ref.set(contentProducer);
Throwable error = readAndAssertContent(totalContentBytesCount, originalContentString, buffers.length + 1, contentProducer);
assertThat(error, nullValue());
}
@Test
public void testBlockingContentProducerNoInterceptorWithError()
{
ByteBuffer[] buffers = new ByteBuffer[3];
buffers[0] = ByteBuffer.wrap("1 hello 1".getBytes(StandardCharsets.ISO_8859_1));
buffers[1] = ByteBuffer.wrap("2 howdy 2".getBytes(StandardCharsets.ISO_8859_1));
buffers[2] = ByteBuffer.wrap("3 hey ya 3".getBytes(StandardCharsets.ISO_8859_1));
final int totalContentBytesCount = countRemaining(buffers);
final String originalContentString = asString(buffers);
final Throwable expectedError = new EofException("Early EOF");
AtomicReference<ContentProducer> ref = new AtomicReference<>();
ArrayDelayedHttpChannel httpChannel = new ArrayDelayedHttpChannel(buffers, new HttpInput.ErrorContent(expectedError), scheduledExecutorService, () -> ref.get().onContentProducible());
ContentProducer contentProducer = new BlockingContentProducer(new AsyncContentProducer(httpChannel));
ref.set(contentProducer);
Throwable error = readAndAssertContent(totalContentBytesCount, originalContentString, buffers.length + 1, contentProducer);
assertThat(error, is(expectedError));
}
@Test
public void testBlockingContentProducerGzipInterceptor()
{
ByteBuffer[] uncompressedBuffers = new ByteBuffer[3];
uncompressedBuffers[0] = ByteBuffer.wrap("1 hello 1".getBytes(StandardCharsets.ISO_8859_1));
uncompressedBuffers[1] = ByteBuffer.wrap("2 howdy 2".getBytes(StandardCharsets.ISO_8859_1));
uncompressedBuffers[2] = ByteBuffer.wrap("3 hey ya 3".getBytes(StandardCharsets.ISO_8859_1));
final int totalContentBytesCount = countRemaining(uncompressedBuffers);
final String originalContentString = asString(uncompressedBuffers);
ByteBuffer[] buffers = new ByteBuffer[3];
buffers[0] = gzipByteBuffer(uncompressedBuffers[0]);
buffers[1] = gzipByteBuffer(uncompressedBuffers[1]);
buffers[2] = gzipByteBuffer(uncompressedBuffers[2]);
AtomicReference<ContentProducer> ref = new AtomicReference<>();
ArrayDelayedHttpChannel httpChannel = new ArrayDelayedHttpChannel(buffers, new HttpInput.EofContent(), scheduledExecutorService, () -> ref.get().onContentProducible());
ContentProducer contentProducer = new BlockingContentProducer(new AsyncContentProducer(httpChannel));
ref.set(contentProducer);
contentProducer.setInterceptor(new GzipHttpInputInterceptor(inflaterPool, new ArrayByteBufferPool(1, 1, 2), 32));
Throwable error = readAndAssertContent(totalContentBytesCount, originalContentString, buffers.length + 1, contentProducer);
assertThat(error, nullValue());
}
@Test
public void testBlockingContentProducerGzipInterceptorWithTinyBuffers()
{
ByteBuffer[] uncompressedBuffers = new ByteBuffer[3];
uncompressedBuffers[0] = ByteBuffer.wrap("1 hello 1".getBytes(StandardCharsets.ISO_8859_1));
uncompressedBuffers[1] = ByteBuffer.wrap("2 howdy 2".getBytes(StandardCharsets.ISO_8859_1));
uncompressedBuffers[2] = ByteBuffer.wrap("3 hey ya 3".getBytes(StandardCharsets.ISO_8859_1));
final int totalContentBytesCount = countRemaining(uncompressedBuffers);
final String originalContentString = asString(uncompressedBuffers);
ByteBuffer[] buffers = new ByteBuffer[3];
buffers[0] = gzipByteBuffer(uncompressedBuffers[0]);
buffers[1] = gzipByteBuffer(uncompressedBuffers[1]);
buffers[2] = gzipByteBuffer(uncompressedBuffers[2]);
AtomicReference<ContentProducer> ref = new AtomicReference<>();
ArrayDelayedHttpChannel httpChannel = new ArrayDelayedHttpChannel(buffers, new HttpInput.EofContent(), scheduledExecutorService, () -> ref.get().onContentProducible());
ContentProducer contentProducer = new BlockingContentProducer(new AsyncContentProducer(httpChannel));
ref.set(contentProducer);
contentProducer.setInterceptor(new GzipHttpInputInterceptor(inflaterPool, new ArrayByteBufferPool(1, 1, 2), 1));
Throwable error = readAndAssertContent(totalContentBytesCount, originalContentString, totalContentBytesCount + 1, contentProducer);
assertThat(error, nullValue());
}
@Test
public void testBlockingContentProducerGzipInterceptorWithError()
{
ByteBuffer[] uncompressedBuffers = new ByteBuffer[3];
uncompressedBuffers[0] = ByteBuffer.wrap("1 hello 1".getBytes(StandardCharsets.ISO_8859_1));
uncompressedBuffers[1] = ByteBuffer.wrap("2 howdy 2".getBytes(StandardCharsets.ISO_8859_1));
uncompressedBuffers[2] = ByteBuffer.wrap("3 hey ya 3".getBytes(StandardCharsets.ISO_8859_1));
final int totalContentBytesCount = countRemaining(uncompressedBuffers);
final String originalContentString = asString(uncompressedBuffers);
final Throwable expectedError = new Throwable("HttpInput idle timeout");
ByteBuffer[] buffers = new ByteBuffer[3];
buffers[0] = gzipByteBuffer(uncompressedBuffers[0]);
buffers[1] = gzipByteBuffer(uncompressedBuffers[1]);
buffers[2] = gzipByteBuffer(uncompressedBuffers[2]);
AtomicReference<ContentProducer> ref = new AtomicReference<>();
ArrayDelayedHttpChannel httpChannel = new ArrayDelayedHttpChannel(buffers, new HttpInput.ErrorContent(expectedError), scheduledExecutorService, () -> ref.get().onContentProducible());
ContentProducer contentProducer = new BlockingContentProducer(new AsyncContentProducer(httpChannel));
ref.set(contentProducer);
contentProducer.setInterceptor(new GzipHttpInputInterceptor(inflaterPool, new ArrayByteBufferPool(1, 1, 2), 32));
Throwable error = readAndAssertContent(totalContentBytesCount, originalContentString, buffers.length + 1, contentProducer);
assertThat(error, is(expectedError));
}
private Throwable readAndAssertContent(int totalContentBytesCount, String originalContentString, int totalContentCount, ContentProducer contentProducer)
{
int readBytes = 0;
int nextContentCount = 0;
String consumedString = "";
Throwable error = null;
while (true)
{
HttpInput.Content content = contentProducer.nextContent();
nextContentCount++;
if (content.isSpecial())
{
if (content.isEof())
break;
error = content.getError();
break;
}
byte[] b = new byte[content.remaining()];
content.getByteBuffer().get(b);
consumedString += new String(b, StandardCharsets.ISO_8859_1);
readBytes += b.length;
}
assertThat(readBytes, is(totalContentBytesCount));
assertThat(nextContentCount, is(totalContentCount));
assertThat(consumedString, is(originalContentString));
return error;
}
private static int countRemaining(ByteBuffer[] byteBuffers)
{
int total = 0;
for (ByteBuffer byteBuffer : byteBuffers)
{
total += byteBuffer.remaining();
}
return total;
}
private static String asString(ByteBuffer[] buffers)
{
StringBuilder sb = new StringBuilder();
for (ByteBuffer buffer : buffers)
{
byte[] b = new byte[buffer.remaining()];
buffer.duplicate().get(b);
sb.append(new String(b, StandardCharsets.ISO_8859_1));
}
return sb.toString();
}
private static ByteBuffer gzipByteBuffer(ByteBuffer uncompressedBuffer)
{
try
{
ByteArrayOutputStream baos = new ByteArrayOutputStream();
GZIPOutputStream output = new GZIPOutputStream(baos);
byte[] b = new byte[uncompressedBuffer.remaining()];
uncompressedBuffer.get(b);
output.write(b);
output.close();
return ByteBuffer.wrap(baos.toByteArray());
}
catch (IOException e)
{
throw new RuntimeException(e);
}
}
private interface ContentListener
{
void onContent();
}
private static class ArrayDelayedHttpChannel extends HttpChannel
{
private final ByteBuffer[] byteBuffers;
private final HttpInput.Content finalContent;
private final ScheduledExecutorService scheduledExecutorService;
private final ContentListener contentListener;
private int counter;
private volatile HttpInput.Content nextContent;
public ArrayDelayedHttpChannel(ByteBuffer[] byteBuffers, HttpInput.Content finalContent, ScheduledExecutorService scheduledExecutorService, ContentListener contentListener)
{
super(new MockConnector(), new HttpConfiguration(), null, null);
this.byteBuffers = new ByteBuffer[byteBuffers.length];
this.finalContent = finalContent;
this.scheduledExecutorService = scheduledExecutorService;
this.contentListener = contentListener;
for (int i = 0; i < byteBuffers.length; i++)
{
this.byteBuffers[i] = byteBuffers[i].duplicate();
}
}
@Override
public boolean needContent()
{
if (nextContent != null)
return true;
scheduledExecutorService.schedule(() ->
{
if (byteBuffers.length > counter)
nextContent = new HttpInput.Content(byteBuffers[counter++]);
else
nextContent = finalContent;
contentListener.onContent();
}, 50, TimeUnit.MILLISECONDS);
return false;
}
@Override
public HttpInput.Content produceContent()
{
HttpInput.Content result = nextContent;
nextContent = null;
return result;
}
@Override
public boolean failAllContent(Throwable failure)
{
nextContent = null;
counter = byteBuffers.length;
return false;
}
@Override
public boolean failed(Throwable x)
{
return false;
}
@Override
protected boolean eof()
{
return false;
}
}
}

735
jetty-server/src/test/java/org/eclipse/jetty/server/HttpInputAsyncStateTest.java
View File

@ -1,735 +0,0 @@
//
// ========================================================================
// Copyright (c) 1995-2020 Mort Bay Consulting Pty Ltd and others.
//
// This program and the accompanying materials are made available under
// the terms of the Eclipse Public License 2.0 which is available at
// https://www.eclipse.org/legal/epl-2.0
//
// This Source Code may also be made available under the following
// Secondary Licenses when the conditions for such availability set
// forth in the Eclipse Public License, v. 2.0 are satisfied:
// the Apache License v2.0 which is available at
// https://www.apache.org/licenses/LICENSE-2.0
//
// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
// ========================================================================
//
package org.eclipse.jetty.server;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Queue;
import java.util.concurrent.LinkedBlockingQueue;
import javax.servlet.ReadListener;
import org.eclipse.jetty.server.HttpChannelState.Action;
import org.eclipse.jetty.server.HttpInput.Content;
import org.eclipse.jetty.util.BufferUtil;
import org.eclipse.jetty.util.thread.Scheduler;
import org.hamcrest.Matchers;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import static org.eclipse.jetty.server.HttpInput.EARLY_EOF_CONTENT;
import static org.eclipse.jetty.server.HttpInput.EOF_CONTENT;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.empty;
import static org.hamcrest.Matchers.equalTo;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
/**
* this tests HttpInput and its interaction with HttpChannelState
*/
public class HttpInputAsyncStateTest
{
private static final Queue<String> __history = new LinkedBlockingQueue<>();
private ByteBuffer _expected = BufferUtil.allocate(16 * 1024);
private boolean _eof;
private boolean _noReadInDataAvailable;
private boolean _completeInOnDataAvailable;
private final ReadListener _listener = new ReadListener()
{
@Override
public void onError(Throwable t)
{
__history.add("onError:" + t);
}
@Override
public void onDataAvailable() throws IOException
{
__history.add("onDataAvailable");
if (!_noReadInDataAvailable && readAvailable() && _completeInOnDataAvailable)
{
__history.add("complete");
_state.complete();
}
}
@Override
public void onAllDataRead() throws IOException
{
__history.add("onAllDataRead");
}
};
private HttpInput _in;
HttpChannelState _state;
public static class TContent extends HttpInput.Content
{
public TContent(String content)
{
super(BufferUtil.toBuffer(content));
}
}
@BeforeEach
public void before()
{
_noReadInDataAvailable = false;
_in = new HttpInput(new HttpChannelState(new HttpChannel(new MockConnector(), new HttpConfiguration(), null, null)
{
@Override
public void onAsyncWaitForContent()
{
__history.add("onAsyncWaitForContent");
}
@Override
public Scheduler getScheduler()
{
return null;
}
})
{
@Override
public void onReadUnready()
{
super.onReadUnready();
__history.add("onReadUnready");
}
@Override
public boolean onContentAdded()
{
boolean wake = super.onContentAdded();
__history.add("onReadPossible " + wake);
return wake;
}
@Override
public boolean onReadReady()
{
boolean wake = super.onReadReady();
__history.add("onReadReady " + wake);
return wake;
}
})
{
@Override
public void wake()
{
__history.add("wake");
}
};
_state = _in.getHttpChannelState();
__history.clear();
}
private void check(String... history)
{
if (history == null || history.length == 0)
assertThat(__history, empty());
else
assertThat(__history.toArray(new String[__history.size()]), Matchers.arrayContaining(history));
__history.clear();
}
private void wake()
{
handle(null);
}
private void handle()
{
handle(null);
}
private void handle(Runnable run)
{
Action action = _state.handling();
loop:
while (true)
{
switch (action)
{
case DISPATCH:
if (run == null)
fail("Run is null during DISPATCH");
run.run();
break;
case READ_CALLBACK:
_in.run();
break;
case TERMINATED:
case WAIT:
break loop;
case COMPLETE:
__history.add("COMPLETE");
break;
case READ_REGISTER:
_state.getHttpChannel().onAsyncWaitForContent();
break;
default:
fail("Bad Action: " + action);
}
action = _state.unhandle();
}
}
private void deliver(Content... content)
{
if (content != null)
{
for (Content c : content)
{
if (c == EOF_CONTENT)
{
_in.eof();
_eof = true;
}
else if (c == HttpInput.EARLY_EOF_CONTENT)
{
_in.earlyEOF();
_eof = true;
}
else
{
_in.addContent(c);
BufferUtil.append(_expected, c.getByteBuffer().slice());
}
}
}
}
boolean readAvailable() throws IOException
{
int len = 0;
try
{
while (_in.isReady())
{
int b = _in.read();
if (b < 0)
{
if (len > 0)
__history.add("read " + len);
__history.add("read -1");
assertTrue(BufferUtil.isEmpty(_expected));
assertTrue(_eof);
return true;
}
else
{
len++;
assertFalse(BufferUtil.isEmpty(_expected));
int a = 0xff & _expected.get();
assertThat(b, equalTo(a));
}
}
__history.add("read " + len);
assertTrue(BufferUtil.isEmpty(_expected));
}
catch (IOException e)
{
if (len > 0)
__history.add("read " + len);
__history.add("read " + e);
throw e;
}
return false;
}
@AfterEach
public void after()
{
assertThat(__history.poll(), Matchers.nullValue());
}
@Test
public void testInitialEmptyListenInHandle() throws Exception
{
deliver(EOF_CONTENT);
check();
handle(() ->
{
_state.startAsync(null);
_in.setReadListener(_listener);
check("onReadReady false");
});
check("onAllDataRead");
}
@Test
public void testInitialEmptyListenAfterHandle() throws Exception
{
deliver(EOF_CONTENT);
handle(() ->
{
_state.startAsync(null);
check();
});
_in.setReadListener(_listener);
check("onReadReady true", "wake");
wake();
check("onAllDataRead");
}
@Test
public void testListenInHandleEmpty() throws Exception
{
handle(() ->
{
_state.startAsync(null);
_in.setReadListener(_listener);
check("onReadUnready");
});
check("onAsyncWaitForContent");
deliver(EOF_CONTENT);
check("onReadPossible true");
handle();
check("onAllDataRead");
}
@Test
public void testEmptyListenAfterHandle() throws Exception
{
handle(() ->
{
_state.startAsync(null);
check();
});
deliver(EOF_CONTENT);
check();
_in.setReadListener(_listener);
check("onReadReady true", "wake");
wake();
check("onAllDataRead");
}
@Test
public void testListenAfterHandleEmpty() throws Exception
{
handle(() ->
{
_state.startAsync(null);
check();
});
_in.setReadListener(_listener);
check("onAsyncWaitForContent", "onReadUnready");
deliver(EOF_CONTENT);
check("onReadPossible true");
handle();
check("onAllDataRead");
}
@Test
public void testInitialEarlyEOFListenInHandle() throws Exception
{
deliver(EARLY_EOF_CONTENT);
check();
handle(() ->
{
_state.startAsync(null);
_in.setReadListener(_listener);
check("onReadReady false");
});
check("onError:org.eclipse.jetty.io.EofException: Early EOF");
}
@Test
public void testInitialEarlyEOFListenAfterHandle() throws Exception
{
deliver(EARLY_EOF_CONTENT);
handle(() ->
{
_state.startAsync(null);
check();
});
_in.setReadListener(_listener);
check("onReadReady true", "wake");
wake();
check("onError:org.eclipse.jetty.io.EofException: Early EOF");
}
@Test
public void testListenInHandleEarlyEOF() throws Exception
{
handle(() ->
{
_state.startAsync(null);
_in.setReadListener(_listener);
check("onReadUnready");
});
check("onAsyncWaitForContent");
deliver(EARLY_EOF_CONTENT);
check("onReadPossible true");
handle();
check("onError:org.eclipse.jetty.io.EofException: Early EOF");
}
@Test
public void testEarlyEOFListenAfterHandle() throws Exception
{
handle(() ->
{
_state.startAsync(null);
check();
});
deliver(EARLY_EOF_CONTENT);
check();
_in.setReadListener(_listener);
check("onReadReady true", "wake");
wake();
check("onError:org.eclipse.jetty.io.EofException: Early EOF");
}
@Test
public void testListenAfterHandleEarlyEOF() throws Exception
{
handle(() ->
{
_state.startAsync(null);
check();
});
_in.setReadListener(_listener);
check("onAsyncWaitForContent", "onReadUnready");
deliver(EARLY_EOF_CONTENT);
check("onReadPossible true");
handle();
check("onError:org.eclipse.jetty.io.EofException: Early EOF");
}
@Test
public void testInitialAllContentListenInHandle() throws Exception
{
deliver(new TContent("Hello"), EOF_CONTENT);
check();
handle(() ->
{
_state.startAsync(null);
_in.setReadListener(_listener);
check("onReadReady false");
});
check("onDataAvailable", "read 5", "read -1", "onAllDataRead");
}
@Test
public void testInitialAllContentListenAfterHandle() throws Exception
{
deliver(new TContent("Hello"), EOF_CONTENT);
handle(() ->
{
_state.startAsync(null);
check();
});
_in.setReadListener(_listener);
check("onReadReady true", "wake");
wake();
check("onDataAvailable", "read 5", "read -1", "onAllDataRead");
}
@Test
public void testListenInHandleAllContent() throws Exception
{
handle(() ->
{
_state.startAsync(null);
_in.setReadListener(_listener);
check("onReadUnready");
});
check("onAsyncWaitForContent");
deliver(new TContent("Hello"), EOF_CONTENT);
check("onReadPossible true", "onReadPossible false");
handle();
check("onDataAvailable", "read 5", "read -1", "onAllDataRead");
}
@Test
public void testAllContentListenAfterHandle() throws Exception
{
handle(() ->
{
_state.startAsync(null);
check();
});
deliver(new TContent("Hello"), EOF_CONTENT);
check();
_in.setReadListener(_listener);
check("onReadReady true", "wake");
wake();
check("onDataAvailable", "read 5", "read -1", "onAllDataRead");
}
@Test
public void testListenAfterHandleAllContent() throws Exception
{
handle(() ->
{
_state.startAsync(null);
check();
});
_in.setReadListener(_listener);
check("onAsyncWaitForContent", "onReadUnready");
deliver(new TContent("Hello"), EOF_CONTENT);
check("onReadPossible true", "onReadPossible false");
handle();
check("onDataAvailable", "read 5", "read -1", "onAllDataRead");
}
@Test
public void testInitialIncompleteContentListenInHandle() throws Exception
{
deliver(new TContent("Hello"), EARLY_EOF_CONTENT);
check();
handle(() ->
{
_state.startAsync(null);
_in.setReadListener(_listener);
check("onReadReady false");
});
check(
"onDataAvailable",
"read 5",
"read org.eclipse.jetty.io.EofException: Early EOF",
"onError:org.eclipse.jetty.io.EofException: Early EOF");
}
@Test
public void testInitialPartialContentListenAfterHandle() throws Exception
{
deliver(new TContent("Hello"), EARLY_EOF_CONTENT);
handle(() ->
{
_state.startAsync(null);
check();
});
_in.setReadListener(_listener);
check("onReadReady true", "wake");
wake();
check(
"onDataAvailable",
"read 5",
"read org.eclipse.jetty.io.EofException: Early EOF",
"onError:org.eclipse.jetty.io.EofException: Early EOF");
}
@Test
public void testListenInHandlePartialContent() throws Exception
{
handle(() ->
{
_state.startAsync(null);
_in.setReadListener(_listener);
check("onReadUnready");
});
check("onAsyncWaitForContent");
deliver(new TContent("Hello"), EARLY_EOF_CONTENT);
check("onReadPossible true", "onReadPossible false");
handle();
check(
"onDataAvailable",
"read 5",
"read org.eclipse.jetty.io.EofException: Early EOF",
"onError:org.eclipse.jetty.io.EofException: Early EOF");
}
@Test
public void testPartialContentListenAfterHandle() throws Exception
{
handle(() ->
{
_state.startAsync(null);
check();
});
deliver(new TContent("Hello"), EARLY_EOF_CONTENT);
check();
_in.setReadListener(_listener);
check("onReadReady true", "wake");
wake();
check(
"onDataAvailable",
"read 5",
"read org.eclipse.jetty.io.EofException: Early EOF",
"onError:org.eclipse.jetty.io.EofException: Early EOF");
}
@Test
public void testListenAfterHandlePartialContent() throws Exception
{
handle(() ->
{
_state.startAsync(null);
check();
});
_in.setReadListener(_listener);
check("onAsyncWaitForContent", "onReadUnready");
deliver(new TContent("Hello"), EARLY_EOF_CONTENT);
check("onReadPossible true", "onReadPossible false");
handle();
check(
"onDataAvailable",
"read 5",
"read org.eclipse.jetty.io.EofException: Early EOF",
"onError:org.eclipse.jetty.io.EofException: Early EOF");
}
@Test
public void testReadAfterOnDataAvailable() throws Exception
{
_noReadInDataAvailable = true;
handle(() ->
{
_state.startAsync(null);
_in.setReadListener(_listener);
check("onReadUnready");
});
check("onAsyncWaitForContent");
deliver(new TContent("Hello"), EOF_CONTENT);
check("onReadPossible true", "onReadPossible false");
handle();
check("onDataAvailable");
readAvailable();
check("wake", "read 5", "read -1");
wake();
check("onAllDataRead");
}
@Test
public void testReadOnlyExpectedAfterOnDataAvailable() throws Exception
{
_noReadInDataAvailable = true;
handle(() ->
{
_state.startAsync(null);
_in.setReadListener(_listener);
check("onReadUnready");
});
check("onAsyncWaitForContent");
deliver(new TContent("Hello"), EOF_CONTENT);
check("onReadPossible true", "onReadPossible false");
handle();
check("onDataAvailable");
byte[] buffer = new byte[_expected.remaining()];
assertThat(_in.read(buffer), equalTo(buffer.length));
assertThat(new String(buffer), equalTo(BufferUtil.toString(_expected)));
BufferUtil.clear(_expected);
check();
assertTrue(_in.isReady());
check();
assertThat(_in.read(), equalTo(-1));
check("wake");
wake();
check("onAllDataRead");
}
@Test
public void testReadAndCompleteInOnDataAvailable() throws Exception
{
_completeInOnDataAvailable = true;
handle(() ->
{
_state.startAsync(null);
_in.setReadListener(_listener);
check("onReadUnready");
});
check("onAsyncWaitForContent");
deliver(new TContent("Hello"), EOF_CONTENT);
check("onReadPossible true", "onReadPossible false");
handle(() ->
{
__history.add(_state.getState().toString());
});
System.err.println(__history);
check(
"onDataAvailable",
"read 5",
"read -1",
"complete",
"COMPLETE"
);
}
}

614
jetty-server/src/test/java/org/eclipse/jetty/server/HttpInputTest.java
View File

@ -1,614 +0,0 @@
//
// ========================================================================
// Copyright (c) 1995-2020 Mort Bay Consulting Pty Ltd and others.
//
// This program and the accompanying materials are made available under
// the terms of the Eclipse Public License 2.0 which is available at
// https://www.eclipse.org/legal/epl-2.0
//
// This Source Code may also be made available under the following
// Secondary Licenses when the conditions for such availability set
// forth in the Eclipse Public License, v. 2.0 are satisfied:
// the Apache License v2.0 which is available at
// https://www.apache.org/licenses/LICENSE-2.0
//
// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
// ========================================================================
//
package org.eclipse.jetty.server;
import java.io.EOFException;
import java.io.IOException;
import java.util.Queue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeoutException;
import javax.servlet.ReadListener;
import org.eclipse.jetty.util.BufferUtil;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.nullValue;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
public class HttpInputTest
{
private final Queue<String> _history = new LinkedBlockingQueue<>();
private final Queue<String> _fillAndParseSimulate = new LinkedBlockingQueue<>();
private final ReadListener _listener = new ReadListener()
{
@Override
public void onError(Throwable t)
{
_history.add("l.onError:" + t);
}
@Override
public void onDataAvailable() throws IOException
{
_history.add("l.onDataAvailable");
}
@Override
public void onAllDataRead() throws IOException
{
_history.add("l.onAllDataRead");
}
};
private HttpInput _in;
public class TContent extends HttpInput.Content
{
private final String _content;
public TContent(String content)
{
super(BufferUtil.toBuffer(content));
_content = content;
}
@Override
public void succeeded()
{
_history.add("Content succeeded " + _content);
super.succeeded();
}
@Override
public void failed(Throwable x)
{
_history.add("Content failed " + _content);
super.failed(x);
}
}
public class TestHttpInput extends HttpInput
{
public TestHttpInput(HttpChannelState state)
{
super(state);
}
@Override
protected void produceContent() throws IOException
{
_history.add("produceContent " + _fillAndParseSimulate.size());
for (String s = _fillAndParseSimulate.poll(); s != null; s = _fillAndParseSimulate.poll())
{
if ("_EOF_".equals(s))
_in.eof();
else
_in.addContent(new TContent(s));
}
}
@Override
protected void blockForContent() throws IOException
{
_history.add("blockForContent");
super.blockForContent();
}
}
public class TestHttpChannelState extends HttpChannelState
{
private boolean _fakeAsyncState;
public TestHttpChannelState(HttpChannel channel)
{
super(channel);
}
public boolean isFakeAsyncState()
{
return _fakeAsyncState;
}
public void setFakeAsyncState(boolean fakeAsyncState)
{
_fakeAsyncState = fakeAsyncState;
}
@Override
public boolean isAsyncStarted()
{
if (isFakeAsyncState())
return true;
return super.isAsyncStarted();
}
@Override
public void onReadUnready()
{
_history.add("s.onReadUnready");
super.onReadUnready();
}
@Override
public boolean onReadPossible()
{
_history.add("s.onReadPossible");
return super.onReadPossible();
}
@Override
public boolean onContentAdded()
{
_history.add("s.onDataAvailable");
return super.onContentAdded();
}
@Override
public boolean onReadReady()
{
_history.add("s.onReadReady");
return super.onReadReady();
}
}
@BeforeEach
public void before()
{
_in = new TestHttpInput(new TestHttpChannelState(new HttpChannel(new MockConnector(), new HttpConfiguration(), null, null)
{
@Override
public void onAsyncWaitForContent()
{
_history.add("asyncReadInterested");
}
})
);
}
@AfterEach
public void after()
{
assertThat(_history.poll(), nullValue());
}
@Test
public void testEmpty() throws Exception
{
assertThat(_in.available(), equalTo(0));
assertThat(_history.poll(), equalTo("produceContent 0"));
assertThat(_history.poll(), nullValue());
assertThat(_in.isFinished(), equalTo(false));
assertThat(_in.isReady(), equalTo(true));
assertThat(_history.poll(), nullValue());
}
@Test
public void testRead() throws Exception
{
_in.addContent(new TContent("AB"));
_in.addContent(new TContent("CD"));
_fillAndParseSimulate.offer("EF");
_fillAndParseSimulate.offer("GH");
assertThat(_in.available(), equalTo(2));
assertThat(_in.isFinished(), equalTo(false));
assertThat(_in.isReady(), equalTo(true));
assertThat(_in.getContentConsumed(), equalTo(0L));
assertThat(_in.read(), equalTo((int)'A'));
assertThat(_in.getContentConsumed(), equalTo(1L));
assertThat(_in.read(), equalTo((int)'B'));
assertThat(_in.getContentConsumed(), equalTo(2L));
assertThat(_history.poll(), equalTo("Content succeeded AB"));
assertThat(_history.poll(), nullValue());
assertThat(_in.read(), equalTo((int)'C'));
assertThat(_in.read(), equalTo((int)'D'));
assertThat(_history.poll(), equalTo("Content succeeded CD"));
assertThat(_history.poll(), nullValue());
assertThat(_in.read(), equalTo((int)'E'));
assertThat(_in.read(), equalTo((int)'F'));
assertThat(_history.poll(), equalTo("produceContent 2"));
assertThat(_history.poll(), equalTo("Content succeeded EF"));
assertThat(_history.poll(), nullValue());
assertThat(_in.read(), equalTo((int)'G'));
assertThat(_in.read(), equalTo((int)'H'));
assertThat(_history.poll(), equalTo("Content succeeded GH"));
assertThat(_history.poll(), nullValue());
assertThat(_in.getContentConsumed(), equalTo(8L));
assertThat(_history.poll(), nullValue());
}
@Test
public void testBlockingRead() throws Exception
{
new Thread(() ->
{
try
{
Thread.sleep(500);
_in.addContent(new TContent("AB"));
}
catch (Throwable th)
{
th.printStackTrace();
}
}).start();
assertThat(_in.read(), equalTo((int)'A'));
assertThat(_history.poll(), equalTo("produceContent 0"));
assertThat(_history.poll(), equalTo("blockForContent"));
assertThat(_history.poll(), nullValue());
assertThat(_in.read(), equalTo((int)'B'));
assertThat(_history.poll(), equalTo("Content succeeded AB"));
assertThat(_history.poll(), nullValue());
}
@Test
public void testReadEOF() throws Exception
{
_in.addContent(new TContent("AB"));
_in.addContent(new TContent("CD"));
_in.eof();
assertThat(_in.isFinished(), equalTo(false));
assertThat(_in.available(), equalTo(2));
assertThat(_in.isFinished(), equalTo(false));
assertThat(_in.read(), equalTo((int)'A'));
assertThat(_in.read(), equalTo((int)'B'));
assertThat(_history.poll(), equalTo("Content succeeded AB"));
assertThat(_history.poll(), nullValue());
assertThat(_in.read(), equalTo((int)'C'));
assertThat(_in.isFinished(), equalTo(false));
assertThat(_in.read(), equalTo((int)'D'));
assertThat(_history.poll(), equalTo("Content succeeded CD"));
assertThat(_history.poll(), nullValue());
assertThat(_in.isFinished(), equalTo(false));
assertThat(_in.read(), equalTo(-1));
assertThat(_in.isFinished(), equalTo(true));
assertThat(_history.poll(), nullValue());
}
@Test
public void testReadEarlyEOF() throws Exception
{
_in.addContent(new TContent("AB"));
_in.addContent(new TContent("CD"));
_in.earlyEOF();
assertThat(_in.isFinished(), equalTo(false));
assertThat(_in.available(), equalTo(2));
assertThat(_in.isFinished(), equalTo(false));
assertThat(_in.read(), equalTo((int)'A'));
assertThat(_in.read(), equalTo((int)'B'));
assertThat(_in.read(), equalTo((int)'C'));
assertThat(_in.isFinished(), equalTo(false));
assertThat(_in.read(), equalTo((int)'D'));
assertThrows(EOFException.class, () -> _in.read());
assertTrue(_in.isFinished());
assertThat(_history.poll(), equalTo("Content succeeded AB"));
assertThat(_history.poll(), equalTo("Content succeeded CD"));
assertThat(_history.poll(), nullValue());
}
@Test
public void testBlockingEOF() throws Exception
{
new Thread(() ->
{
try
{
Thread.sleep(500);
_in.eof();
}
catch (Throwable th)
{
th.printStackTrace();
}
}).start();
assertThat(_in.isFinished(), equalTo(false));
assertThat(_in.read(), equalTo(-1));
assertThat(_in.isFinished(), equalTo(true));
assertThat(_history.poll(), equalTo("produceContent 0"));
assertThat(_history.poll(), equalTo("blockForContent"));
assertThat(_history.poll(), nullValue());
}
@Test
public void testAsyncEmpty() throws Exception
{
((TestHttpChannelState)_in.getHttpChannelState()).setFakeAsyncState(true);
_in.setReadListener(_listener);
((TestHttpChannelState)_in.getHttpChannelState()).setFakeAsyncState(false);
assertThat(_history.poll(), equalTo("produceContent 0"));
assertThat(_history.poll(), equalTo("s.onReadUnready"));
assertThat(_history.poll(), nullValue());
assertThat(_in.isReady(), equalTo(false));
assertThat(_history.poll(), nullValue());
assertThat(_in.isReady(), equalTo(false));
assertThat(_history.poll(), nullValue());
}
@Test
public void testAsyncRead() throws Exception
{
((TestHttpChannelState)_in.getHttpChannelState()).setFakeAsyncState(true);
_in.setReadListener(_listener);
((TestHttpChannelState)_in.getHttpChannelState()).setFakeAsyncState(false);
assertThat(_history.poll(), equalTo("produceContent 0"));
assertThat(_history.poll(), equalTo("s.onReadUnready"));
assertThat(_history.poll(), nullValue());
assertThat(_in.isReady(), equalTo(false));
assertThat(_history.poll(), nullValue());
_in.addContent(new TContent("AB"));
_fillAndParseSimulate.add("CD");
assertThat(_history.poll(), equalTo("s.onDataAvailable"));
assertThat(_history.poll(), nullValue());
_in.run();
assertThat(_history.poll(), equalTo("l.onDataAvailable"));
assertThat(_history.poll(), nullValue());
assertThat(_in.isReady(), equalTo(true));
assertThat(_in.read(), equalTo((int)'A'));
assertThat(_in.isReady(), equalTo(true));
assertThat(_in.read(), equalTo((int)'B'));
assertThat(_history.poll(), equalTo("Content succeeded AB"));
assertThat(_history.poll(), nullValue());
assertThat(_in.isReady(), equalTo(true));
assertThat(_history.poll(), equalTo("produceContent 1"));
assertThat(_history.poll(), equalTo("s.onDataAvailable"));
assertThat(_history.poll(), nullValue());
assertThat(_in.read(), equalTo((int)'C'));
assertThat(_in.isReady(), equalTo(true));
assertThat(_in.read(), equalTo((int)'D'));
assertThat(_history.poll(), equalTo("Content succeeded CD"));
assertThat(_history.poll(), nullValue());
assertThat(_in.isReady(), equalTo(false));
assertThat(_history.poll(), equalTo("produceContent 0"));
assertThat(_history.poll(), equalTo("s.onReadUnready"));
assertThat(_history.poll(), nullValue());
}
@Test
public void testAsyncEOF() throws Exception
{
((TestHttpChannelState)_in.getHttpChannelState()).setFakeAsyncState(true);
_in.setReadListener(_listener);
((TestHttpChannelState)_in.getHttpChannelState()).setFakeAsyncState(false);
assertThat(_history.poll(), equalTo("produceContent 0"));
assertThat(_history.poll(), equalTo("s.onReadUnready"));
assertThat(_history.poll(), nullValue());
_in.eof();
assertThat(_in.isReady(), equalTo(true));
assertThat(_in.isFinished(), equalTo(false));
assertThat(_history.poll(), equalTo("s.onDataAvailable"));
assertThat(_history.poll(), nullValue());
assertThat(_in.read(), equalTo(-1));
assertThat(_in.isFinished(), equalTo(true));
assertThat(_history.poll(), nullValue());
}
@Test
public void testAsyncReadEOF() throws Exception
{
((TestHttpChannelState)_in.getHttpChannelState()).setFakeAsyncState(true);
_in.setReadListener(_listener);
((TestHttpChannelState)_in.getHttpChannelState()).setFakeAsyncState(false);
assertThat(_history.poll(), equalTo("produceContent 0"));
assertThat(_history.poll(), equalTo("s.onReadUnready"));
assertThat(_history.poll(), nullValue());
assertThat(_in.isReady(), equalTo(false));
assertThat(_history.poll(), nullValue());
_in.addContent(new TContent("AB"));
_fillAndParseSimulate.add("_EOF_");
assertThat(_history.poll(), equalTo("s.onDataAvailable"));
assertThat(_history.poll(), nullValue());
_in.run();
assertThat(_history.poll(), equalTo("l.onDataAvailable"));
assertThat(_history.poll(), nullValue());
assertThat(_in.isReady(), equalTo(true));
assertThat(_in.read(), equalTo((int)'A'));
assertThat(_in.isReady(), equalTo(true));
assertThat(_in.read(), equalTo((int)'B'));
assertThat(_history.poll(), equalTo("Content succeeded AB"));
assertThat(_history.poll(), nullValue());
assertThat(_in.isFinished(), equalTo(false));
assertThat(_in.isReady(), equalTo(true));
assertThat(_history.poll(), equalTo("produceContent 1"));
assertThat(_history.poll(), equalTo("s.onDataAvailable"));
assertThat(_history.poll(), nullValue());
assertThat(_in.isFinished(), equalTo(false));
assertThat(_in.read(), equalTo(-1));
assertThat(_in.isFinished(), equalTo(true));
assertThat(_history.poll(), nullValue());
assertThat(_in.isReady(), equalTo(true));
assertThat(_history.poll(), nullValue());
}
@Test
public void testAsyncError() throws Exception
{
((TestHttpChannelState)_in.getHttpChannelState()).setFakeAsyncState(true);
_in.setReadListener(_listener);
((TestHttpChannelState)_in.getHttpChannelState()).setFakeAsyncState(false);
assertThat(_history.poll(), equalTo("produceContent 0"));
assertThat(_history.poll(), equalTo("s.onReadUnready"));
assertThat(_history.poll(), nullValue());
assertThat(_in.isReady(), equalTo(false));
assertThat(_history.poll(), nullValue());
_in.failed(new TimeoutException());
assertThat(_history.poll(), equalTo("s.onDataAvailable"));
assertThat(_history.poll(), nullValue());
_in.run();
assertThat(_in.isFinished(), equalTo(true));
assertThat(_history.poll(), equalTo("l.onError:java.util.concurrent.TimeoutException"));
assertThat(_history.poll(), nullValue());
assertThat(_in.isReady(), equalTo(true));
IOException e = assertThrows(IOException.class, () -> _in.read());
assertThat(e.getCause(), instanceOf(TimeoutException.class));
assertThat(_in.isFinished(), equalTo(true));
assertThat(_history.poll(), nullValue());
}
@Test
public void testSetListenerWithNull() throws Exception
{
//test can't be null
assertThrows(NullPointerException.class, () ->
{
_in.setReadListener(null);
});
}
@Test
public void testSetListenerNotAsync() throws Exception
{
//test not async
assertThrows(IllegalStateException.class, () ->
{
_in.setReadListener(_listener);
});
}
@Test
public void testSetListenerAlreadySet() throws Exception
{
//set up a listener
((TestHttpChannelState)_in.getHttpChannelState()).setFakeAsyncState(true);
_in.setReadListener(_listener);
//throw away any events generated by setting the listener
_history.clear();
((TestHttpChannelState)_in.getHttpChannelState()).setFakeAsyncState(false);
//now test that you can't set another listener
assertThrows(IllegalStateException.class, () ->
{
_in.setReadListener(_listener);
});
}
@Test
public void testRecycle() throws Exception
{
testAsyncRead();
_in.recycle();
testAsyncRead();
_in.recycle();
testReadEOF();
}
@Test
public void testConsumeAll() throws Exception
{
_in.addContent(new TContent("AB"));
_in.addContent(new TContent("CD"));
_fillAndParseSimulate.offer("EF");
_fillAndParseSimulate.offer("GH");
assertThat(_in.read(), equalTo((int)'A'));
assertFalse(_in.consumeAll());
assertThat(_in.getContentConsumed(), equalTo(8L));
assertThat(_history.poll(), equalTo("Content succeeded AB"));
assertThat(_history.poll(), equalTo("Content succeeded CD"));
assertThat(_history.poll(), equalTo("produceContent 2"));
assertThat(_history.poll(), equalTo("Content succeeded EF"));
assertThat(_history.poll(), equalTo("Content succeeded GH"));
assertThat(_history.poll(), equalTo("produceContent 0"));
assertThat(_history.poll(), nullValue());
}
@Test
public void testConsumeAllEOF() throws Exception
{
_in.addContent(new TContent("AB"));
_in.addContent(new TContent("CD"));
_fillAndParseSimulate.offer("EF");
_fillAndParseSimulate.offer("GH");
_fillAndParseSimulate.offer("_EOF_");
assertThat(_in.read(), equalTo((int)'A'));
assertTrue(_in.consumeAll());
assertThat(_in.getContentConsumed(), equalTo(8L));
assertThat(_history.poll(), equalTo("Content succeeded AB"));
assertThat(_history.poll(), equalTo("Content succeeded CD"));
assertThat(_history.poll(), equalTo("produceContent 3"));
assertThat(_history.poll(), equalTo("Content succeeded EF"));
assertThat(_history.poll(), equalTo("Content succeeded GH"));
assertThat(_history.poll(), nullValue());
}
}

30
jetty-server/src/test/java/org/eclipse/jetty/server/HttpWriterTest.java
View File

@ -48,11 +48,41 @@ public class HttpWriterTest
HttpChannel channel = new HttpChannel(new MockConnector(), new HttpConfiguration(), null, null)
{
@Override
public boolean needContent()
{
return false;
}
@Override
public HttpInput.Content produceContent()
{
return null;
}
@Override
public boolean failAllContent(Throwable failure)
{
return false;
}
@Override
public ByteBufferPool getByteBufferPool()
{
return pool;
}
@Override
public boolean failed(Throwable x)
{
return false;
}
@Override
protected boolean eof()
{
return false;
}
};
_httpOut = new HttpOutput(channel)

33
jetty-server/src/test/java/org/eclipse/jetty/server/ResponseTest.java
View File

@ -175,7 +175,38 @@ public class ResponseTest
{
_channelError = failure;
}
});
})
{
@Override
public boolean needContent()
{
return false;
}
@Override
public HttpInput.Content produceContent()
{
return null;
}
@Override
public boolean failAllContent(Throwable failure)
{
return false;
}
@Override
public boolean failed(Throwable x)
{
return false;
}
@Override
protected boolean eof()
{
return false;
}
};
}
@AfterEach

343
tests/test-http-client-transport/src/test/java/org/eclipse/jetty/http/client/AsyncIOServletTest.java
View File

@ -36,6 +36,8 @@ import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.zip.GZIPOutputStream;
import javax.servlet.AsyncContext;
import javax.servlet.DispatcherType;
import javax.servlet.ReadListener;
@ -71,8 +73,11 @@ import org.eclipse.jetty.server.HttpInput.Content;
import org.eclipse.jetty.server.Request;
import org.eclipse.jetty.server.handler.ContextHandler;
import org.eclipse.jetty.server.handler.ContextHandler.Context;
import org.eclipse.jetty.server.handler.gzip.GzipHttpInputInterceptor;
import org.eclipse.jetty.util.BufferUtil;
import org.eclipse.jetty.util.FuturePromise;
import org.eclipse.jetty.util.compression.CompressionPool;
import org.eclipse.jetty.util.compression.InflaterPool;
import org.hamcrest.Matchers;
import org.junit.jupiter.api.Assumptions;
import org.junit.jupiter.api.Disabled;
@ -87,6 +92,7 @@ import static org.eclipse.jetty.http.client.Transport.HTTP;
import static org.eclipse.jetty.util.BufferUtil.toArray;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.empty;
import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.is;
import static org.junit.jupiter.api.Assertions.assertEquals;
@ -776,11 +782,19 @@ public class AsyncIOServletTest extends AbstractTest<AsyncIOServletTest.AsyncTra
throw new IllegalStateException();
if (input.read() != 'X')
throw new IllegalStateException();
if (!input.isReady())
throw new IllegalStateException();
if (input.isReady())
{
try
{
if (input.read() != -1)
throw new IllegalStateException();
}
catch (IOException e)
{
// ignore
}
}
}
catch (IOException x)
{
throw new UncheckedIOException(x);
@ -1204,7 +1218,7 @@ public class AsyncIOServletTest extends AbstractTest<AsyncIOServletTest.AsyncTra
{
case 0:
// null transform
if (content.isEmpty())
content.skip(content.remaining());
state++;
return null;
@ -1254,7 +1268,7 @@ public class AsyncIOServletTest extends AbstractTest<AsyncIOServletTest.AsyncTra
}
default:
return null;
return content;
}
}
});
@ -1300,7 +1314,6 @@ public class AsyncIOServletTest extends AbstractTest<AsyncIOServletTest.AsyncTra
CountDownLatch clientLatch = new CountDownLatch(1);
String expected =
"S0" +
"S1" +
"S2" +
"S3S3" +
@ -1345,6 +1358,316 @@ public class AsyncIOServletTest extends AbstractTest<AsyncIOServletTest.AsyncTra
assertTrue(clientLatch.await(10, TimeUnit.SECONDS));
}
@ParameterizedTest
@ArgumentsSource(TransportProvider.class)
public void testAsyncEcho(Transport transport) throws Exception
{
init(transport);
scenario.start(new HttpServlet()
{
@Override
protected void service(HttpServletRequest request, HttpServletResponse response) throws IOException
{
System.err.println("Service " + request);
AsyncContext asyncContext = request.startAsync();
ServletInputStream input = request.getInputStream();
input.setReadListener(new ReadListener()
{
@Override
public void onDataAvailable() throws IOException
{
while (input.isReady())
{
int b = input.read();
if (b >= 0)
{
// System.err.printf("0x%2x %s %n", b, Character.isISOControl(b)?"?":(""+(char)b));
response.getOutputStream().write(b);
}
else
return;
}
}
@Override
public void onAllDataRead() throws IOException
{
asyncContext.complete();
}
@Override
public void onError(Throwable x)
{
}
});
}
});
AsyncRequestContent contentProvider = new AsyncRequestContent();
CountDownLatch clientLatch = new CountDownLatch(1);
AtomicReference<Result> resultRef = new AtomicReference<>();
scenario.client.newRequest(scenario.newURI())
.method(HttpMethod.POST)
.path(scenario.servletPath)
.body(contentProvider)
.send(new BufferingResponseListener(16 * 1024 * 1024)
{
@Override
public void onComplete(Result result)
{
resultRef.set(result);
clientLatch.countDown();
}
});
for (int i = 0; i < 1_000_000; i++)
{
contentProvider.offer(BufferUtil.toBuffer("S" + i));
}
contentProvider.close();
assertTrue(clientLatch.await(30, TimeUnit.SECONDS));
assertThat(resultRef.get().isSucceeded(), Matchers.is(true));
assertThat(resultRef.get().getResponse().getStatus(), Matchers.equalTo(HttpStatus.OK_200));
}
@ParameterizedTest
@ArgumentsSource(TransportProvider.class)
public void testAsyncInterceptedTwice(Transport transport) throws Exception
{
init(transport);
scenario.start(new HttpServlet()
{
@Override
protected void service(HttpServletRequest request, HttpServletResponse response) throws IOException
{
System.err.println("Service " + request);
final HttpInput httpInput = ((Request)request).getHttpInput();
httpInput.addInterceptor(new GzipHttpInputInterceptor(new InflaterPool(-1, true), ((Request)request).getHttpChannel().getByteBufferPool(), 1024));
httpInput.addInterceptor(content ->
{
ByteBuffer byteBuffer = content.getByteBuffer();
byte[] bytes = new byte[2];
bytes[1] = byteBuffer.get();
bytes[0] = byteBuffer.get();
return new Content(wrap(bytes));
});
AsyncContext asyncContext = request.startAsync();
ServletInputStream input = request.getInputStream();
ByteArrayOutputStream out = new ByteArrayOutputStream();
input.setReadListener(new ReadListener()
{
@Override
public void onDataAvailable() throws IOException
{
while (input.isReady())
{
int b = input.read();
if (b > 0)
{
// System.err.printf("0x%2x %s %n", b, Character.isISOControl(b)?"?":(""+(char)b));
out.write(b);
}
else if (b < 0)
return;
}
}
@Override
public void onAllDataRead() throws IOException
{
response.getOutputStream().write(out.toByteArray());
asyncContext.complete();
}
@Override
public void onError(Throwable x)
{
}
});
}
});
AsyncRequestContent contentProvider = new AsyncRequestContent();
CountDownLatch clientLatch = new CountDownLatch(1);
String expected =
"0S" +
"1S" +
"2S" +
"3S" +
"4S" +
"5S" +
"6S";
scenario.client.newRequest(scenario.newURI())
.method(HttpMethod.POST)
.path(scenario.servletPath)
.body(contentProvider)
.send(new BufferingResponseListener()
{
@Override
public void onComplete(Result result)
{
if (result.isSucceeded())
{
Response response = result.getResponse();
assertThat(response.getStatus(), Matchers.equalTo(HttpStatus.OK_200));
assertThat(getContentAsString(), Matchers.equalTo(expected));
clientLatch.countDown();
}
}
});
for (int i = 0; i < 7; i++)
{
contentProvider.offer(gzipToBuffer("S" + i));
contentProvider.flush();
}
contentProvider.close();
assertTrue(clientLatch.await(10, TimeUnit.SECONDS));
}
@ParameterizedTest
@ArgumentsSource(TransportProvider.class)
public void testAsyncInterceptedTwiceWithNulls(Transport transport) throws Exception
{
init(transport);
scenario.start(new HttpServlet()
{
@Override
protected void service(HttpServletRequest request, HttpServletResponse response) throws IOException
{
System.err.println("Service " + request);
final HttpInput httpInput = ((Request)request).getHttpInput();
httpInput.addInterceptor(content ->
{
if (content.isEmpty())
return content;
// skip contents with odd numbers
ByteBuffer duplicate = content.getByteBuffer().duplicate();
duplicate.get();
byte integer = duplicate.get();
int idx = Character.getNumericValue(integer);
Content contentCopy = new Content(content.getByteBuffer().duplicate());
content.skip(content.remaining());
if (idx % 2 == 0)
return contentCopy;
return null;
});
httpInput.addInterceptor(content ->
{
if (content.isEmpty())
return content;
// reverse the bytes
ByteBuffer byteBuffer = content.getByteBuffer();
byte[] bytes = new byte[2];
bytes[1] = byteBuffer.get();
bytes[0] = byteBuffer.get();
return new Content(wrap(bytes));
});
AsyncContext asyncContext = request.startAsync();
ServletInputStream input = request.getInputStream();
ByteArrayOutputStream out = new ByteArrayOutputStream();
input.setReadListener(new ReadListener()
{
@Override
public void onDataAvailable() throws IOException
{
while (input.isReady())
{
int b = input.read();
if (b > 0)
{
// System.err.printf("0x%2x %s %n", b, Character.isISOControl(b)?"?":(""+(char)b));
out.write(b);
}
else if (b < 0)
return;
}
}
@Override
public void onAllDataRead() throws IOException
{
response.getOutputStream().write(out.toByteArray());
asyncContext.complete();
}
@Override
public void onError(Throwable x)
{
}
});
}
});
AsyncRequestContent contentProvider = new AsyncRequestContent();
CountDownLatch clientLatch = new CountDownLatch(1);
String expected =
"0S" +
"2S" +
"4S" +
"6S";
scenario.client.newRequest(scenario.newURI())
.method(HttpMethod.POST)
.path(scenario.servletPath)
.body(contentProvider)
.send(new BufferingResponseListener()
{
@Override
public void onComplete(Result result)
{
if (result.isSucceeded())
{
Response response = result.getResponse();
assertThat(response.getStatus(), Matchers.equalTo(HttpStatus.OK_200));
assertThat(getContentAsString(), Matchers.equalTo(expected));
clientLatch.countDown();
}
}
});
contentProvider.offer(BufferUtil.toBuffer("S0"));
contentProvider.flush();
contentProvider.offer(BufferUtil.toBuffer("S1"));
contentProvider.flush();
contentProvider.offer(BufferUtil.toBuffer("S2"));
contentProvider.flush();
contentProvider.offer(BufferUtil.toBuffer("S3"));
contentProvider.flush();
contentProvider.offer(BufferUtil.toBuffer("S4"));
contentProvider.flush();
contentProvider.offer(BufferUtil.toBuffer("S5"));
contentProvider.flush();
contentProvider.offer(BufferUtil.toBuffer("S6"));
contentProvider.close();
assertTrue(clientLatch.await(10, TimeUnit.SECONDS));
}
private ByteBuffer gzipToBuffer(String s) throws IOException
{
ByteArrayOutputStream baos = new ByteArrayOutputStream();
GZIPOutputStream gzos = new GZIPOutputStream(baos);
gzos.write(s.getBytes(StandardCharsets.ISO_8859_1));
gzos.close();
return BufferUtil.toBuffer(baos.toByteArray());
}
@ParameterizedTest
@ArgumentsSource(TransportProvider.class)
public void testWriteListenerFromOtherThread(Transport transport) throws Exception
@ -1387,18 +1710,21 @@ public class AsyncIOServletTest extends AbstractTest<AsyncIOServletTest.AsyncTra
}))
.send();
assertEquals(HttpStatus.OK_200, response.getStatus());
latch.countDown();
}
catch (Throwable x)
{
failures.offer(x);
}
finally
{
latch.countDown();
}
}
});
}
assertTrue(latch.await(30, TimeUnit.SECONDS));
assertTrue(failures.isEmpty());
assertThat(failures, empty());
}
private static class Listener implements ReadListener, WriteListener
@ -1527,10 +1853,11 @@ public class AsyncIOServletTest extends AbstractTest<AsyncIOServletTest.AsyncTra
}
@Override
public void stopServer()
public void stopServer() throws Exception
{
checkScope();
scope.set(null);
super.stopServer();
}
}
}