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);