From bcf28bb2f35dc55e4fe3734dd95f450b2f6602cf Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Fri, 30 Oct 2026 09:59:19 -0400 Subject: [PATCH] fix: send the SSE response head before the first event The Netty server connector writes the response head without flushing it, and flushes it with the first body bytes. An EventStreamRepresentation whose writer waited before its first event, without heartbeats, so sent no status or headers until then. EventStreamRepresentation.write() now flushes the output stream before running the writer. A connector test checks, with the JDK HTTP client, that the head of an idle stream arrives over HTTP/1.1 and HTTP/2. Fixes #1598. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Jerome Louvel <374450+jlouvel@users.noreply.github.com> --- changes.md | 2 + .../EventStreamRepresentation.java | 2 + .../EventStreamConnectorTestCase.java | 79 +++++++++++++++++++ 3 files changed, 83 insertions(+) diff --git a/changes.md b/changes.md index b11842b2dd..462c033051 100644 --- a/changes.md +++ b/changes.md @@ -129,6 +129,8 @@ Changes log that readers process it at once; content containing the delimiter is rejected. The reader returns parts as they arrive, with a maximum part size. Added the `MediaType.MULTIPART_MIXED` constant. Issue #1587. + - `EventStreamRepresentation` sends the response head before the first event, so that + clients get the status and headers of an idle stream without heartbeats. Issue #1598. - Added the experimental `org.restlet.ext.websocket` extension, on HTTP/1.1 upgrades. `WebSocketUpgrader` (or `WebSocketUpgrade.accept`) accepts upgrades that went through the normal routing, filters and authenticators, and runs a `WebSocketHandler` on its own diff --git a/org.restlet/src/main/java/org/restlet/representation/EventStreamRepresentation.java b/org.restlet/src/main/java/org/restlet/representation/EventStreamRepresentation.java index 764fcc8c94..6b40bf865c 100644 --- a/org.restlet/src/main/java/org/restlet/representation/EventStreamRepresentation.java +++ b/org.restlet/src/main/java/org/restlet/representation/EventStreamRepresentation.java @@ -59,6 +59,8 @@ public EventStreamRepresentation(EventWriter writer, Duration heartbeatInterval) @Override public void write(OutputStream outputStream) throws IOException { + // Sends the response head now, so that clients get it even if the first event is late. + outputStream.flush(); try (EventSink sink = new EventSink(outputStream)) { if (heartbeatInterval == null) { writer.write(sink); diff --git a/org.restlet/src/test/java/org/restlet/engine/connector/EventStreamConnectorTestCase.java b/org.restlet/src/test/java/org/restlet/engine/connector/EventStreamConnectorTestCase.java index d264cba544..6dcd85db08 100644 --- a/org.restlet/src/test/java/org/restlet/engine/connector/EventStreamConnectorTestCase.java +++ b/org.restlet/src/test/java/org/restlet/engine/connector/EventStreamConnectorTestCase.java @@ -14,6 +14,10 @@ import static org.junit.jupiter.api.Assertions.assertTrue; import java.io.InputStream; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import org.junit.jupiter.api.AfterEach; @@ -69,6 +73,16 @@ void flushesEventsBeforeTheResourceCompletesOverHttp2() throws Exception { assertStreamsBeforeCompletion(true); } + @Test + void sendsTheResponseHeadBeforeTheFirstEventOverHttp11() throws Exception { + assertHeadBeforeFirstEvent(false); + } + + @Test + void sendsTheResponseHeadBeforeTheFirstEventOverHttp2() throws Exception { + assertHeadBeforeFirstEvent(true); + } + @Test void enforcesTheConfiguredOpenStreamLimit() throws Exception { CountDownLatch firstStreamStarted = new CountDownLatch(1); @@ -219,6 +233,71 @@ public void handle(Request request, Response response) { } } + private void assertHeadBeforeFirstEvent(boolean http2) throws Exception { + CountDownLatch sendEvent = new CountDownLatch(1); + Restlet restlet = + new Restlet() { + @Override + public void handle(Request request, Response response) { + response.setEntity( + new EventStreamRepresentation( + sink -> { + // An idle stream, without heartbeats + try { + if (!sendEvent.await(10, TimeUnit.SECONDS)) { + throw new java.io.IOException( + "Timed out waiting to send the event"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new java.io.IOException( + "Interrupted while awaiting an event", e); + } + sink.send( + ServerSentEvent.builder().data("late").build()); + })); + } + }; + Context serverContext = new Context(); + if (http2) { + serverContext.getParameters().add("http.transport.protocol", "HTTP2"); + } + server = new Server(serverContext, Protocol.HTTP, 0, restlet); + server.start(); + + // The JDK client returns the response as soon as its head arrives, before any body byte + try (HttpClient jdkClient = + HttpClient.newBuilder() + .version(http2 ? HttpClient.Version.HTTP_2 : HttpClient.Version.HTTP_1_1) + .build()) { + HttpRequest request = + HttpRequest.newBuilder( + URI.create("http://localhost:" + server.getActualPort() + "/")) + .build(); + HttpResponse response = + jdkClient + .sendAsync(request, HttpResponse.BodyHandlers.ofInputStream()) + .get(5, TimeUnit.SECONDS); + assertEquals(200, response.statusCode()); + assertEquals( + http2 ? HttpClient.Version.HTTP_2 : HttpClient.Version.HTTP_1_1, + response.version()); + assertTrue( + response.headers() + .firstValue("Content-Type") + .orElseThrow() + .startsWith(MediaType.TEXT_EVENT_STREAM.getName())); + + sendEvent.countDown(); + try (EventStreamReader reader = new EventStreamReader(response.body())) { + assertEquals("late", reader.readEvent().getData()); + assertNull(reader.readEvent()); + } + } finally { + sendEvent.countDown(); + } + } + private void assertStreamsBeforeCompletion(boolean http2) throws Exception { CountDownLatch firstEventSent = new CountDownLatch(1); CountDownLatch continueWriting = new CountDownLatch(1);