Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions changes.md
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,13 @@ Changes log
leftover of the Servlet mode, which is no longer supported. `Protocol.valueOf("war")`
now returns a new, unknown protocol. Use CLAP or FILE references instead.
- Enhancements
- HTTP/2 and HTTP/3 clients deliver response headers without waiting for the first
response body byte. This allows idle duplex and SSE responses to expose their
headers before the peer sends a message.
- Empty HTTP/2 and HTTP/3 server responses with no entity or trailers end with their
initial headers, rather than a subsequent empty DATA frame.
- Added test-only MCP and gRPC SDK feasibility probes over the Restlet HTTP/2
connectors. These do not introduce production RPC extensions.
- Added the `EventStreamRepresentation`, `ServerSentEvent` and `EventStreamReader` APIs for
streaming and reading Server-Sent Events over the HTTP connectors, including optional
comment heartbeats and reconnects honoring `retry:` and `Last-Event-ID`. The
Expand Down
38 changes: 38 additions & 0 deletions org.restlet/pom.xml
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,8 @@
<description>Restlet API and Engine</description>

<properties>
<grpc.version>1.85.1</grpc.version>
<mcp.version>2.1.0</mcp.version>
<!-- Known high-priority SpotBugs findings, to be fixed (see Engine, Resource, ...) -->
<spotbugs.max.violations>9</spotbugs.max.violations>
<!-- Native QUIC library for the HTTP/3 tests, overridden per platform by the profiles
Expand Down Expand Up @@ -76,6 +78,42 @@
<version>${lib-hamcrest-version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-core</artifactId>
<version>${grpc.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-protobuf</artifactId>
<version>${grpc.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-stub</artifactId>
<version>${grpc.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-netty-shaded</artifactId>
<version>${grpc.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.modelcontextprotocol.sdk</groupId>
<artifactId>mcp-core</artifactId>
<version>${mcp.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.modelcontextprotocol.sdk</groupId>
<artifactId>mcp-json-jackson2</artifactId>
<version>${mcp.version}</version>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<testResources>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -155,6 +155,9 @@ class NettyClientCall {
/** Indicates whether the channel was released. */
private boolean released;

/** The empty multiplexed response entity to release once the response completes. */
private Representation pendingEmptyEntity;

/**
* Constructor.
*
Expand Down Expand Up @@ -484,6 +487,12 @@ private void updateResponse(org.restlet.Response response, Status status) {
response.setEntity(entity);

if (entity != null && (entity.isEmpty() || !hasBody(response.getStatus()))) {
if (kind.isStream()
&& hasBody(response.getStatus())
&& deferEmptyEntityRelease(entity)) {
// Releasing now would reset the stream and lose any trailers still to come.
return;
}
entity.release();
if (Status.SUCCESS_RESET_CONTENT.equals(response.getStatus())
|| response.getStatus().isInformational()) {
Expand All @@ -492,6 +501,25 @@ private void updateResponse(org.restlet.Response response, Status status) {
}
}

/**
* Defers the release of an empty multiplexed response entity until the response completes.
*
* @param entity The empty response entity.
* @return True if the release was deferred, false if the response is already complete.
*/
private boolean deferEmptyEntityRelease(Representation entity) {
lock.lock();
try {
if (responseComplete) {
return false;
}
pendingEmptyEntity = entity;
return true;
} finally {
lock.unlock();
}
}

/** Indicates whether a response with the given status to this request has a body. */
private boolean hasBody(Status status) {
return !Method.HEAD.getName().equals(method)
Expand All @@ -517,7 +545,11 @@ private Representation getResponseEntity(Status status) {
: HeaderUtils.getContentLength(responseHeaders);

if (hasBody(status)) {
InputStream stream = nonEmpty(getResponseEntityStream());
InputStream stream = getResponseEntityStream();
// Multiplexed streams may remain idle after their response headers arrive.
if (!kind.isStream()) {
stream = nonEmpty(stream);
}
Comment thread
jlouvel marked this conversation as resolved.
Resolved
if (stream != null) {
result = new InputRepresentation(stream, null);
result.setSize(size);
Expand Down Expand Up @@ -791,12 +823,18 @@ void onResponseComplete(io.netty.handler.codec.http.HttpHeaders trailers) {
if (stream != null) {
stream.end();
}
Representation emptyEntity;
lock.lock();
try {
responseComplete = true;
emptyEntity = pendingEmptyEntity;
pendingEmptyEntity = null;
} finally {
lock.unlock();
}
if (emptyEntity != null) {
emptyEntity.release();
}
maybeRelease();
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelPipeline;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.DefaultHttpContent;
import io.netty.handler.codec.http.DefaultHttpResponse;
import io.netty.handler.codec.http.DefaultLastHttpContent;
Expand Down Expand Up @@ -124,6 +125,9 @@ class NettyServerCall {
/** Indicates whether the response head was written. */
private volatile boolean responseHeadWritten;

/** Indicates that an empty multiplexed response ended with its initial headers. */
private boolean responseEndedWithHead;

/** Indicates whether the exchange was completed. */
private volatile boolean completed;

Expand Down Expand Up @@ -608,7 +612,8 @@ private void writeHead(boolean emptyBody) {
responseHeadWritten = true;

int code = statusCode;
HttpResponse head = newResponseHead(code);
responseEndedWithHead = kind.isStream() && emptyBody && responseTrailers.isEmpty();
HttpResponse head = newResponseHead(code, responseEndedWithHead);
boolean bodyAllowed =
code != 204 && code != 304 && code >= 200 && !"HEAD".equals(getMethod());
boolean hasTrailers = bodyAllowed && !responseTrailers.isEmpty();
Expand All @@ -628,13 +633,17 @@ private void writeHead(boolean emptyBody) {
}

/** Creates the response head with the status and the response headers. */
private HttpResponse newResponseHead(int code) {
private HttpResponse newResponseHead(int code, boolean endWithHead) {
String reason = reasonPhrase;
HttpResponseStatus status =
(reason != null && !reason.isEmpty())
? new HttpResponseStatus(code, reason)
: HttpResponseStatus.valueOf(code);
HttpResponse head = new DefaultHttpResponse(HttpVersion.HTTP_1_1, status);
HttpResponse head =
endWithHead
? new DefaultFullHttpResponse(
HttpVersion.HTTP_1_1, status, Unpooled.EMPTY_BUFFER)
: new DefaultHttpResponse(HttpVersion.HTTP_1_1, status);

for (Header header : responseHeaders) {
String name =
Expand Down Expand Up @@ -691,6 +700,11 @@ void complete() {
requestBody.close();
return;
}
if (responseEndedWithHead) {
channel.flush();
requestBody.close();
return;
}

boolean keepAlive = HttpUtil.isKeepAlive(request) && requestBody.isFullyConsumed();
LastHttpContent lastContent = LastHttpContent.EMPTY_LAST_CONTENT;
Expand Down
Loading
Loading