From 0738ef02970c67a3987d64439fdc0a7691c8bb16 Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Thu, 29 Oct 2026 23:19:14 -0400 Subject: [PATCH 1/6] feat: add a streaming multipart representation and reader Add protocol-neutral streaming multipart/* support (RFC 2046), the counterpart of EventStreamRepresentation and EventStreamReader for protocols whose responses are made of parts produced over time, such as GraphQL incremental delivery: - MultipartStreamRepresentation: a PartWriter sends parts through a PartSink; each part is flushed followed by the next delimiter, so that readers can process it at once, and the close delimiter is written when the writer returns. Parts whose content contains the delimiter are rejected, as they would forge parts (protocols can use short fixed boundaries such as "-"). The other parameters of the media type are kept. - MultipartStreamReader: returns parts as they arrive, as MultiPartRepresentation.Part, ignoring the preamble, the epilogue and transport padding, with a maximum part size and a Knuth-Morris-Pratt delimiter search. - MediaType.MULTIPART_MIXED. Refs #1587 Co-Authored-By: Claude Opus 5.5 Signed-off-by: Jerome Louvel <374450+jlouvel@users.noreply.github.com> --- changes.md | 6 + .../main/java/org/restlet/data/MediaType.java | 3 + .../representation/MultipartStreamReader.java | 287 ++++++++++++++++++ .../MultipartStreamRepresentation.java | 216 +++++++++++++ .../MultipartStreamTestCase.java | 273 +++++++++++++++++ 5 files changed, 785 insertions(+) create mode 100644 org.restlet/src/main/java/org/restlet/representation/MultipartStreamReader.java create mode 100644 org.restlet/src/main/java/org/restlet/representation/MultipartStreamRepresentation.java create mode 100644 org.restlet/src/test/java/org/restlet/representation/MultipartStreamTestCase.java diff --git a/changes.md b/changes.md index 1fdea34b88..b1beebf515 100644 --- a/changes.md +++ b/changes.md @@ -123,6 +123,12 @@ Changes log streaming and reading Server-Sent Events over the HTTP connectors, including optional comment heartbeats and reconnects honoring `retry:` and `Last-Event-ID`. The `server.maxEventStreams` parameter limits concurrent event streams. + - Added the `MultipartStreamRepresentation` and `MultipartStreamReader` APIs for streaming + `multipart/*` content whose parts are produced over time, such as `multipart/mixed` + responses of incremental protocols. Each part is flushed with its trailing delimiter so + 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. - 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/data/MediaType.java b/org.restlet/src/main/java/org/restlet/data/MediaType.java index d870a7d7c3..0c36f0fc0f 100644 --- a/org.restlet/src/main/java/org/restlet/data/MediaType.java +++ b/org.restlet/src/main/java/org/restlet/data/MediaType.java @@ -444,6 +444,9 @@ public final class MediaType extends Metadata { public static final MediaType MULTIPART_FORM_DATA = register("multipart/form-data", "Multipart form data"); + public static final MediaType MULTIPART_MIXED = + register("multipart/mixed", "Multipart mixed content"); + public static final MediaType TEXT_ALL = register("text/*", "All texts"); public static final MediaType TEXT_CALENDAR = register("text/calendar", "iCalendar event"); diff --git a/org.restlet/src/main/java/org/restlet/representation/MultipartStreamReader.java b/org.restlet/src/main/java/org/restlet/representation/MultipartStreamReader.java new file mode 100644 index 0000000000..49f3dd32a2 --- /dev/null +++ b/org.restlet/src/main/java/org/restlet/representation/MultipartStreamReader.java @@ -0,0 +1,287 @@ +/** + * Copyright 2005-2026 Qlik + *

+ * The content of this file is subject to the terms of the Apache 2.0 open + * source license available at https://www.opensource.org/licenses/apache-2.0 + *

+ * Restlet is a registered trademark of QlikTech International AB. + */ +package org.restlet.representation; + +import java.io.BufferedInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.UncheckedIOException; +import java.nio.charset.StandardCharsets; +import java.util.Iterator; +import java.util.NoSuchElementException; +import java.util.Objects; +import org.restlet.data.Header; +import org.restlet.data.MediaType; +import org.restlet.util.Series; + +/** + * Reads the parts of a {@code multipart/*} stream (RFC 2046) as they arrive, for example a {@link + * MultipartStreamRepresentation} received by a client. Each part is returned once its closing + * delimiter is read. The preamble, the epilogue and transport padding are ignored. + * + *

Part sizes are bounded: a part whose headers and content exceed the limit fails the reader + * with an {@link IOException}, never truncated. + */ +public final class MultipartStreamReader + implements Iterator, AutoCloseable { + + private static final int MAX_HEADER_LINE = 8192; + + private final InputStream input; + + private final byte[] delimiter; + + private final int[] failure; + + private final long maxPartSize; + + private boolean started; + + private boolean finished; + + private MultiPartRepresentation.Part next; + + /** + * Creates a reader for a multipart entity, using the boundary of its media type. + * + * @param entity The multipart entity. + * @param maxPartSize The maximum size of a part, headers included, in bytes. + * @throws IOException If the entity has no boundary or can't be read. + */ + public MultipartStreamReader(Representation entity, long maxPartSize) throws IOException { + this( + entity.getStream(), + boundary(Objects.requireNonNull(entity, "entity").getMediaType()), + maxPartSize); + } + + /** + * Creates a reader. + * + * @param inputStream The multipart stream. + * @param boundary The boundary. + * @param maxPartSize The maximum size of a part, headers included, in bytes. + */ + public MultipartStreamReader(InputStream inputStream, String boundary, long maxPartSize) { + Objects.requireNonNull(inputStream, "inputStream"); + if (boundary == null || boundary.isEmpty()) { + throw new IllegalArgumentException("A boundary is required"); + } + if (maxPartSize <= 0) { + throw new IllegalArgumentException("The maximum part size must be positive"); + } + this.input = new BufferedInputStream(inputStream); + this.delimiter = ("\r\n--" + boundary).getBytes(StandardCharsets.US_ASCII); + this.failure = failureTable(delimiter); + this.maxPartSize = maxPartSize; + } + + private static String boundary(MediaType mediaType) throws IOException { + String boundary = mediaType == null ? null : MultiPartRepresentation.getBoundary(mediaType); + if (boundary == null) { + throw new IOException("The multipart media type has no boundary: " + mediaType); + } + return boundary; + } + + /** The Knuth-Morris-Pratt failure table of the delimiter. */ + private static int[] failureTable(byte[] pattern) { + int[] table = new int[pattern.length]; + int k = 0; + for (int i = 1; i < pattern.length; i++) { + while (k > 0 && pattern[i] != pattern[k]) { + k = table[k - 1]; + } + if (pattern[i] == pattern[k]) { + k++; + } + table[i] = k; + } + return table; + } + + /** + * Reads the next part, blocking as needed. + * + * @return The next part, or null after the close delimiter. + * @throws IOException If reading fails, the stream ends before the close delimiter, or a part + * exceeds the maximum size. + */ + public MultiPartRepresentation.Part readPart() throws IOException { + if (next != null) { + MultiPartRepresentation.Part result = next; + next = null; + return result; + } + if (finished) { + return null; + } + if (!started) { + started = true; + // The first delimiter may start the stream, without the leading line break. + if (!skipToDelimiter(true)) { + throw new IOException("No multipart delimiter found"); + } + } + + // After a delimiter: "--" closes the stream, otherwise padding then a line break. + int first = input.read(); + int second = input.read(); + if (first == '-' && second == '-') { + finished = true; + return null; + } + int c = second; + if (first != '\r' || second != '\n') { + c = first; + while (c == ' ' || c == '\t') { + c = second; + second = input.read(); + } + if (c != '\r' || second != '\n') { + throw new IOException("Invalid multipart delimiter line"); + } + } + + long[] size = {0}; + Series

headers = readHeaders(size); + byte[] content = readContent(size[0]); + return new MultiPartRepresentation.Part(null, null, headers, content(headers, content)); + } + + private Series
readHeaders(long[] size) throws IOException { + Series
headers = new Series<>(Header.class); + ByteArrayOutputStream line = new ByteArrayOutputStream(); + while (true) { + int c = input.read(); + if (c == -1) { + throw new IOException("The multipart stream ended in part headers"); + } + if (++size[0] > maxPartSize || line.size() > MAX_HEADER_LINE) { + throw new IOException("A multipart part exceeds " + maxPartSize + " bytes"); + } + if (c != '\n') { + line.write(c); + continue; + } + String text = line.toString(StandardCharsets.UTF_8); + line.reset(); + if (text.endsWith("\r")) { + text = text.substring(0, text.length() - 1); + } + if (text.isEmpty()) { + return headers; + } + int colon = text.indexOf(':'); + if (colon <= 0) { + throw new IOException("Invalid multipart part header: " + text); + } + headers.add( + new Header(text.substring(0, colon).trim(), text.substring(colon + 1).trim())); + } + } + + /** Reads the content up to the next delimiter, which is consumed. */ + private byte[] readContent(long headerSize) throws IOException { + ByteArrayOutputStream content = new ByteArrayOutputStream(); + int matched = 0; + long size = headerSize; + while (matched < delimiter.length) { + int c = input.read(); + if (c == -1) { + throw new IOException("The multipart stream ended in a part"); + } + while (matched > 0 && c != (delimiter[matched] & 0xFF)) { + // Bytes that looked like a delimiter are content after all. + int fallback = failure[matched - 1]; + content.write(delimiter, 0, matched - fallback); + matched = fallback; + } + if (c == (delimiter[matched] & 0xFF)) { + matched++; + } else { + content.write(c); + } + if (++size > maxPartSize + delimiter.length) { + throw new IOException("A multipart part exceeds " + maxPartSize + " bytes"); + } + } + if (content.size() + headerSize > maxPartSize) { + throw new IOException("A multipart part exceeds " + maxPartSize + " bytes"); + } + return content.toByteArray(); + } + + /** + * Skips the preamble up to the first delimiter. + * + * @param optionalLineBreak True if the delimiter may lack its leading line break. + * @return False if the stream ended first. + */ + private boolean skipToDelimiter(boolean optionalLineBreak) throws IOException { + int matched = optionalLineBreak ? 2 : 0; + long skipped = 0; + while (matched < delimiter.length) { + int c = input.read(); + if (c == -1) { + return false; + } + while (matched > 0 && c != (delimiter[matched] & 0xFF)) { + matched = failure[matched - 1]; + } + if (c == (delimiter[matched] & 0xFF)) { + matched++; + } + if (++skipped > maxPartSize) { + throw new IOException("The multipart preamble exceeds " + maxPartSize + " bytes"); + } + } + return true; + } + + private static Representation content(Series
headers, byte[] content) { + String contentType = headers.getFirstValue("Content-Type", true); + MediaType mediaType = contentType == null ? null : MediaType.valueOf(contentType); + return new ByteArrayRepresentation(content, mediaType); + } + + @Override + public boolean hasNext() { + if (next == null && !finished) { + try { + next = readPart(); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } + return next != null; + } + + @Override + public MultiPartRepresentation.Part next() { + if (!hasNext()) { + throw new NoSuchElementException(); + } + MultiPartRepresentation.Part result = next; + next = null; + return result; + } + + /** + * Closes the underlying stream. + * + * @throws IOException If closing fails. + */ + @Override + public void close() throws IOException { + finished = true; + input.close(); + } +} diff --git a/org.restlet/src/main/java/org/restlet/representation/MultipartStreamRepresentation.java b/org.restlet/src/main/java/org/restlet/representation/MultipartStreamRepresentation.java new file mode 100644 index 0000000000..98fad97812 --- /dev/null +++ b/org.restlet/src/main/java/org/restlet/representation/MultipartStreamRepresentation.java @@ -0,0 +1,216 @@ +/** + * Copyright 2005-2026 Qlik + *

+ * The content of this file is subject to the terms of the Apache 2.0 open + * source license available at https://www.opensource.org/licenses/apache-2.0 + *

+ * Restlet is a registered trademark of QlikTech International AB. + */ +package org.restlet.representation; + +import java.io.IOException; +import java.io.OutputStream; +import java.nio.charset.StandardCharsets; +import java.util.Objects; +import org.restlet.data.Header; +import org.restlet.data.MediaType; +import org.restlet.data.Parameter; +import org.restlet.util.Series; + +/** + * A streaming {@code multipart/*} representation (RFC 2046) whose parts are produced over time, + * such as {@code multipart/mixed} responses of incremental protocols. The writer runs on the + * connector's call thread; each part is flushed, followed by the next delimiter, before {@link + * PartSink#send(Series, byte[])} returns, so that readers can process it at once. The close + * delimiter is written when the writer returns. + * + * @see MultipartStreamReader + */ +public final class MultipartStreamRepresentation extends OutputRepresentation { + + private final PartWriter writer; + + private final String boundary; + + /** + * Creates a {@code multipart/mixed} stream with a generated boundary. + * + * @param writer Writes parts to the provided sink until the stream is complete. + */ + public MultipartStreamRepresentation(PartWriter writer) { + this(MediaType.MULTIPART_MIXED, null, writer); + } + + /** + * Creates a multipart stream. + * + * @param mediaType The multipart media type, for example {@link MediaType#MULTIPART_MIXED}. Its + * parameters are kept, and its boundary parameter replaced. + * @param boundary The boundary, or null to generate one. + * @param writer Writes parts to the provided sink until the stream is complete. + */ + public MultipartStreamRepresentation(MediaType mediaType, String boundary, PartWriter writer) { + Objects.requireNonNull(mediaType, "mediaType"); + Objects.requireNonNull(writer, "writer"); + if (!MediaType.MULTIPART_ALL.includes(mediaType, true)) { + throw new IllegalArgumentException("Not a multipart media type: " + mediaType); + } + String actualBoundary = + boundary == null ? MultiPartRepresentation.generateBoundary() : boundary; + if (actualBoundary.isEmpty() + || actualBoundary.length() > 70 + || actualBoundary.chars().anyMatch(c -> c < 0x20 || c > 0x7E)) { + throw new IllegalArgumentException("Invalid boundary: " + actualBoundary); + } + super(withBoundary(mediaType, actualBoundary)); + this.writer = writer; + this.boundary = actualBoundary; + setTransient(true); + } + + /** Sets the boundary parameter, keeping the other parameters of the media type. */ + private static MediaType withBoundary(MediaType mediaType, String boundary) { + Series parameters = new Series<>(Parameter.class); + for (Parameter parameter : mediaType.getParameters()) { + if (!"boundary".equalsIgnoreCase(parameter.getName())) { + parameters.add(parameter); + } + } + parameters.add(new Parameter("boundary", boundary)); + return new MediaType(mediaType, parameters); + } + + /** + * Returns the boundary. + * + * @return The boundary. + */ + public String getBoundary() { + return boundary; + } + + @Override + public void write(OutputStream outputStream) throws IOException { + try (PartSink sink = new PartSink(outputStream, boundary)) { + writer.write(sink); + } + } + + /** Writes parts for the lifetime of the response. */ + @FunctionalInterface + public interface PartWriter { + + /** + * Writes parts until the stream is complete. + * + * @param sink The part sink. + * @throws IOException If writing a part fails. + */ + void write(PartSink sink) throws IOException; + } + + /** A blocking sink for the parts of a {@link MultipartStreamRepresentation}. */ + public static final class PartSink implements AutoCloseable { + + private final OutputStream outputStream; + + private final byte[] delimiter; + + private boolean started; + + private boolean closed; + + private PartSink(OutputStream outputStream, String boundary) { + this.outputStream = Objects.requireNonNull(outputStream, "outputStream"); + this.delimiter = ("\r\n--" + boundary).getBytes(StandardCharsets.US_ASCII); + } + + /** + * Writes and flushes one part, followed by the next delimiter. + * + * @param headers The part headers, such as "Content-Type", or null. + * @param content The part content, which must not contain the delimiter. + * @throws IOException If the response stream has closed or writing fails. + * @throws IllegalArgumentException If a header is invalid or the content contains the + * delimiter, which would let it forge parts. + */ + public synchronized void send(Series

headers, byte[] content) throws IOException { + Objects.requireNonNull(content, "content"); + if (closed) { + throw new IOException("Part sink is closed"); + } + if (contains(content, delimiter)) { + throw new IllegalArgumentException("The part content contains the delimiter"); + } + StringBuilder head = new StringBuilder(); + if (headers != null) { + for (Header header : headers) { + String name = header.getName(); + String value = header.getValue(); + if (name.isEmpty() + || name.indexOf(':') >= 0 + || containsLineBreak(name) + || containsLineBreak(value)) { + throw new IllegalArgumentException("Invalid part header: " + name); + } + head.append(name).append(": ").append(value).append("\r\n"); + } + } + head.append("\r\n"); + if (!started) { + outputStream.write(delimiter); + started = true; + } + outputStream.write('\r'); + outputStream.write('\n'); + outputStream.write(head.toString().getBytes(StandardCharsets.UTF_8)); + outputStream.write(content); + outputStream.write(delimiter); + outputStream.flush(); + } + + /** + * Writes the close delimiter, without closing the connector-owned response stream. + * + * @throws IOException If writing fails. + */ + @Override + public synchronized void close() throws IOException { + if (!closed) { + closed = true; + if (!started) { + outputStream.write(delimiter); + } + outputStream.write("--\r\n".getBytes(StandardCharsets.US_ASCII)); + outputStream.flush(); + } + } + + /** Searches a pattern with the Knuth-Morris-Pratt algorithm. */ + private static boolean contains(byte[] content, byte[] pattern) { + int[] failure = new int[pattern.length]; + for (int i = 1, k = 0; i < pattern.length; i++) { + while (k > 0 && pattern[i] != pattern[k]) { + k = failure[k - 1]; + } + if (pattern[i] == pattern[k]) { + k++; + } + failure[i] = k; + } + for (int i = 0, matched = 0; i < content.length; i++) { + while (matched > 0 && content[i] != pattern[matched]) { + matched = failure[matched - 1]; + } + if (content[i] == pattern[matched] && ++matched == pattern.length) { + return true; + } + } + return false; + } + + private static boolean containsLineBreak(String value) { + return value.indexOf('\r') >= 0 || value.indexOf('\n') >= 0; + } + } +} diff --git a/org.restlet/src/test/java/org/restlet/representation/MultipartStreamTestCase.java b/org.restlet/src/test/java/org/restlet/representation/MultipartStreamTestCase.java new file mode 100644 index 0000000000..150d24a86a --- /dev/null +++ b/org.restlet/src/test/java/org/restlet/representation/MultipartStreamTestCase.java @@ -0,0 +1,273 @@ +/** + * Copyright 2005-2026 Qlik + *

+ * The content of this file is subject to the terms of the Apache 2.0 open + * source license available at https://www.opensource.org/licenses/apache-2.0 + *

+ * Restlet is a registered trademark of QlikTech International AB. + */ +package org.restlet.representation; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +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.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.UncheckedIOException; +import java.nio.charset.StandardCharsets; +import java.util.NoSuchElementException; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.Test; +import org.restlet.Client; +import org.restlet.Request; +import org.restlet.Response; +import org.restlet.Restlet; +import org.restlet.Server; +import org.restlet.data.Header; +import org.restlet.data.MediaType; +import org.restlet.data.Method; +import org.restlet.data.Parameter; +import org.restlet.data.Protocol; +import org.restlet.util.Series; + +class MultipartStreamTestCase { + + private static Series

headers(String... nameValues) { + Series
headers = new Series<>(Header.class); + for (int i = 0; i < nameValues.length; i += 2) { + headers.add(new Header(nameValues[i], nameValues[i + 1])); + } + return headers; + } + + private static byte[] write(MultipartStreamRepresentation representation) throws IOException { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + representation.write(out); + return out.toByteArray(); + } + + private static MultipartStreamReader reader(String content, String boundary, long max) { + return new MultipartStreamReader( + new ByteArrayInputStream(content.getBytes(StandardCharsets.UTF_8)), boundary, max); + } + + @Test + void writesFlushedPartsFollowedByTheirDelimiter() throws Exception { + MultipartStreamRepresentation representation = + new MultipartStreamRepresentation( + MediaType.MULTIPART_MIXED, + "b", + sink -> { + sink.send(headers("Content-Type", "application/json"), "{}".getBytes()); + sink.send(null, "two".getBytes()); + }); + assertEquals("b", representation.getBoundary()); + assertEquals("b", MultiPartRepresentation.getBoundary(representation.getMediaType())); + assertTrue(representation.isTransient()); + assertEquals( + "\r\n--b\r\nContent-Type: application/json\r\n\r\n{}\r\n--b\r\n\r\ntwo\r\n--b--\r\n", + new String(write(representation), StandardCharsets.UTF_8)); + + assertEquals( + "\r\n--b--\r\n", + new String( + write( + new MultipartStreamRepresentation( + MediaType.MULTIPART_MIXED, "b", sink -> {})), + StandardCharsets.UTF_8)); + } + + @Test + void roundTripsBinaryContentResemblingTheDelimiter() throws Exception { + byte[] tricky = + "x\r\n--bo\r\n-\r\n--boundar\r\n--boundarX\r".getBytes(StandardCharsets.UTF_8); + byte[] binary = new byte[256]; + for (int i = 0; i < binary.length; i++) { + binary[i] = (byte) i; + } + MultipartStreamRepresentation representation = + new MultipartStreamRepresentation( + MediaType.MULTIPART_MIXED, + "boundary", + sink -> { + sink.send(headers("X-Index", "1"), tricky); + sink.send(headers("Content-Type", "application/octet-stream"), binary); + sink.send(null, new byte[0]); + }); + try (MultipartStreamReader reader = + new MultipartStreamReader( + new ByteArrayInputStream(write(representation)), "boundary", 1024)) { + MultiPartRepresentation.Part first = reader.next(); + assertEquals("1", first.getHeaders().getFirstValue("X-Index")); + assertArrayEquals(tricky, first.getContent().getStream().readAllBytes()); + + MultiPartRepresentation.Part second = reader.next(); + assertEquals(MediaType.APPLICATION_OCTET_STREAM, second.getContent().getMediaType()); + assertArrayEquals(binary, second.getContent().getStream().readAllBytes()); + + assertEquals(0, reader.next().getContent().getStream().readAllBytes().length); + assertFalse(reader.hasNext()); + assertNull(reader.readPart()); + assertThrows(NoSuchElementException.class, reader::next); + } + } + + @Test + void ignoresPreambleEpilogueAndTransportPadding() throws Exception { + String content = + "preamble\r\n--b \t\r\nA: 1\r\n\r\nfirst\r\n--b\r\n\r\nsecond\r\n--b--\r\nepilogue"; + try (MultipartStreamReader reader = reader(content, "b", 100)) { + MultiPartRepresentation.Part first = reader.readPart(); + assertEquals("1", first.getHeaders().getFirstValue("A")); + assertEquals("first", first.getContentAsString(StandardCharsets.UTF_8)); + assertEquals("second", reader.readPart().getContentAsString(StandardCharsets.UTF_8)); + assertNull(reader.readPart()); + } + try (MultipartStreamReader reader = reader("--b\r\n\r\nx\r\n--b--", "b", 100)) { + assertEquals("x", reader.readPart().getContentAsString(StandardCharsets.UTF_8)); + assertNull(reader.readPart()); + } + } + + @Test + void rejectsMalformedAndOversizedStreams() { + assertThrows(IOException.class, () -> reader("no delimiter", "b", 100).readPart()); + assertThrows(IOException.class, () -> reader("--bX\r\n", "b", 100).readPart()); + assertThrows(IOException.class, () -> reader("--b\r\nA: 1", "b", 100).readPart()); + assertThrows(IOException.class, () -> reader("--b\r\nbad\r\n\r\n", "b", 100).readPart()); + assertThrows(IOException.class, () -> reader("--b\r\n\r\ntruncated", "b", 100).readPart()); + assertThrows( + IOException.class, + () -> reader("--b\r\n\r\n" + "x".repeat(200) + "\r\n--b--", "b", 100).readPart()); + assertThrows( + IOException.class, + () -> reader("x".repeat(200) + "\r\n--b--", "b", 100).readPart()); + assertThrows( + IOException.class, + () -> + reader("--b\r\nA: " + "x".repeat(200) + "\r\n\r\n\r\n--b--", "b", 100) + .readPart()); + assertThrows( + UncheckedIOException.class, + () -> reader("--b\r\n\r\ntruncated", "b", 100).hasNext()); + assertThrows( + IllegalArgumentException.class, + () -> new MultipartStreamReader(new ByteArrayInputStream(new byte[0]), "", 1)); + assertThrows( + IllegalArgumentException.class, + () -> new MultipartStreamReader(new ByteArrayInputStream(new byte[0]), "b", 0)); + assertThrows( + IOException.class, + () -> + new MultipartStreamReader( + new StringRepresentation("x", MediaType.MULTIPART_MIXED), 1)); + } + + @Test + void validatesTheRepresentation() { + assertThrows( + IllegalArgumentException.class, + () -> new MultipartStreamRepresentation(MediaType.TEXT_PLAIN, null, sink -> {})); + assertThrows( + IllegalArgumentException.class, + () -> + new MultipartStreamRepresentation( + MediaType.MULTIPART_MIXED, "a\r\nb", sink -> {})); + assertThrows( + IllegalArgumentException.class, + () -> + write( + new MultipartStreamRepresentation( + sink -> sink.send(headers("A", "1\r\n"), new byte[0])))); + assertThrows( + IllegalArgumentException.class, + () -> + write( + new MultipartStreamRepresentation( + MediaType.MULTIPART_MIXED, + "-", + sink -> sink.send(null, "a\r\n---b".getBytes())))); + assertThrows( + IOException.class, + () -> + write( + new MultipartStreamRepresentation( + sink -> { + sink.close(); + sink.send(null, new byte[0]); + }))); + Series original = new Series<>(Parameter.class); + original.add(new Parameter("deferSpec", "20220824")); + original.add(new Parameter("boundary", "old")); + MultipartStreamRepresentation parameters = + new MultipartStreamRepresentation( + new MediaType(MediaType.MULTIPART_MIXED, original), "-", sink -> {}); + assertEquals( + "20220824", parameters.getMediaType().getParameters().getFirstValue("deferSpec")); + assertEquals("-", MultiPartRepresentation.getBoundary(parameters.getMediaType())); + assertEquals(2, parameters.getMediaType().getParameters().size()); + MultipartStreamRepresentation generated = new MultipartStreamRepresentation(sink -> {}); + assertTrue(MediaType.MULTIPART_MIXED.equals(generated.getMediaType(), true)); + assertFalse(generated.getBoundary().isEmpty()); + } + + @Test + void partsArriveBeforeTheStreamEnds() throws Exception { + CountDownLatch firstSent = new CountDownLatch(1); + CountDownLatch continueWriting = new CountDownLatch(1); + Restlet restlet = + new Restlet() { + @Override + public void handle(Request request, Response response) { + response.setEntity( + new MultipartStreamRepresentation( + sink -> { + sink.send( + headers("Content-Type", "text/plain"), + "first".getBytes()); + firstSent.countDown(); + try { + if (!continueWriting.await(5, TimeUnit.SECONDS)) { + throw new IOException("Timed out"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException(e); + } + sink.send(null, "second".getBytes()); + })); + } + }; + Server server = new Server(Protocol.HTTP, 0, restlet); + server.start(); + Client client = new Client(Protocol.HTTP); + client.start(); + try { + Response response = + client.handle( + new Request(Method.GET, "http://localhost:" + server.getActualPort())); + assertTrue(MediaType.MULTIPART_MIXED.equals(response.getEntity().getMediaType(), true)); + try (MultipartStreamReader reader = + new MultipartStreamReader(response.getEntity(), 1024)) { + assertTrue(firstSent.await(5, TimeUnit.SECONDS)); + MultiPartRepresentation.Part first = reader.readPart(); + assertEquals("first", first.getContentAsString(StandardCharsets.UTF_8)); + assertTrue(MediaType.TEXT_PLAIN.equals(first.getContent().getMediaType(), true)); + continueWriting.countDown(); + assertEquals( + "second", reader.readPart().getContentAsString(StandardCharsets.UTF_8)); + assertNull(reader.readPart()); + } + } finally { + continueWriting.countDown(); + client.stop(); + server.stop(); + } + } +} From 96de18c71f186ccec6976dd6b2f480086a7c26e0 Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Fri, 30 Oct 2026 15:48:08 -0400 Subject: [PATCH 2/6] fix: harden the streaming multipart boundary and size checks - Drop the old boundary from the media type name, not only from its parameters, so the Content-Type advertises a single boundary. - Reject part content starting with the boundary, which would form a delimiter after the header line break and forge parts. - Validate the boundary and size limit before acquiring the entity stream, which would otherwise be left open. - Look up the boundary parameter case-insensitively. - Avoid overflowing a Long.MAX_VALUE part size limit. - Don't count the opening delimiter in the preamble size. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Jerome Louvel <374450+jlouvel@users.noreply.github.com> --- .../representation/MultipartStreamReader.java | 32 +++++--- .../MultipartStreamRepresentation.java | 18 ++++- .../MultipartStreamTestCase.java | 73 ++++++++++++++++++- 3 files changed, 109 insertions(+), 14 deletions(-) diff --git a/org.restlet/src/main/java/org/restlet/representation/MultipartStreamReader.java b/org.restlet/src/main/java/org/restlet/representation/MultipartStreamReader.java index 49f3dd32a2..c29361d376 100644 --- a/org.restlet/src/main/java/org/restlet/representation/MultipartStreamReader.java +++ b/org.restlet/src/main/java/org/restlet/representation/MultipartStreamReader.java @@ -56,10 +56,10 @@ public final class MultipartStreamReader * @throws IOException If the entity has no boundary or can't be read. */ public MultipartStreamReader(Representation entity, long maxPartSize) throws IOException { - this( - entity.getStream(), - boundary(Objects.requireNonNull(entity, "entity").getMediaType()), - maxPartSize); + // Validate before acquiring the stream, which would otherwise be left open. + String boundary = boundary(Objects.requireNonNull(entity, "entity").getMediaType()); + checkMaxPartSize(maxPartSize); + this(entity.getStream(), boundary, maxPartSize); } /** @@ -74,18 +74,26 @@ public MultipartStreamReader(InputStream inputStream, String boundary, long maxP if (boundary == null || boundary.isEmpty()) { throw new IllegalArgumentException("A boundary is required"); } - if (maxPartSize <= 0) { - throw new IllegalArgumentException("The maximum part size must be positive"); - } + checkMaxPartSize(maxPartSize); this.input = new BufferedInputStream(inputStream); this.delimiter = ("\r\n--" + boundary).getBytes(StandardCharsets.US_ASCII); this.failure = failureTable(delimiter); this.maxPartSize = maxPartSize; } + private static void checkMaxPartSize(long maxPartSize) { + if (maxPartSize <= 0) { + throw new IllegalArgumentException("The maximum part size must be positive"); + } + } + private static String boundary(MediaType mediaType) throws IOException { - String boundary = mediaType == null ? null : MultiPartRepresentation.getBoundary(mediaType); - if (boundary == null) { + // Parameter names are case-insensitive. + String boundary = + mediaType == null + ? null + : mediaType.getParameters().getFirstValue("boundary", true); + if (boundary == null || boundary.isEmpty()) { throw new IOException("The multipart media type has no boundary: " + mediaType); } return boundary; @@ -209,7 +217,8 @@ private byte[] readContent(long headerSize) throws IOException { } else { content.write(c); } - if (++size > maxPartSize + delimiter.length) { + // Subtract the delimiter allowance rather than add it, which could overflow the limit. + if (++size - delimiter.length > maxPartSize) { throw new IOException("A multipart part exceeds " + maxPartSize + " bytes"); } } @@ -239,7 +248,8 @@ private boolean skipToDelimiter(boolean optionalLineBreak) throws IOException { if (c == (delimiter[matched] & 0xFF)) { matched++; } - if (++skipped > maxPartSize) { + // The bytes currently matching the delimiter aren't part of the preamble. + if (++skipped - matched > maxPartSize) { throw new IOException("The multipart preamble exceeds " + maxPartSize + " bytes"); } } diff --git a/org.restlet/src/main/java/org/restlet/representation/MultipartStreamRepresentation.java b/org.restlet/src/main/java/org/restlet/representation/MultipartStreamRepresentation.java index 98fad97812..05b799786f 100644 --- a/org.restlet/src/main/java/org/restlet/representation/MultipartStreamRepresentation.java +++ b/org.restlet/src/main/java/org/restlet/representation/MultipartStreamRepresentation.java @@ -11,6 +11,7 @@ import java.io.IOException; import java.io.OutputStream; import java.nio.charset.StandardCharsets; +import java.util.Arrays; import java.util.Objects; import org.restlet.data.Header; import org.restlet.data.MediaType; @@ -77,7 +78,11 @@ private static MediaType withBoundary(MediaType mediaType, String boundary) { } } parameters.add(new Parameter("boundary", boundary)); - return new MediaType(mediaType, parameters); + // The name of the original type may carry its parameters, including the old boundary. + return new MediaType( + mediaType.getMainType() + "/" + mediaType.getSubType(), + parameters, + mediaType.getDescription()); } /** @@ -125,6 +130,13 @@ private PartSink(OutputStream outputStream, String boundary) { this.delimiter = ("\r\n--" + boundary).getBytes(StandardCharsets.US_ASCII); } + /** Indicates whether the content starts with the delimiter without its line break. */ + private boolean startsWithDashBoundary(byte[] content) { + int length = delimiter.length - 2; + return content.length >= length + && Arrays.equals(content, 0, length, delimiter, 2, delimiter.length); + } + /** * Writes and flushes one part, followed by the next delimiter. * @@ -139,7 +151,9 @@ public synchronized void send(Series
headers, byte[] content) throws IOE if (closed) { throw new IOException("Part sink is closed"); } - if (contains(content, delimiter)) { + // The line break ending the headers precedes the content, so content starting with + // the boundary would also form a delimiter. + if (contains(content, delimiter) || startsWithDashBoundary(content)) { throw new IllegalArgumentException("The part content contains the delimiter"); } StringBuilder head = new StringBuilder(); diff --git a/org.restlet/src/test/java/org/restlet/representation/MultipartStreamTestCase.java b/org.restlet/src/test/java/org/restlet/representation/MultipartStreamTestCase.java index 150d24a86a..28e572148b 100644 --- a/org.restlet/src/test/java/org/restlet/representation/MultipartStreamTestCase.java +++ b/org.restlet/src/test/java/org/restlet/representation/MultipartStreamTestCase.java @@ -18,11 +18,14 @@ import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; import java.io.IOException; +import java.io.InputStream; import java.io.UncheckedIOException; import java.nio.charset.StandardCharsets; import java.util.NoSuchElementException; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Function; import org.junit.jupiter.api.Test; import org.restlet.Client; import org.restlet.Request; @@ -170,7 +173,51 @@ void rejectsMalformedAndOversizedStreams() { } @Test - void validatesTheRepresentation() { + void validatesTheEntityBeforeAcquiringItsStream() { + AtomicBoolean acquired = new AtomicBoolean(); + Function entity = + mediaType -> + new StringRepresentation("", mediaType) { + @Override + public InputStream getStream() throws IOException { + acquired.set(true); + return super.getStream(); + } + }; + assertThrows( + IOException.class, + () -> new MultipartStreamReader(entity.apply(MediaType.MULTIPART_MIXED), 1)); + assertThrows( + IllegalArgumentException.class, + () -> + new MultipartStreamReader( + entity.apply( + new MediaType(MediaType.MULTIPART_MIXED, "boundary", "b")), + 0)); + assertFalse(acquired.get()); + } + + @Test + void acceptsCaseInsensitiveBoundaryAndLargeLimits() throws Exception { + Representation entity = + new StringRepresentation( + "--b\r\n\r\nx\r\n--b--", + new MediaType(MediaType.MULTIPART_MIXED, "Boundary", "b")); + try (MultipartStreamReader reader = new MultipartStreamReader(entity, Long.MAX_VALUE)) { + assertEquals("x", reader.readPart().getContentAsString(StandardCharsets.UTF_8)); + assertNull(reader.readPart()); + } + + // The opening delimiter is longer than the limit, but isn't part of the preamble. + try (MultipartStreamReader reader = + reader("--boundary\r\n\r\nx\r\n--boundary--", "boundary", 8)) { + assertEquals("x", reader.readPart().getContentAsString(StandardCharsets.UTF_8)); + assertNull(reader.readPart()); + } + } + + @Test + void validatesTheRepresentation() throws Exception { assertThrows( IllegalArgumentException.class, () -> new MultipartStreamRepresentation(MediaType.TEXT_PLAIN, null, sink -> {})); @@ -212,6 +259,30 @@ void validatesTheRepresentation() { "20220824", parameters.getMediaType().getParameters().getFirstValue("deferSpec")); assertEquals("-", MultiPartRepresentation.getBoundary(parameters.getMediaType())); assertEquals(2, parameters.getMediaType().getParameters().size()); + assertEquals( + "multipart/mixed; deferSpec=20220824; boundary=-", + parameters.getMediaType().getName()); + // Content starting with the boundary would forge a part after the header line break. + assertThrows( + IllegalArgumentException.class, + () -> + write( + new MultipartStreamRepresentation( + MediaType.MULTIPART_MIXED, + "b", + sink -> + sink.send( + null, + "--b\r\nA: 1\r\n\r\nforged".getBytes())))); + assertEquals( + "\r\n--bb\r\n\r\n--b\r\n--bb--\r\n", + new String( + write( + new MultipartStreamRepresentation( + MediaType.MULTIPART_MIXED, + "bb", + sink -> sink.send(null, "--b".getBytes()))), + StandardCharsets.UTF_8)); MultipartStreamRepresentation generated = new MultipartStreamRepresentation(sink -> {}); assertTrue(MediaType.MULTIPART_MIXED.equals(generated.getMediaType(), true)); assertFalse(generated.getBoundary().isEmpty()); From 20bfee5d47ab95c60cd9a26ee19d0b896ffc5392 Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Sat, 31 Oct 2026 02:39:25 -0400 Subject: [PATCH 3/6] fix: don't close a multipart stream whose writer failed Writing the close delimiter after a writer failure made a truncated stream look complete to readers. Only write it when the writer returns. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Jerome Louvel <374450+jlouvel@users.noreply.github.com> --- .../MultipartStreamRepresentation.java | 16 +++++++-- .../MultipartStreamTestCase.java | 36 +++++++++++++++++++ 2 files changed, 50 insertions(+), 2 deletions(-) diff --git a/org.restlet/src/main/java/org/restlet/representation/MultipartStreamRepresentation.java b/org.restlet/src/main/java/org/restlet/representation/MultipartStreamRepresentation.java index 05b799786f..573b2bee3c 100644 --- a/org.restlet/src/main/java/org/restlet/representation/MultipartStreamRepresentation.java +++ b/org.restlet/src/main/java/org/restlet/representation/MultipartStreamRepresentation.java @@ -23,7 +23,8 @@ * such as {@code multipart/mixed} responses of incremental protocols. The writer runs on the * connector's call thread; each part is flushed, followed by the next delimiter, before {@link * PartSink#send(Series, byte[])} returns, so that readers can process it at once. The close - * delimiter is written when the writer returns. + * delimiter is written when the writer returns, but not when it throws, so that readers see a + * truncated stream rather than a complete one. * * @see MultipartStreamReader */ @@ -96,9 +97,15 @@ public String getBoundary() { @Override public void write(OutputStream outputStream) throws IOException { - try (PartSink sink = new PartSink(outputStream, boundary)) { + PartSink sink = new PartSink(outputStream, boundary); + try { writer.write(sink); + } catch (IOException | RuntimeException | Error e) { + // Don't write the close delimiter, which would make a truncated stream look complete. + sink.abort(); + throw e; } + sink.close(); } /** Writes parts for the lifetime of the response. */ @@ -200,6 +207,11 @@ public synchronized void close() throws IOException { } } + /** Closes the sink after a failure, without writing the close delimiter. */ + private synchronized void abort() { + closed = true; + } + /** Searches a pattern with the Knuth-Morris-Pratt algorithm. */ private static boolean contains(byte[] content, byte[] pattern) { int[] failure = new int[pattern.length]; diff --git a/org.restlet/src/test/java/org/restlet/representation/MultipartStreamTestCase.java b/org.restlet/src/test/java/org/restlet/representation/MultipartStreamTestCase.java index 28e572148b..aa4f592fd7 100644 --- a/org.restlet/src/test/java/org/restlet/representation/MultipartStreamTestCase.java +++ b/org.restlet/src/test/java/org/restlet/representation/MultipartStreamTestCase.java @@ -172,6 +172,42 @@ void rejectsMalformedAndOversizedStreams() { new StringRepresentation("x", MediaType.MULTIPART_MIXED), 1)); } + @Test + void writerFailuresLeaveTheStreamTruncated() { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + MultipartStreamRepresentation.PartSink[] sinks = {null}; + MultipartStreamRepresentation representation = + new MultipartStreamRepresentation( + MediaType.MULTIPART_MIXED, + "b", + sink -> { + sinks[0] = sink; + sink.send(null, "one".getBytes()); + throw new IOException("failed"); + }); + IOException failure = assertThrows(IOException.class, () -> representation.write(out)); + assertEquals("failed", failure.getMessage()); + assertEquals("\r\n--b\r\n\r\none\r\n--b", out.toString(StandardCharsets.UTF_8)); + assertThrows(IOException.class, () -> sinks[0].send(null, new byte[0])); + assertThrows( + UncheckedIOException.class, + () -> + new MultipartStreamReader( + new ByteArrayInputStream(out.toByteArray()), "b", 100) + .forEachRemaining(part -> {})); + + ByteArrayOutputStream runtime = new ByteArrayOutputStream(); + assertThrows( + IllegalStateException.class, + () -> + new MultipartStreamRepresentation( + sink -> { + throw new IllegalStateException(); + }) + .write(runtime)); + assertFalse(runtime.toString(StandardCharsets.UTF_8).contains("--\r\n")); + } + @Test void validatesTheEntityBeforeAcquiringItsStream() { AtomicBoolean acquired = new AtomicBoolean(); From 1300f7d3158ce33db5b50e8c40db170b0322797b Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Thu, 29 Oct 2026 23:24:28 -0400 Subject: [PATCH 4/6] feat: add experimental GraphQL incremental delivery (G5) @defer results over multipart/mixed, on the streaming multipart support of core. Experimental and off by default: the GraphQL incremental delivery RFC is not finalized, and graphql-java 26.1 implements @defer only (no @stream). - Server: setIncrementalDeliveryEnabled(true). When a query or mutation accepts multipart/mixed, graphql-java's incremental support is enabled for that execution. An IncrementalExecutionResult is streamed as parts (boundary "-", deferSpec=20220824, graphql-java's format): the initial result, then each DelayedIncrementalPartialResult, requested one at a time and bounded by maxQueuedBytes. A failure aborts the response, so that clients can't mistake a partial result for a complete one. Without multipart/mixed, @defer is executed inline as before. - Client: setIncrementalDeliveryEnabled(true) accepts multipart/mixed. execute() returns graphql-java's IncrementalExecutionResult, with its incremental payloads rebuilt as DeferPayload and StreamPayload, and an item publisher that reads the following parts on demand (single subscriber, released on completion, failure or cancellation, parts bounded by maxResponseBytes). - GraphqlResultQueue now serializes any result type. Refs #1588 Co-Authored-By: Claude Opus 5.5 Signed-off-by: Jerome Louvel <374450+jlouvel@users.noreply.github.com> --- changes.md | 5 +- .../ext/graphql/GraphqlClientAdapter.java | 44 +++ .../graphql/GraphqlIncrementalPublisher.java | 294 +++++++++++++++ .../ext/graphql/GraphqlServerAdapter.java | 54 ++- .../internal/GraphqlIncrementalStream.java | 99 +++++ .../graphql/internal/GraphqlResultQueue.java | 25 +- .../graphql/GraphqlIncrementalTestCase.java | 348 ++++++++++++++++++ 7 files changed, 863 insertions(+), 6 deletions(-) create mode 100644 org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlIncrementalPublisher.java create mode 100644 org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/internal/GraphqlIncrementalStream.java create mode 100644 org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlIncrementalTestCase.java diff --git a/changes.md b/changes.md index b1beebf515..943d9fdc1a 100644 --- a/changes.md +++ b/changes.md @@ -162,7 +162,10 @@ Changes log the server adapter also accepts `graphql-transport-ws` WebSocket upgrades on its own path (HTTP/1.1 only), with several operations per connection, a `connection_init` timeout and the protocol's close codes; `org.restlet.ext.websocket` is an optional dependency, - loaded only then. Issue #1581. + loaded only then. Incremental delivery of `@defer` results over `multipart/mixed` is + available on both sides when enabled (experimental, off by default): the client + returns graphql-java's `IncrementalExecutionResult`, whose item publisher reads the + delayed results. Issue #1581. - Added the `NettyConnectionUpgrade` engine hook: a Restlet returning a "101 Switching Protocols" response with this response attribute takes over the HTTP/1.1 connection (`NettyUpgradeContext`), which the WebSocket extension relies on. diff --git a/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlClientAdapter.java b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlClientAdapter.java index aaa7928aab..a893286f07 100644 --- a/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlClientAdapter.java +++ b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlClientAdapter.java @@ -77,6 +77,9 @@ public class GraphqlClientAdapter { private static final int EXCERPT_BYTES = 1024; + private static final MediaType INCREMENTAL = + new MediaType(MediaType.MULTIPART_MIXED, "deferSpec", "20220824"); + private enum State { NEW, STARTED, @@ -99,6 +102,8 @@ private enum State { private volatile State state = State.NEW; + private volatile boolean incrementalDeliveryEnabled; + private final Set subscriptions = ConcurrentHashMap.newKeySet(); @@ -250,6 +255,33 @@ public void setRequestCustomizer(Consumer requestCustomizer) { this.requestCustomizer = requestCustomizer; } + /** + * Indicates if incremental delivery is accepted. + * + * @return True if incremental delivery is accepted. + */ + public boolean isIncrementalDeliveryEnabled() { + return incrementalDeliveryEnabled; + } + + /** + * Indicates if incremental delivery ({@code @defer}) is accepted. Experimental and off by + * default: the GraphQL incremental delivery RFC is not finalized. + * + *

When enabled, {@code execute} accepts {@code multipart/mixed} responses in graphql-java's + * format ({@code deferSpec=20220824}). Such a response is returned as an {@link + * graphql.incremental.IncrementalExecutionResult} with the initial result. When it has more + * results, its item publisher reads them from the response, which stays open until the + * publisher completes, fails or is cancelled: subscribe to it, or cancel its subscription. Each + * part is bounded by {@link #getMaxResponseBytes()}. + * + * @param incrementalDeliveryEnabled True to accept incremental delivery. + */ + public void setIncrementalDeliveryEnabled(boolean incrementalDeliveryEnabled) { + checkNew(); + this.incrementalDeliveryEnabled = incrementalDeliveryEnabled; + } + /** * Validates and freezes the configuration. * @@ -531,8 +563,20 @@ private ExecutionResult send( request.getClientInfo() .getAcceptedMediaTypes() .add(new Preference<>(MediaType.APPLICATION_JSON, 0.9F)); + if (incrementalDeliveryEnabled) { + request.getClientInfo().getAcceptedMediaTypes().add(new Preference<>(INCREMENTAL, 1F)); + } Response response = client.handle(request); + Representation multipart = response.getEntity(); + if (incrementalDeliveryEnabled + && response.getStatus().isSuccess() + && multipart != null + && multipart.getMediaType() != null + && MediaType.MULTIPART_MIXED.equals(multipart.getMediaType(), true)) { + // Releases the response itself, once the last result is read. + return GraphqlIncrementalPublisher.read(response, maxResponseBytes); + } try { return toResult(response); } finally { diff --git a/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlIncrementalPublisher.java b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlIncrementalPublisher.java new file mode 100644 index 0000000000..2cfb2b542b --- /dev/null +++ b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlIncrementalPublisher.java @@ -0,0 +1,294 @@ +/** + * Copyright 2005-2026 Qlik + *

+ * The content of this file is subject to the terms of the Apache 2.0 open + * source license available at https://www.opensource.org/licenses/apache-2.0 + *

+ * Restlet is a registered trademark of QlikTech International AB. + */ +package org.restlet.ext.graphql; + +import graphql.ExecutionResult; +import graphql.GraphQLError; +import graphql.incremental.DeferPayload; +import graphql.incremental.DelayedIncrementalPartialResult; +import graphql.incremental.DelayedIncrementalPartialResultImpl; +import graphql.incremental.IncrementalExecutionResult; +import graphql.incremental.IncrementalExecutionResultImpl; +import graphql.incremental.IncrementalPayload; +import graphql.incremental.StreamPayload; +import java.io.IOException; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.concurrent.atomic.AtomicBoolean; +import org.reactivestreams.Publisher; +import org.reactivestreams.Subscriber; +import org.reactivestreams.Subscription; +import org.restlet.Response; +import org.restlet.data.Status; +import org.restlet.ext.graphql.internal.GraphqlJson; +import org.restlet.representation.MultiPartRepresentation; +import org.restlet.representation.MultipartStreamReader; + +/** + * Reads an incremental response: the first part becomes an {@link IncrementalExecutionResult} whose + * item publisher reads the following parts. The publisher accepts a single subscriber; the response + * is released when it completes, fails or is cancelled. + */ +final class GraphqlIncrementalPublisher implements Publisher { + + private final Response response; + + private final MultipartStreamReader reader; + + private final AtomicBoolean subscribed = new AtomicBoolean(); + + private GraphqlIncrementalPublisher(Response response, MultipartStreamReader reader) { + this.response = response; + this.reader = reader; + } + + /** + * Reads the initial result of an incremental response. The response is released unless more + * results follow. + * + * @param response The response, with a multipart entity. + * @param maxPartBytes The maximum size of a part. + * @return The initial result. + * @throws GraphqlTransportException If the response is not a valid incremental response. + */ + static ExecutionResult read(Response response, long maxPartBytes) + throws GraphqlTransportException { + Status status = response.getStatus(); + boolean keep = false; + try { + MultipartStreamReader reader = + new MultipartStreamReader(response.getEntity(), maxPartBytes); + Map initial = part(reader.readPart(), status); + boolean hasNext = Boolean.TRUE.equals(initial.get("hasNext")); + IncrementalExecutionResultImpl.Builder result = + IncrementalExecutionResultImpl.fromExecutionResult( + ExecutionResult.fromSpecification(initial)) + .hasNext(hasNext) + .incremental(payloads(initial.get("incremental"))); + if (hasNext) { + result.incrementalItemPublisher(new GraphqlIncrementalPublisher(response, reader)); + keep = true; + } + return result.build(); + } catch (IOException e) { + throw new GraphqlTransportException( + "Unable to read the GraphQL incremental response", status, null, e); + } finally { + if (!keep) { + response.getEntity().release(); + } + } + } + + private static Map part(MultiPartRepresentation.Part part, Status status) + throws IOException { + if (part == null) { + throw new GraphqlTransportException( + "The GraphQL incremental response ended early", status, null, null); + } + Map result = null; + try { + result = + GraphqlJson.asObject( + GraphqlJson.parse(part.getContent().getStream().readAllBytes())); + } catch (GraphqlJson.InvalidJsonException _) { + // Reported below + } + if (result == null) { + throw new GraphqlTransportException( + "Invalid GraphQL incremental part", status, null, null); + } + return result; + } + + private static List payloads(Object incremental) { + List payloads = new ArrayList<>(); + if (incremental instanceof List items) { + for (Object item : items) { + Map map = GraphqlJson.asObject(item); + if (map != null) { + payloads.add(payload(map)); + } + } + } + return payloads; + } + + @SuppressWarnings("unchecked") + private static IncrementalPayload payload(Map map) { + List errors = + map.get("errors") instanceof List list + ? list.stream() + .map(GraphqlJson::asObject) + .filter(Objects::nonNull) + .map(GraphQLError::fromSpecification) + .toList() + : List.of(); + List path = + map.get("path") instanceof List list ? (List) list : List.of(); + Map extensions = GraphqlJson.asObject(map.get("extensions")); + String label = map.get("label") instanceof String string ? string : null; + if (map.get("items") instanceof List items) { + StreamPayload.Builder builder = + StreamPayload.newStreamedItem().items((List) items); + builder.path(path).label(label).errors(errors); + if (extensions != null) { + builder.extensions(new LinkedHashMap(extensions)); + } + return builder.build(); + } + DeferPayload.Builder builder = DeferPayload.newDeferredItem().data(map.get("data")); + builder.path(path).label(label).errors(errors); + if (extensions != null) { + builder.extensions(new LinkedHashMap(extensions)); + } + return builder.build(); + } + + @Override + public void subscribe(Subscriber subscriber) { + Objects.requireNonNull(subscriber, "subscriber"); + if (!subscribed.compareAndSet(false, true)) { + subscriber.onSubscribe( + new Subscription() { + @Override + public void request(long n) {} + + @Override + public void cancel() {} + }); + subscriber.onError( + new IllegalStateException("The incremental results have one subscriber")); + return; + } + subscriber.onSubscribe(new PartSubscription(subscriber)); + } + + /** Reads the parts on demand, on a virtual thread that emits all the signals. */ + private final class PartSubscription implements Subscription { + + private final Subscriber subscriber; + + private long demand; + + private boolean started; + + private boolean invalidDemand; + + private volatile boolean cancelled; + + PartSubscription(Subscriber subscriber) { + this.subscriber = subscriber; + } + + @Override + public void request(long n) { + synchronized (this) { + if (cancelled) { + return; + } + if (n <= 0) { + invalidDemand = true; + } else { + demand = demand + n < 0 ? Long.MAX_VALUE : demand + n; + } + notifyAll(); + if (started) { + return; + } + started = true; + } + Thread.ofVirtual().name("restlet-graphql-incremental").start(this::run); + } + + @Override + public void cancel() { + synchronized (this) { + if (cancelled) { + return; + } + cancelled = true; + notifyAll(); + } + response.abort(); + } + + private void run() { + try { + while (true) { + Map part = part(reader.readPart(), response.getStatus()); + boolean hasNext = Boolean.TRUE.equals(part.get("hasNext")); + DelayedIncrementalPartialResultImpl.Builder result = + DelayedIncrementalPartialResultImpl.newIncrementalExecutionResult() + .hasNext(hasNext) + .incrementalItems(payloads(part.get("incremental"))); + Map extensions = GraphqlJson.asObject(part.get("extensions")); + if (extensions != null) { + result.extensions(Map.copyOf(extensions)); + } + if (!awaitDemand()) { + return; + } + subscriber.onNext(result.build()); + if (!hasNext) { + if (!cancelled) { + subscriber.onComplete(); + } + return; + } + } + } catch (IOException | RuntimeException e) { + if (!cancelled) { + cancelled = true; + subscriber.onError( + e instanceof GraphqlTransportException transport + ? transport + : new GraphqlTransportException( + "Unable to read the GraphQL incremental response", + response.getStatus(), + null, + e)); + } + } finally { + response.getEntity().release(); + } + } + + /** Waits for demand. Returns false if cancelled meanwhile. */ + private boolean awaitDemand() { + synchronized (this) { + while (!cancelled && demand == 0 && !invalidDemand) { + try { + wait(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return false; + } + } + if (cancelled) { + return false; + } + if (invalidDemand) { + cancelled = true; + subscriber.onError( + new IllegalArgumentException( + "The requested number of results must be positive")); + return false; + } + if (demand != Long.MAX_VALUE) { + demand--; + } + return true; + } + } + } +} diff --git a/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlServerAdapter.java b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlServerAdapter.java index 7e25a974bc..4d331c9c7b 100644 --- a/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlServerAdapter.java +++ b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlServerAdapter.java @@ -13,6 +13,7 @@ import graphql.GraphQL; import graphql.GraphQLError; import graphql.GraphqlErrorBuilder; +import graphql.incremental.IncrementalExecutionResult; import graphql.language.Document; import graphql.language.OperationDefinition; import graphql.parser.InvalidSyntaxException; @@ -43,10 +44,12 @@ import org.restlet.data.Preference; import org.restlet.data.Status; import org.restlet.ext.graphql.internal.GraphqlEventStream; +import org.restlet.ext.graphql.internal.GraphqlIncrementalStream; import org.restlet.ext.graphql.internal.GraphqlJson; import org.restlet.ext.graphql.internal.GraphqlMediaTypes; import org.restlet.representation.ByteArrayRepresentation; import org.restlet.representation.EventStreamRepresentation; +import org.restlet.representation.MultipartStreamRepresentation; import org.restlet.representation.Representation; /** @@ -141,6 +144,8 @@ public class GraphqlServerAdapter extends Restlet { private volatile Duration connectionInitTimeout = DEFAULT_CONNECTION_INIT_TIMEOUT; + private volatile boolean incrementalDeliveryEnabled; + private volatile GraphqlUpgradeSupport upgradeSupport; /** Constructor. */ @@ -374,6 +379,33 @@ public void setConnectionInitTimeout(Duration connectionInitTimeout) { this.connectionInitTimeout = connectionInitTimeout; } + /** + * Indicates if incremental delivery is enabled. + * + * @return True if incremental delivery is enabled. + */ + public boolean isIncrementalDeliveryEnabled() { + return incrementalDeliveryEnabled; + } + + /** + * Indicates if incremental delivery ({@code @defer} and {@code @stream}) is enabled. + * Experimental and off by default: the GraphQL incremental delivery RFC is not finalized. + * + *

When enabled, and a query or mutation accepts {@code multipart/mixed}, graphql-java's + * incremental support is enabled for that execution. An incremental result is then streamed as + * {@code multipart/mixed} parts (boundary "-", {@code deferSpec=20220824}): the initial result, + * then each delayed result, in graphql-java's format. Delayed results are requested one at a + * time and bounded by {@link #getMaxQueuedBytes()}. Without {@code multipart/mixed} in the + * {@code Accept} header, {@code @defer} and {@code @stream} are not executed incrementally. + * + * @param incrementalDeliveryEnabled True to enable incremental delivery. + */ + public void setIncrementalDeliveryEnabled(boolean incrementalDeliveryEnabled) { + checkNotStarted(); + this.incrementalDeliveryEnabled = incrementalDeliveryEnabled; + } + /** Returns the counter of active subscriptions, shared by all transports. */ AtomicInteger getActiveSubscriptionCounter() { return activeSubscriptions; @@ -450,9 +482,14 @@ public void handle(Request request, Response response) { subscriptionTransports.contains(GraphqlTransport.SSE) && GraphqlMediaTypes.quality(accepted, MediaType.TEXT_EVENT_STREAM, true) > 0F; + boolean multipart = + incrementalDeliveryEnabled + && !sse + && GraphqlMediaTypes.quality(accepted, MediaType.MULTIPART_MIXED, true) + > 0F; MediaType responseType = GraphqlMediaTypes.negotiate(accepted); if (responseType == null) { - if (!sse) { + if (!sse && !multipart) { response.setStatus(Status.CLIENT_ERROR_NOT_ACCEPTABLE); return; } @@ -520,8 +557,21 @@ public void handle(Request request, Response response) { return; } + if (multipart && !subscription) { + GraphQL.unusualConfiguration(executionInput) + .incrementalSupport() + .enableIncrementalSupport(true); + } ExecutionResult result = graphQL.execute(executionInput); - if (!sse || !result.isDataPresent()) { + if (multipart && result instanceof IncrementalExecutionResult incremental) { + response.setEntity( + new MultipartStreamRepresentation( + GraphqlIncrementalStream.MEDIA_TYPE, + GraphqlIncrementalStream.BOUNDARY, + new GraphqlIncrementalStream( + incremental, maxQueuedBytes, getLogger()))); + response.setStatus(Status.SUCCESS_OK); + } else if (!sse || !result.isDataPresent()) { // Without an event stream, or when execution didn't start: a single JSON result. write(response, responseType, result, sse); } else if (result.getData() instanceof Publisher publisher) { diff --git a/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/internal/GraphqlIncrementalStream.java b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/internal/GraphqlIncrementalStream.java new file mode 100644 index 0000000000..aecb3cbf81 --- /dev/null +++ b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/internal/GraphqlIncrementalStream.java @@ -0,0 +1,99 @@ +/** + * Copyright 2005-2026 Qlik + *

+ * The content of this file is subject to the terms of the Apache 2.0 open + * source license available at https://www.opensource.org/licenses/apache-2.0 + *

+ * Restlet is a registered trademark of QlikTech International AB. + */ +package org.restlet.ext.graphql.internal; + +import graphql.incremental.DelayedIncrementalPartialResult; +import graphql.incremental.IncrementalExecutionResult; +import java.io.IOException; +import java.io.InterruptedIOException; +import java.nio.charset.StandardCharsets; +import java.util.logging.Level; +import java.util.logging.Logger; +import org.restlet.data.Header; +import org.restlet.data.MediaType; +import org.restlet.representation.MultipartStreamRepresentation.PartSink; +import org.restlet.representation.MultipartStreamRepresentation.PartWriter; +import org.restlet.util.Series; + +/** + * Writes an incremental result as {@code multipart/mixed} parts: the initial result, then each + * delayed result as it completes. Delayed results are requested one at a time, once the previous + * part is written. If they fail, the response is aborted, so that clients don't mistake a partial + * result for a complete one. + */ +public final class GraphqlIncrementalStream implements PartWriter { + + /** The media type of incremental responses, in the format of graphql-java. */ + public static final MediaType MEDIA_TYPE = + new MediaType(MediaType.MULTIPART_MIXED, "deferSpec", "20220824"); + + /** The boundary of incremental responses, as in the GraphQL incremental delivery RFC. */ + public static final String BOUNDARY = "-"; + + private final IncrementalExecutionResult initial; + + private final long maxQueuedBytes; + + private final Logger logger; + + /** + * Constructor. + * + * @param initial The initial result. + * @param maxQueuedBytes The maximum number of bytes of delayed results not yet written. + * @param logger The logger of failures. + */ + public GraphqlIncrementalStream( + IncrementalExecutionResult initial, long maxQueuedBytes, Logger logger) { + this.initial = initial; + this.maxQueuedBytes = maxQueuedBytes; + this.logger = logger; + } + + private static Series

headers() { + Series
headers = new Series<>(Header.class); + headers.add(new Header("Content-Type", "application/json; charset=utf-8")); + return headers; + } + + @Override + public void write(PartSink sink) throws IOException { + sink.send(headers(), GraphqlJson.write(initial.toSpecification())); + if (!initial.hasNext()) { + return; + } + + GraphqlResultQueue queue = + new GraphqlResultQueue( + maxQueuedBytes, + item -> ((DelayedIncrementalPartialResult) item).toSpecification()); + try { + initial.getIncrementalItemPublisher().subscribe(queue); + while (true) { + GraphqlResultQueue.Signal signal = queue.take(); + if (signal instanceof GraphqlResultQueue.Next next) { + sink.send(headers(), next.json().getBytes(StandardCharsets.UTF_8)); + queue.request(); + } else if (signal instanceof GraphqlResultQueue.Failure failure) { + logger.log( + Level.WARNING, "GraphQL incremental delivery failed", failure.cause()); + throw new IOException( + "GraphQL incremental delivery failed: " + failure.message()); + } else { + return; + } + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new InterruptedIOException("Incremental delivery interrupted"); + } finally { + queue.cancel(); + } + } +} diff --git a/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/internal/GraphqlResultQueue.java b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/internal/GraphqlResultQueue.java index 5ff1d76746..9420601ab1 100644 --- a/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/internal/GraphqlResultQueue.java +++ b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/internal/GraphqlResultQueue.java @@ -10,7 +10,9 @@ import graphql.ExecutionResult; import java.nio.charset.StandardCharsets; +import java.util.Map; import java.util.concurrent.LinkedBlockingQueue; +import java.util.function.Function; import org.reactivestreams.Subscriber; import org.reactivestreams.Subscription; @@ -19,8 +21,11 @@ * writes them. One result is requested at a time: the writer calls {@link #request()} once a result * is written. Results received without being requested are queued up to a byte limit, beyond which * the subscription is cancelled. + * + *

Items are {@link ExecutionResult} values by default, or other results with their own + * serialization, such as the delayed results of incremental delivery. */ -public final class GraphqlResultQueue implements Subscriber { +public final class GraphqlResultQueue implements Subscriber { /** A signal taken from the queue. */ public sealed interface Signal permits Next, Failure, Completed, Cancelled {} @@ -51,6 +56,8 @@ public record Cancelled() implements Signal {} private final long maxQueuedBytes; + private final Function> specification; + private long queuedBytes; private volatile Subscription subscription; @@ -65,7 +72,19 @@ public record Cancelled() implements Signal {} * @param maxQueuedBytes The maximum number of bytes of results received but not yet taken. */ public GraphqlResultQueue(long maxQueuedBytes) { + this(maxQueuedBytes, item -> ((ExecutionResult) item).toSpecification()); + } + + /** + * Constructor. + * + * @param maxQueuedBytes The maximum number of bytes of results received but not yet taken. + * @param specification Returns the specification map of an item, serialized to JSON. + */ + public GraphqlResultQueue( + long maxQueuedBytes, Function> specification) { this.maxQueuedBytes = maxQueuedBytes; + this.specification = specification; } @Override @@ -79,8 +98,8 @@ public void onSubscribe(Subscription subscription) { } @Override - public void onNext(ExecutionResult item) { - String json = GraphqlJson.writeString(item.toSpecification()); + public void onNext(Object item) { + String json = GraphqlJson.writeString(specification.apply(item)); long size = json.getBytes(StandardCharsets.UTF_8).length; synchronized (this) { if (terminated) { diff --git a/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlIncrementalTestCase.java b/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlIncrementalTestCase.java new file mode 100644 index 0000000000..e1607d29c1 --- /dev/null +++ b/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlIncrementalTestCase.java @@ -0,0 +1,348 @@ +/** + * Copyright 2005-2026 Qlik + *

+ * The content of this file is subject to the terms of the Apache 2.0 open + * source license available at https://www.opensource.org/licenses/apache-2.0 + *

+ * Restlet is a registered trademark of QlikTech International AB. + */ +package org.restlet.ext.graphql; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import graphql.ExecutionResult; +import graphql.GraphQL; +import graphql.incremental.DeferPayload; +import graphql.incremental.DelayedIncrementalPartialResult; +import graphql.incremental.IncrementalExecutionResult; +import graphql.schema.DataFetcher; +import graphql.schema.idl.RuntimeWiring; +import graphql.schema.idl.SchemaGenerator; +import graphql.schema.idl.SchemaParser; +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.Map; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.reactivestreams.Subscriber; +import org.reactivestreams.Subscription; +import org.restlet.Client; +import org.restlet.Component; +import org.restlet.Request; +import org.restlet.Response; +import org.restlet.data.MediaType; +import org.restlet.data.Method; +import org.restlet.data.Preference; +import org.restlet.data.Protocol; +import org.restlet.data.Status; +import org.restlet.ext.graphql.internal.GraphqlJson; +import org.restlet.representation.MultiPartRepresentation; +import org.restlet.representation.MultipartStreamReader; +import org.restlet.representation.StringRepresentation; + +/** Incremental delivery of {@code @defer} results over {@code multipart/mixed}. */ +class GraphqlIncrementalTestCase { + + private static final String SDL = + """ + directive @defer(if: Boolean, label: String) on FRAGMENT_SPREAD | INLINE_FRAGMENT + type Query { + fast: String + slow: String + broken: String + } + """; + + private static final String DEFERRED = "{ fast ... @defer(label: \"d\") { slow } }"; + + private final CountDownLatch release = new CountDownLatch(1); + + private Component component; + + private Client client; + + private String base; + + @BeforeEach + void setUp() throws Exception { + DataFetcher slow = + env -> + CompletableFuture.supplyAsync( + () -> { + try { + release.await(5, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return "slow"; + }); + DataFetcher broken = + env -> { + throw new IllegalStateException("broken"); + }; + RuntimeWiring wiring = + RuntimeWiring.newRuntimeWiring() + .type( + "Query", + type -> + type.dataFetcher("fast", env -> "fast") + .dataFetcher("slow", slow) + .dataFetcher("broken", broken)) + .build(); + GraphQL graphQL = + GraphQL.newGraphQL( + new SchemaGenerator() + .makeExecutableSchema( + new SchemaParser().parse(SDL), wiring)) + .build(); + + component = new Component(); + component.getServers().add(Protocol.HTTP, 0); + GraphqlServerAdapter incremental = new GraphqlServerAdapter(); + incremental.setGraphQL(graphQL); + incremental.setIncrementalDeliveryEnabled(true); + component.getDefaultHost().attach("/graphql", incremental); + + GraphqlServerAdapter limited = new GraphqlServerAdapter(); + limited.setGraphQL(graphQL); + limited.setIncrementalDeliveryEnabled(true); + limited.setMaxQueuedBytes(10); + component.getDefaultHost().attach("/limited", limited); + + GraphqlServerAdapter disabled = new GraphqlServerAdapter(); + disabled.setGraphQL(graphQL); + component.getDefaultHost().attach("/disabled", disabled); + component.start(); + base = "http://localhost:" + component.getServers().get(0).getActualPort(); + client = new Client(Protocol.HTTP); + client.start(); + } + + @AfterEach + void tearDown() throws Exception { + release.countDown(); + client.stop(); + component.stop(); + } + + private Response post(String path, String query, MediaType... accepted) { + Request request = new Request(Method.POST, base + path); + request.setEntity( + new StringRepresentation( + GraphqlJson.writeString(Map.of("query", query)), + MediaType.APPLICATION_JSON)); + for (MediaType type : accepted) { + request.getClientInfo().getAcceptedMediaTypes().add(new Preference<>(type)); + } + return client.handle(request); + } + + private static Map json(MultiPartRepresentation.Part part) throws Exception { + assertEquals( + "application/json; charset=utf-8", part.getHeaders().getFirstValue("Content-Type")); + return GraphqlJson.asObject( + GraphqlJson.parse(part.getContentAsString(StandardCharsets.UTF_8))); + } + + private GraphqlClientAdapter graphql(String path) { + GraphqlClientAdapter graphql = new GraphqlClientAdapter(); + graphql.setClient(client); + graphql.setEndpoint(base + path); + graphql.setIncrementalDeliveryEnabled(true); + graphql.start(); + return graphql; + } + + @Test + void initialResultIsSentBeforeDeferredOnes() throws Exception { + Response response = post("/graphql", DEFERRED, MediaType.MULTIPART_MIXED); + assertEquals(Status.SUCCESS_OK, response.getStatus()); + MediaType type = response.getEntity().getMediaType(); + assertTrue(MediaType.MULTIPART_MIXED.equals(type, true)); + assertEquals("-", MultiPartRepresentation.getBoundary(type)); + assertEquals("20220824", type.getParameters().getFirstValue("deferSpec")); + + try (MultipartStreamReader reader = new MultipartStreamReader(response.getEntity(), 4096)) { + Map initial = json(reader.readPart()); + assertEquals(Map.of("fast", "fast"), initial.get("data")); + assertEquals(true, initial.get("hasNext")); + + release.countDown(); + Map deferred = json(reader.readPart()); + assertEquals(false, deferred.get("hasNext")); + Map payload = + GraphqlJson.asObject(((List) deferred.get("incremental")).get(0)); + assertEquals(Map.of("slow", "slow"), payload.get("data")); + assertEquals("d", payload.get("label")); + assertEquals(List.of(), payload.get("path")); + assertNull(reader.readPart()); + } + } + + @Test + void withoutMultipartDeferIsNotIncremental() throws Exception { + release.countDown(); + Response json = post("/graphql", DEFERRED, MediaType.APPLICATION_JSON); + assertEquals( + "{\"data\":{\"fast\":\"fast\",\"slow\":\"slow\"}}", json.getEntity().getText()); + + Response disabled = + post("/disabled", DEFERRED, MediaType.MULTIPART_MIXED, MediaType.APPLICATION_JSON); + assertTrue(MediaType.APPLICATION_JSON.equals(disabled.getEntity().getMediaType(), true)); + assertEquals( + Status.CLIENT_ERROR_NOT_ACCEPTABLE, + post("/disabled", DEFERRED, MediaType.MULTIPART_MIXED).getStatus()); + + Response plain = post("/graphql", "{ fast }", MediaType.MULTIPART_MIXED); + assertTrue(MediaType.APPLICATION_JSON.equals(plain.getEntity().getMediaType(), true)); + } + + @Test + void clientPublishesDeferredResults() throws Exception { + GraphqlClientAdapter graphql = graphql("/graphql"); + assertTrue(graphql.isIncrementalDeliveryEnabled()); + ExecutionResult result = graphql.execute(DEFERRED, null, null); + IncrementalExecutionResult incremental = + assertInstanceOf(IncrementalExecutionResult.class, result); + assertTrue(incremental.hasNext()); + assertEquals(Map.of("fast", "fast"), incremental.getData()); + + TestSubscriber subscriber = new TestSubscriber(); + incremental.getIncrementalItemPublisher().subscribe(subscriber); + subscriber.subscription.request(1); + release.countDown(); + DelayedIncrementalPartialResult delayed = subscriber.next(); + assertFalse(delayed.hasNext()); + DeferPayload payload = + assertInstanceOf(DeferPayload.class, delayed.getIncremental().get(0)); + assertEquals(Map.of("slow", "slow"), payload.getData()); + assertEquals("d", payload.getLabel()); + assertTrue(subscriber.done.await(5, TimeUnit.SECONDS)); + assertNull(subscriber.error.get()); + + TestSubscriber second = new TestSubscriber(); + incremental.getIncrementalItemPublisher().subscribe(second); + assertInstanceOf(IllegalStateException.class, second.error.get()); + } + + @Test + void deferredErrorsArePayloadErrors() throws Exception { + GraphqlClientAdapter graphql = graphql("/graphql"); + IncrementalExecutionResult incremental = + (IncrementalExecutionResult) + graphql.execute("{ fast ... @defer { broken } }", null, null); + TestSubscriber subscriber = new TestSubscriber(); + incremental.getIncrementalItemPublisher().subscribe(subscriber); + subscriber.subscription.request(Long.MAX_VALUE); + DeferPayload payload = (DeferPayload) subscriber.next().getIncremental().get(0); + assertEquals(1, payload.getErrors().size()); + assertEquals(List.of("broken"), payload.getErrors().get(0).getPath()); + assertTrue(subscriber.done.await(5, TimeUnit.SECONDS)); + } + + @Test + void plainResultsStayPlain() throws Exception { + GraphqlClientAdapter graphql = graphql("/graphql"); + ExecutionResult result = graphql.execute("{ fast }", null, null); + assertFalse(result instanceof IncrementalExecutionResult); + assertEquals(Map.of("fast", "fast"), result.getData()); + } + + @Test + void cancellationAndInvalidDemand() throws Exception { + GraphqlClientAdapter graphql = graphql("/graphql"); + IncrementalExecutionResult cancelled = + (IncrementalExecutionResult) graphql.execute(DEFERRED, null, null); + TestSubscriber subscriber = new TestSubscriber(); + cancelled.getIncrementalItemPublisher().subscribe(subscriber); + subscriber.subscription.request(1); + subscriber.subscription.cancel(); + subscriber.subscription.cancel(); + release.countDown(); + assertFalse(subscriber.done.await(200, TimeUnit.MILLISECONDS)); + + IncrementalExecutionResult invalid = + (IncrementalExecutionResult) graphql.execute(DEFERRED, null, null); + TestSubscriber negative = new TestSubscriber(); + invalid.getIncrementalItemPublisher().subscribe(negative); + negative.subscription.request(0); + assertTrue(negative.done.await(5, TimeUnit.SECONDS)); + assertInstanceOf(IllegalArgumentException.class, negative.error.get()); + } + + @Test + void failedDeliveryIsATransportError() throws Exception { + release.countDown(); + GraphqlClientAdapter graphql = graphql("/limited"); + IncrementalExecutionResult incremental = + (IncrementalExecutionResult) graphql.execute(DEFERRED, null, null); + TestSubscriber subscriber = new TestSubscriber(); + incremental.getIncrementalItemPublisher().subscribe(subscriber); + subscriber.subscription.request(1); + assertTrue(subscriber.done.await(5, TimeUnit.SECONDS)); + assertInstanceOf(GraphqlTransportException.class, subscriber.error.get()); + + GraphqlServerAdapter adapter = new GraphqlServerAdapter(); + assertFalse(adapter.isIncrementalDeliveryEnabled()); + adapter.setGraphQL( + GraphQL.newGraphQL( + GraphqlTestSchemas.schema( + new java.util.concurrent.atomic.AtomicInteger())) + .build()); + adapter.start(); + assertThrows( + IllegalStateException.class, () -> adapter.setIncrementalDeliveryEnabled(true)); + } + + /** Collects the signals of a subscription. */ + private static final class TestSubscriber + implements Subscriber { + + final BlockingQueue items = new LinkedBlockingQueue<>(); + + final CountDownLatch done = new CountDownLatch(1); + + final AtomicReference error = new AtomicReference<>(); + + volatile Subscription subscription; + + @Override + public void onSubscribe(Subscription subscription) { + this.subscription = subscription; + } + + @Override + public void onNext(DelayedIncrementalPartialResult item) { + items.add(item); + } + + @Override + public void onError(Throwable throwable) { + error.set(throwable); + done.countDown(); + } + + @Override + public void onComplete() { + done.countDown(); + } + + DelayedIncrementalPartialResult next() throws InterruptedException { + DelayedIncrementalPartialResult item = items.poll(5, TimeUnit.SECONDS); + assertTrue(item != null, "Timed out waiting for a result"); + return item; + } + } +} From 5950d5cf4c5981f858e7c680b1bd300572581a15 Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Sat, 31 Oct 2026 02:29:49 -0400 Subject: [PATCH 5/6] fix: harden the GraphQL incremental result publisher Signal non-positive demand without waiting for the next part, keep null extension values, require the close delimiter after the last result, and close the multipart reader on completion, failure and cancellation, including cancellation before any demand. Co-Authored-By: Claude Opus 5.5 --- .../graphql/GraphqlIncrementalPublisher.java | 102 ++++++++++++++---- .../graphql/GraphqlIncrementalTestCase.java | 74 +++++++++++++ 2 files changed, 153 insertions(+), 23 deletions(-) diff --git a/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlIncrementalPublisher.java b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlIncrementalPublisher.java index 2cfb2b542b..9f02b6d91b 100644 --- a/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlIncrementalPublisher.java +++ b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlIncrementalPublisher.java @@ -64,9 +64,9 @@ static ExecutionResult read(Response response, long maxPartBytes) throws GraphqlTransportException { Status status = response.getStatus(); boolean keep = false; + MultipartStreamReader reader = null; try { - MultipartStreamReader reader = - new MultipartStreamReader(response.getEntity(), maxPartBytes); + reader = new MultipartStreamReader(response.getEntity(), maxPartBytes); Map initial = part(reader.readPart(), status); boolean hasNext = Boolean.TRUE.equals(initial.get("hasNext")); IncrementalExecutionResultImpl.Builder result = @@ -77,6 +77,8 @@ static ExecutionResult read(Response response, long maxPartBytes) if (hasNext) { result.incrementalItemPublisher(new GraphqlIncrementalPublisher(response, reader)); keep = true; + } else { + checkEnd(reader, status); } return result.build(); } catch (IOException e) { @@ -84,11 +86,34 @@ static ExecutionResult read(Response response, long maxPartBytes) "Unable to read the GraphQL incremental response", status, null, e); } finally { if (!keep) { - response.getEntity().release(); + release(response, reader); } } } + /** Checks that the last part is followed by the close delimiter. */ + private static void checkEnd(MultipartStreamReader reader, Status status) throws IOException { + if (reader.readPart() != null) { + throw new GraphqlTransportException( + "The GraphQL incremental response has parts after the last result", + status, + null, + null); + } + } + + /** Closes the reader, which owns the stream of the entity, and releases the entity. */ + private static void release(Response response, MultipartStreamReader reader) { + if (reader != null) { + try { + reader.close(); + } catch (IOException _) { + // Released anyway + } + } + response.getEntity().release(); + } + private static Map part(MultiPartRepresentation.Part part, Status status) throws IOException { if (part == null) { @@ -182,7 +207,7 @@ private final class PartSubscription implements Subscription { private boolean started; - private boolean invalidDemand; + private volatile boolean invalidDemand; private volatile boolean cancelled; @@ -192,34 +217,59 @@ private final class PartSubscription implements Subscription { @Override public void request(long n) { + boolean signal = false; + boolean start = false; synchronized (this) { if (cancelled) { return; } if (n <= 0) { invalidDemand = true; + // Without a reader thread, signal from here and prevent one from starting. + signal = !started; + if (signal) { + cancelled = true; + } } else { demand = demand + n < 0 ? Long.MAX_VALUE : demand + n; - } - notifyAll(); - if (started) { - return; + start = !started; } started = true; + notifyAll(); + } + if (signal) { + release(response, reader); + subscriber.onError(invalidDemandError()); + } else if (start) { + Thread.ofVirtual().name("restlet-graphql-incremental").start(this::run); + } else if (n <= 0) { + // Unblocks a pending read, then the reader thread signals the error. + response.abort(); } - Thread.ofVirtual().name("restlet-graphql-incremental").start(this::run); } @Override public void cancel() { + boolean idle; synchronized (this) { if (cancelled) { return; } cancelled = true; + // Without a reader thread to clean up, prevent one from starting. + idle = !started; + started = true; notifyAll(); } - response.abort(); + if (idle) { + release(response, reader); + } else { + response.abort(); + } + } + + private IllegalArgumentException invalidDemandError() { + return new IllegalArgumentException("The requested number of results must be positive"); } private void run() { @@ -233,7 +283,11 @@ private void run() { .incrementalItems(payloads(part.get("incremental"))); Map extensions = GraphqlJson.asObject(part.get("extensions")); if (extensions != null) { - result.extensions(Map.copyOf(extensions)); + // Extensions can have null values. + result.extensions(new LinkedHashMap(extensions)); + } + if (!hasNext) { + checkEnd(reader, response.getStatus()); } if (!awaitDemand()) { return; @@ -249,20 +303,24 @@ private void run() { } catch (IOException | RuntimeException e) { if (!cancelled) { cancelled = true; - subscriber.onError( - e instanceof GraphqlTransportException transport - ? transport - : new GraphqlTransportException( - "Unable to read the GraphQL incremental response", - response.getStatus(), - null, - e)); + subscriber.onError(invalidDemand ? invalidDemandError() : failure(e)); } } finally { - response.getEntity().release(); + release(response, reader); } } + /** Returns the error of a failed read, unless the read was aborted by invalid demand. */ + private GraphqlTransportException failure(Exception e) { + return e instanceof GraphqlTransportException transport + ? transport + : new GraphqlTransportException( + "Unable to read the GraphQL incremental response", + response.getStatus(), + null, + e); + } + /** Waits for demand. Returns false if cancelled meanwhile. */ private boolean awaitDemand() { synchronized (this) { @@ -279,9 +337,7 @@ private boolean awaitDemand() { } if (invalidDemand) { cancelled = true; - subscriber.onError( - new IllegalArgumentException( - "The requested number of results must be positive")); + subscriber.onError(invalidDemandError()); return false; } if (demand != Long.MAX_VALUE) { diff --git a/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlIncrementalTestCase.java b/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlIncrementalTestCase.java index e1607d29c1..230eb61a2c 100644 --- a/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlIncrementalTestCase.java +++ b/org.restlet.ext.graphql/src/test/java/org/restlet/ext/graphql/GraphqlIncrementalTestCase.java @@ -42,6 +42,7 @@ import org.restlet.Component; import org.restlet.Request; import org.restlet.Response; +import org.restlet.Restlet; import org.restlet.data.MediaType; import org.restlet.data.Method; import org.restlet.data.Preference; @@ -69,6 +70,9 @@ class GraphqlIncrementalTestCase { private final CountDownLatch release = new CountDownLatch(1); + /** The body of the raw multipart responses, with the "b" boundary. */ + private final AtomicReference raw = new AtomicReference<>(); + private Component component; private Client client; @@ -124,6 +128,19 @@ void setUp() throws Exception { GraphqlServerAdapter disabled = new GraphqlServerAdapter(); disabled.setGraphQL(graphQL); component.getDefaultHost().attach("/disabled", disabled); + component + .getDefaultHost() + .attach( + "/raw", + new Restlet() { + @Override + public void handle(Request request, Response response) { + response.setEntity( + new StringRepresentation( + raw.get(), + MediaType.valueOf("multipart/mixed; boundary=b"))); + } + }); component.start(); base = "http://localhost:" + component.getServers().get(0).getActualPort(); client = new Client(Protocol.HTTP); @@ -282,6 +299,63 @@ void cancellationAndInvalidDemand() throws Exception { assertInstanceOf(IllegalArgumentException.class, negative.error.get()); } + @Test + void invalidDemandDoesNotWaitForParts() throws Exception { + GraphqlClientAdapter graphql = graphql("/graphql"); + IncrementalExecutionResult incremental = + (IncrementalExecutionResult) graphql.execute(DEFERRED, null, null); + TestSubscriber subscriber = new TestSubscriber(); + incremental.getIncrementalItemPublisher().subscribe(subscriber); + // The deferred part is held back while the reader waits for it. + subscriber.subscription.request(1); + Thread.sleep(100); + subscriber.subscription.request(-1); + assertTrue(subscriber.done.await(2, TimeUnit.SECONDS)); + assertInstanceOf(IllegalArgumentException.class, subscriber.error.get()); + } + + @Test + void rawResponsesAreChecked() throws Exception { + String initial = + "--b\r\nContent-Type: application/json\r\n\r\n" + + "{\"data\":{\"fast\":\"fast\"},\"hasNext\":true}\r\n"; + String last = + "--b\r\nContent-Type: application/json\r\n\r\n" + + "{\"incremental\":[{\"data\":{\"slow\":\"slow\"},\"path\":[]}]," + + "\"extensions\":{\"trace\":null},\"hasNext\":false}\r\n"; + GraphqlClientAdapter graphql = graphql("/raw"); + + // Null extension values are kept. + raw.set(initial + last + "--b--\r\n"); + TestSubscriber complete = subscribe(graphql); + DelayedIncrementalPartialResult delayed = complete.next(); + assertTrue(delayed.getExtensions().containsKey("trace")); + assertTrue(complete.done.await(5, TimeUnit.SECONDS)); + assertNull(complete.error.get()); + + // A missing close delimiter or extra parts are errors. + for (String end : List.of("--b", initial + "--b--\r\n")) { + raw.set(initial + last + end); + TestSubscriber invalid = subscribe(graphql); + assertTrue(invalid.done.await(5, TimeUnit.SECONDS), end); + assertInstanceOf(GraphqlTransportException.class, invalid.error.get(), end); + } + raw.set( + "--b\r\nContent-Type: application/json\r\n\r\n" + + "{\"data\":{\"fast\":\"fast\"},\"hasNext\":false}\r\n--b"); + assertThrows( + GraphqlTransportException.class, () -> graphql.execute("{ fast }", null, null)); + } + + private static TestSubscriber subscribe(GraphqlClientAdapter graphql) throws Exception { + IncrementalExecutionResult incremental = + (IncrementalExecutionResult) graphql.execute(DEFERRED, null, null); + TestSubscriber subscriber = new TestSubscriber(); + incremental.getIncrementalItemPublisher().subscribe(subscriber); + subscriber.subscription.request(Long.MAX_VALUE); + return subscriber; + } + @Test void failedDeliveryIsATransportError() throws Exception { release.countDown(); From cfdefaf49a74be9df29edf6f4c970db0a7a8023b Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Sun, 1 Nov 2026 09:58:25 -0500 Subject: [PATCH 6/6] fix: read the body of a GraphQL request refused with 406 A POST answered with 406 Not Acceptable left its body unread, so the Netty server closed the connection after the response, sometimes after the client had already reused it for its next request (1001 Communication Error, seen in GraphqlIncrementalTestCase). Discard bodies within the request size limit before answering. Co-Authored-By: Claude Opus 5.5 --- .../ext/graphql/GraphqlServerAdapter.java | 28 +++++++++++++++++++ 1 file changed, 28 insertions(+) diff --git a/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlServerAdapter.java b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlServerAdapter.java index 4d331c9c7b..9939b5a52c 100644 --- a/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlServerAdapter.java +++ b/org.restlet.ext.graphql/src/main/java/org/restlet/ext/graphql/GraphqlServerAdapter.java @@ -32,6 +32,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.BiConsumer; +import java.util.logging.Level; import org.reactivestreams.Publisher; import org.restlet.Context; import org.restlet.Request; @@ -490,6 +491,7 @@ public void handle(Request request, Response response) { MediaType responseType = GraphqlMediaTypes.negotiate(accepted); if (responseType == null) { if (!sse && !multipart) { + discardBody(request); response.setStatus(Status.CLIENT_ERROR_NOT_ACCEPTABLE); return; } @@ -709,6 +711,32 @@ private static Object parseParameter(Form form, String name) throws RequestExcep } } + /** + * Reads and discards the body of a request answered without reading it, up to {@link + * #getMaxRequestBytes()}. Otherwise the server connector closes the connection after the + * response, possibly once the client has already reused it for its next request. + */ + private void discardBody(Request request) { + Representation entity = request.getEntity(); + if (entity == null || !entity.isAvailable() || entity.getSize() > maxRequestBytes) { + return; + } + try (InputStream in = entity.getStream()) { + if (in != null) { + byte[] buffer = new byte[8192]; + long remaining = maxRequestBytes; + int read; + while (remaining > 0 + && (read = in.read(buffer, 0, (int) Math.min(buffer.length, remaining))) + != -1) { + remaining -= read; + } + } + } catch (IOException e) { + getLogger().log(Level.FINE, "Unable to discard a GraphQL request body", e); + } + } + private Map decodeBody( Request request, Response response, MediaType responseType) throws RequestException { Representation entity = request.getEntity();