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
3 changes: 3 additions & 0 deletions changes.md
Original file line number Diff line number Diff line change
Expand Up @@ -131,6 +131,9 @@ Changes log
`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.
- `EventStreamReader` limits the size of an event, 1 Mi characters by default
(`setMaxEventSize`). A larger line or event fails the read and closes the reader. Issue
#1597.
- 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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -24,9 +24,15 @@
* A blocking reader of Server-Sent Events. When created with an {@link EventStreamSource}, it
* reconnects after EOF, honoring the server's {@code retry:} field and passing the latest event ID
* to the source for use in a {@code Last-Event-ID} request header.
*
* <p>The size of an event is limited (see {@link #setMaxEventSize(int)}), so that a server can't
* make the reader buffer an unbounded line or event in memory.
*/
public final class EventStreamReader implements Iterator<ServerSentEvent>, AutoCloseable {

/** The default maximum size of an event, in characters. */
public static final int DEFAULT_MAX_EVENT_SIZE = 1024 * 1024;

private static final long DEFAULT_RETRY_MS = 3000;

private final EventStreamSource source;
Expand All @@ -41,8 +47,12 @@

private volatile long retryDelayMs = DEFAULT_RETRY_MS;

private volatile int maxEventSize = DEFAULT_MAX_EVENT_SIZE;

private boolean beginning = true;

private boolean skipLineFeed;

private boolean currentEof;

private boolean sourceStopped;
Expand Down Expand Up @@ -104,6 +114,31 @@
return retryDelayMs;
}

/**
* Returns the maximum size of an event: the number of characters of its lines, without line
* terminators. Defaults to {@link #DEFAULT_MAX_EVENT_SIZE}.
*
* @return The maximum size of an event, in characters.
*/
public int getMaxEventSize() {
return maxEventSize;
}

/**
* Sets the maximum size of an event: the number of characters of its lines, without line
* terminators. When an event exceeds it, {@link #readEvent()} throws an {@link IOException} and
* the reader is closed.
*
* @param maxEventSize The maximum size of an event, in characters.
* @throws IllegalArgumentException If the size is not positive.
*/
public void setMaxEventSize(int maxEventSize) {
if (maxEventSize <= 0) {
throw new IllegalArgumentException("The maximum event size must be positive");
}
this.maxEventSize = maxEventSize;
}

/**
* Reads the next event, blocking as needed.
*
Expand Down Expand Up @@ -160,25 +195,66 @@
private ServerSentEvent readCurrentEvent() throws IOException {
PendingEvent pending = new PendingEvent();
ServerSentEvent result = null;
int size = 0;
while (result == null && !closed) {
String line = reader.readLine();
String line = readLine(maxEventSize - size);
if (line == null) {
currentEof = true;
return pending.build();
}
result = processLine(stripBom(line), pending);
size = line.isEmpty() ? 0 : size + line.length();
result = processLine(line, pending);
}
return result;
}

private String stripBom(String line) {
if (beginning) {
beginning = false;
if (!line.isEmpty() && line.charAt(0) == '\uFEFF') {
return line.substring(1);
/**
* Reads a line ended by CRLF, LF or CR, like {@link BufferedReader#readLine()}, but no longer
* than the given number of characters. A line feed following a carriage return is skipped on
* the next read, so that a CR doesn't block until the next character arrives. The optional byte
* order mark at the start of the stream is skipped, and not counted in the limit.
*/
private String readLine(int limit) throws IOException {

Check failure on line 217 in org.restlet/src/main/java/org/restlet/representation/EventStreamReader.java

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Refactor this method to reduce its Cognitive Complexity from 25 to the 15 allowed.

See more on https://sonarcloud.io/project/issues?id=restlet_restlet-framework-java&issues=AaGNm5_Gq8wOaMrNYFeA&open=AaGNm5_Gq8wOaMrNYFeA&pullRequest=1604
StringBuilder line = new StringBuilder();
while (true) {

Check warning on line 219 in org.restlet/src/main/java/org/restlet/representation/EventStreamReader.java

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Reduce the total number of break and continue statements in this loop to use at most one.

See more on https://sonarcloud.io/project/issues?id=restlet_restlet-framework-java&issues=AaGZ_Csl7JDpWu2qTDfn&open=AaGZ_Csl7JDpWu2qTDfn&pullRequest=1604
int c = reader.read();
if (skipLineFeed) {
skipLineFeed = false;
if (c == '\n') {
continue;
}
}
if (beginning) {
beginning = false;
if (c == '\uFEFF') {
continue;
}
}
if (c < 0) {
return line.isEmpty() ? null : line.toString();
}
if (c == '\n') {
return line.toString();
}
if (c == '\r') {
skipLineFeed = true;
return line.toString();
}
if (line.length() >= limit) {
Comment thread
jlouvel marked this conversation as resolved.
Resolved
IOException failure =
new IOException(
"The event exceeds the maximum size of "
+ maxEventSize
+ " characters");
try {
close();
} catch (IOException e) {
failure.addSuppressed(e);
}
throw failure;
}
line.append((char) c);
}
return line;
}

private ServerSentEvent processLine(String line, PendingEvent pending) {
Expand Down Expand Up @@ -251,6 +327,7 @@
}
reader = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8));
beginning = true;
skipLineFeed = false;
currentEof = false;
return true;
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -11,13 +11,16 @@
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.FilterReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.Reader;
import java.io.StringReader;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
Expand Down Expand Up @@ -148,6 +151,96 @@
reader.close();
}

@Test
void readsCarriageReturnLineEndingsWithoutWaitingForMoreInput() throws IOException {
// Fails instead of blocking when asked for more than the event, as a live stream would
Reader input =
new StringReader("data:a\rdata:b\r\r") {
@Override
public int read(char[] buffer, int offset, int length) throws IOException {
int count = super.read(buffer, offset, length);
if (count < 0) {
throw new AssertionError("Read past the available input");
}
return count;
}
};
EventStreamReader reader = new EventStreamReader(input);

assertEquals("a\nb", reader.readEvent().getData());
reader.close();
}

@Test
void acceptsEventsUpToTheMaximumSize() throws IOException {
EventStreamReader reader =
new EventStreamReader(new StringReader("data:1\nid:5\n\ndata:12345\n\n"));
reader.setMaxEventSize(10);

assertEquals(10, reader.getMaxEventSize());
assertEquals("1", reader.readEvent().getData());
assertEquals("12345", reader.readEvent().getData());
assertNull(reader.readEvent());
reader.close();
}

@Test
void rejectsAnOversizedLineAndStopsReading() throws IOException {
EventStreamReader reader =
new EventStreamReader(new StringReader("data:12345678901\n\ndata:next\n\n"));
reader.setMaxEventSize(10);

IOException failure = assertThrows(IOException.class, reader::readEvent);

assertTrue(failure.getMessage().contains("10 characters"));
assertNull(reader.readEvent());
assertFalse(reader.hasNext());
}

@Test
void rejectsAnOversizedEventSpreadOverLines() throws IOException {

Check warning on line 201 in org.restlet/src/test/java/org/restlet/representation/EventStreamTestCase.java

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Remove the declaration of thrown exception 'java.io.IOException', as it cannot be thrown from method's body.

See more on https://sonarcloud.io/project/issues?id=restlet_restlet-framework-java&issues=AaGNm6DOq8wOaMrNYFeB&open=AaGNm6DOq8wOaMrNYFeB&pullRequest=1604
EventStreamReader reader = new EventStreamReader(new StringReader(":12345\n:12345\n\n"));
reader.setMaxEventSize(10);

assertThrows(IOException.class, reader::readEvent);
}

@Test
void doesNotCountTheByteOrderMarkInTheEventSize() throws IOException {
EventStreamReader reader = new EventStreamReader(new StringReader("\uFEFFdata:12345\n\n"));
reader.setMaxEventSize(10);

assertEquals("12345", reader.readEvent().getData());
reader.close();
}

@Test
void keepsTheSizeFailureWhenClosingFails() {
IOException closeFailure = new IOException("Close failed");
EventStreamReader reader =
new EventStreamReader(
new FilterReader(new StringReader("data:12345678901\n\n")) {
@Override
public void close() throws IOException {
throw closeFailure;
}
});
reader.setMaxEventSize(10);

IOException failure = assertThrows(IOException.class, reader::readEvent);

assertTrue(failure.getMessage().contains("10 characters"));
assertSame(closeFailure, failure.getSuppressed()[0]);
}

@Test
void validatesTheMaximumEventSize() throws IOException {
try (EventStreamReader reader = new EventStreamReader(new StringReader(""))) {
assertEquals(EventStreamReader.DEFAULT_MAX_EVENT_SIZE, reader.getMaxEventSize());
assertThrows(IllegalArgumentException.class, () -> reader.setMaxEventSize(0));
}
}

@Test
void validatesEventFields() {
ServerSentEvent.Builder builder = ServerSentEvent.builder();
Expand Down
Loading