diff --git a/sdk/core/azure-core/CHANGELOG.md b/sdk/core/azure-core/CHANGELOG.md index a3faaedbe612..dc023a9c713e 100644 --- a/sdk/core/azure-core/CHANGELOG.md +++ b/sdk/core/azure-core/CHANGELOG.md @@ -4,6 +4,11 @@ ### Features Added +- Added generic `ServerSentEvent`, `CloseableIterableStream`, and `ServerSentEventStreams` APIs for typed, + incrementally decoded server-sent event streams. Synchronous SSE streams are closeable to release an unfinished + response body. Optional service-defined terminal-event predicates stop processing after delivering a matching event; + streams otherwise complete normally on HTTP 204 or response-body EOF. + ### Breaking Changes ### Bugs Fixed @@ -20,6 +25,7 @@ ### Bugs Fixed +- Fixed synchronous streaming of non-replayable `BinaryData` response bodies. - Fixed a bug where retrying requests with bodies created from a mark/reset-capable `InputStream` could fail because the stream was closed between retry attempts. ([#49650](https://github.com/Azure/azure-sdk-for-java/pull/49650)) - Fixed GraalVM native-image compilation with SLF4J 2 by no longer forcing Azure Core logging and logging provider classes to initialize at image build time. ([#49844](https://github.com/Azure/azure-sdk-for-java/issues/49844)) diff --git a/sdk/core/azure-core/src/main/java/com/azure/core/http/ServerSentEvent.java b/sdk/core/azure-core/src/main/java/com/azure/core/http/ServerSentEvent.java new file mode 100644 index 000000000000..9355a0669636 --- /dev/null +++ b/sdk/core/azure-core/src/main/java/com/azure/core/http/ServerSentEvent.java @@ -0,0 +1,97 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.core.http; + +import com.azure.core.annotation.Immutable; +import com.azure.core.implementation.util.ServerSentEventHelper; + +import java.time.Duration; + +/** + * Represents a server-sent event with a typed data payload. + * + *

An emitted server-sent event contains data and may expose an identifier, event name, comment, and retry interval. + * The identifier and retry interval represent the effective stream state when the event was dispatched, including + * values inherited from earlier metadata-only blocks.

+ * + *

The identifier and retry interval are protocol metadata only. Azure Core does not reconnect or replay the + * request. Metadata-only updates received after the latest emitted event aren't exposed as an additional event.

+ * + * @param The type of the event data. + * @see + * Parsing an event stream + */ +@Immutable +public final class ServerSentEvent { + private final String id; + private final String event; + private final T data; + private final String comment; + private final Duration retryAfter; + + static { + ServerSentEventHelper.setAccessor(new ServerSentEventHelper.ServerSentEventAccessor() { + @Override + public ServerSentEvent create(String id, String event, U data, String comment, Duration retryAfter) { + return new ServerSentEvent<>(id, event, data, comment, retryAfter); + } + }); + } + + private ServerSentEvent(String id, String event, T data, String comment, Duration retryAfter) { + this.id = id; + this.event = event; + this.data = data; + this.comment = comment; + this.retryAfter = retryAfter; + } + + /** + * Gets the effective last-event identifier when this event was dispatched. + * + * @return The effective last-event identifier, {@code null} if no valid {@code id} field was received before this + * event, or an empty string if an empty {@code id} field reset the identifier. + */ + public String getId() { + return id; + } + + /** + * Gets the event name. + * + * @return The event name, or {@code message} if no non-empty {@code event} field was specified. + */ + public String getEvent() { + return event; + } + + /** + * Gets the event data. + * + * @return The event data, or {@code null} if event data wasn't specified. + */ + public T getData() { + return data; + } + + /** + * Gets the event comment. + * + * @return The event comment, or {@code null} if it wasn't specified. + */ + public String getComment() { + return comment; + } + + /** + * Gets the effective retry interval when this event was dispatched. + * + * @return The latest valid retry interval received before this event, or {@code null} if no valid + * {@code retry} field was received. + */ + public Duration getRetryAfter() { + return retryAfter; + } + +} diff --git a/sdk/core/azure-core/src/main/java/com/azure/core/http/ServerSentEventStreams.java b/sdk/core/azure-core/src/main/java/com/azure/core/http/ServerSentEventStreams.java new file mode 100644 index 000000000000..bad28e1d26a2 --- /dev/null +++ b/sdk/core/azure-core/src/main/java/com/azure/core/http/ServerSentEventStreams.java @@ -0,0 +1,102 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.core.http; + +import com.azure.core.http.rest.Response; +import com.azure.core.implementation.util.ServerSentEventStream; +import com.azure.core.util.BinaryData; +import com.azure.core.util.CloseableIterableStream; +import reactor.core.publisher.Flux; + +import java.util.function.BiFunction; +import java.util.function.Predicate; + +/** + * Consumes a single HTTP response as a server-sent event stream. + * + *

The response body owns the physical response and closes it when consumption ends.

+ * + *

A returned {@link Flux} consumes one supplied physical response and supports exactly one subscription. Before + * that subscription claims the response, ownership remains with the caller; if it is never subscribed, the caller + * must consume or cancel the response body.

+ * + *

Event streams are always decoded as UTF-8. A {@code charset} parameter in the response Content-Type doesn't + * select another encoding.

+ */ +public final class ServerSentEventStreams { + private ServerSentEventStreams() { + } + + /** + * Decodes a single server-sent event response as a {@link Flux}. + * + *

The response body is validated as {@code text/event-stream}, decoded incrementally, and closed on + * completion, failure, or cancellation. A 204 response produces an empty {@link Flux}. Only HTTP 200 and 204 + * responses are accepted.

+ * + * @param response The streaming response. + * @param converter Converts an event name and data payload into the generated event type. + * @param The type of the event data. + * @return A flux of decoded server-sent events. + */ + public static Flux> toFlux(Response response, + BiFunction converter) { + return ServerSentEventStream.toFlux(response, converter); + } + + /** + * Decodes a single server-sent event response as a {@link Flux} until an inclusive terminal event is emitted. + * + *

The response body is validated as {@code text/event-stream}, decoded incrementally, and closed after a + * terminal event, on response-body EOF, on failure, or on cancellation. A 204 response produces an empty + * {@link Flux} without evaluating the predicate. If the response body ends before a terminal event is emitted, the + * flux completes normally. This method does not reconnect or replay a request.

+ * + * @param response The streaming response. + * @param converter Converts an event name and data payload into the generated event type. + * @param terminalEvent Identifies an inclusive terminal event that ends processing early. + * @param The type of the event data. + * @return A flux of decoded server-sent events. + */ + public static Flux> toFlux(Response response, + BiFunction converter, Predicate> terminalEvent) { + return ServerSentEventStream.toFlux(response, converter, terminalEvent); + } + + /** + * Decodes a single server-sent event response as a closeable iterable stream. + * + *

The response body is validated as {@code text/event-stream} and decoded incrementally. Close the returned + * stream if iteration ends early to release the response body. A 204 response produces an empty stream. Only HTTP + * 200 and 204 responses are accepted.

+ * + * @param response The streaming response. + * @param converter Converts an event name and data payload into the generated event type. + * @param The type of the event data. + * @return A closeable iterable stream of decoded server-sent events. + */ + public static CloseableIterableStream> toIterable(Response response, + BiFunction converter) { + return ServerSentEventStream.toIterable(response, converter); + } + + /** + * Decodes a single server-sent event response as a closeable iterable stream until an inclusive terminal event. + * + *

The response body is validated as {@code text/event-stream} and decoded incrementally. Close the returned + * stream if iteration ends early to release the response body. A 204 response produces an empty stream without + * evaluating the predicate. If the response body ends before a terminal event is emitted, the stream completes + * normally. This method does not reconnect or replay a request.

+ * + * @param response The streaming response. + * @param converter Converts an event name and data payload into the generated event type. + * @param terminalEvent Identifies an inclusive terminal event that ends processing early. + * @param The type of the event data. + * @return A closeable iterable stream of decoded server-sent events. + */ + public static CloseableIterableStream> toIterable(Response response, + BiFunction converter, Predicate> terminalEvent) { + return ServerSentEventStream.toIterable(response, converter, terminalEvent); + } +} diff --git a/sdk/core/azure-core/src/main/java/com/azure/core/implementation/util/ServerSentEventHelper.java b/sdk/core/azure-core/src/main/java/com/azure/core/implementation/util/ServerSentEventHelper.java new file mode 100644 index 000000000000..49fb9166dbf7 --- /dev/null +++ b/sdk/core/azure-core/src/main/java/com/azure/core/implementation/util/ServerSentEventHelper.java @@ -0,0 +1,75 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.core.implementation.util; + +import com.azure.core.http.ServerSentEvent; + +import java.time.Duration; +import java.util.Objects; +import java.util.concurrent.atomic.AtomicReference; + +/** + * Helper class that accesses non-public members of {@link ServerSentEvent}. + */ +public final class ServerSentEventHelper { + private static final AtomicReference ACCESSOR = new AtomicReference<>(); + + private ServerSentEventHelper() { + } + + /** + * Defines access to non-public members of {@link ServerSentEvent}. + */ + public interface ServerSentEventAccessor { + /** + * Creates a server-sent event. + * + * @param id The event identifier. + * @param event The event name. + * @param data The event data. + * @param comment The event comment. + * @param retryAfter The retry interval. + * @param The type of the event data. + * @return The server-sent event. + */ + ServerSentEvent create(String id, String event, T data, String comment, Duration retryAfter); + } + + /** + * Sets the accessor. + * + * @param serverSentEventAccessor The accessor. + */ + public static void setAccessor(final ServerSentEventAccessor serverSentEventAccessor) { + ACCESSOR.set(Objects.requireNonNull(serverSentEventAccessor, "'serverSentEventAccessor' cannot be null.")); + } + + /** + * Creates a server-sent event. + * + * @param id The event identifier. + * @param event The event name. + * @param data The event data. + * @param comment The event comment. + * @param retryAfter The retry interval. + * @param The type of the event data. + * @return The server-sent event. + */ + public static ServerSentEvent create(String id, String event, T data, String comment, Duration retryAfter) { + return getAccessor().create(id, event, data, comment, retryAfter); + } + + private static ServerSentEventAccessor getAccessor() { + ServerSentEventAccessor accessor = ACCESSOR.get(); + if (accessor == null) { + try { + Class.forName(ServerSentEvent.class.getName(), true, ServerSentEvent.class.getClassLoader()); + } catch (ClassNotFoundException exception) { + throw new IllegalStateException("Unable to initialize ServerSentEvent.", exception); + } + accessor = ACCESSOR.get(); + } + return accessor; + } +} diff --git a/sdk/core/azure-core/src/main/java/com/azure/core/implementation/util/ServerSentEventStream.java b/sdk/core/azure-core/src/main/java/com/azure/core/implementation/util/ServerSentEventStream.java new file mode 100644 index 000000000000..e55520c0af23 --- /dev/null +++ b/sdk/core/azure-core/src/main/java/com/azure/core/implementation/util/ServerSentEventStream.java @@ -0,0 +1,515 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.core.implementation.util; + +import com.azure.core.http.ServerSentEvent; +import com.azure.core.http.rest.Response; +import com.azure.core.util.BinaryData; +import com.azure.core.util.CloseableIterableStream; +import com.azure.core.util.logging.ClientLogger; +import reactor.core.publisher.Flux; + +import java.io.IOException; +import java.io.InputStream; +import java.io.UncheckedIOException; +import java.nio.ByteBuffer; +import java.nio.CharBuffer; +import java.nio.charset.CharacterCodingException; +import java.nio.charset.CharsetDecoder; +import java.nio.charset.CoderResult; +import java.nio.charset.CodingErrorAction; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.ArrayList; +import java.util.ArrayDeque; +import java.util.Collections; +import java.util.Deque; +import java.util.Iterator; +import java.util.List; +import java.util.NoSuchElementException; +import java.util.Objects; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.BiFunction; +import java.util.function.Predicate; + +/** + * Implementation support for parsing one server-sent event response. + * + *

Event streams are always decoded as UTF-8. A {@code charset} parameter in the response Content-Type doesn't + * select another encoding.

+ * + *

Response-based methods consume the response body exactly once. Streaming {@link BinaryData} returned by + * RestProxy owns the physical response and closes it when body consumption terminates.

+ */ +public final class ServerSentEventStream { + private static final String DEFAULT_EVENT = "message"; + private static final ClientLogger LOGGER = new ClientLogger(ServerSentEventStream.class); + + private ServerSentEventStream() { + } + + /** + * Decodes one server-sent event response body. + * + * @param body The response body. + * @param deserializer The event data deserializer. + * @param The event data type. + * @return The decoded events. + */ + public static Flux> decode(BinaryData body, BiFunction deserializer) { + Objects.requireNonNull(body, "'body' cannot be null."); + Objects.requireNonNull(deserializer, "'deserializer' cannot be null."); + return decodeBody(body, deserializer); + } + + /** + * Decodes an SSE response, closing its physical response on completion, failure, or cancellation. + */ + public static Flux> toFlux(Response response, + BiFunction deserializer) { + Objects.requireNonNull(response, "'response' cannot be null."); + Objects.requireNonNull(deserializer, "'deserializer' cannot be null."); + + return createFlux(response, deserializer, null); + } + + /** + * Decodes an SSE response until an inclusive terminal event is emitted, closing its physical response. + */ + public static Flux> toFlux(Response response, + BiFunction deserializer, Predicate> terminalEvent) { + Objects.requireNonNull(response, "'response' cannot be null."); + Objects.requireNonNull(deserializer, "'deserializer' cannot be null."); + Objects.requireNonNull(terminalEvent, "'terminalEvent' cannot be null."); + + return createFlux(response, deserializer, terminalEvent); + } + + private static Flux> createFlux(Response response, + BiFunction deserializer, Predicate> terminalEvent) { + AtomicBoolean subscribed = new AtomicBoolean(); + return Flux.defer(() -> { + if (!subscribed.compareAndSet(false, true)) { + return Flux + .error(new IllegalStateException("This server-sent event stream supports only one subscription.")); + } + + ServerSentEventStreamResponse streamResponse = ServerSentEventStreamResponse.fromResponse(response); + Flux> events = streamResponse.getStatusCode() == 204 + ? Flux.empty() + : decodeBody(streamResponse.getBody(), deserializer); + if (terminalEvent != null) { + events = events.takeUntil(terminalEvent); + } + + return events; + }); + } + + /** + * Decodes an SSE response as a blocking closeable iterable. + */ + public static CloseableIterableStream> toIterable(Response response, + BiFunction deserializer) { + Objects.requireNonNull(response, "'response' cannot be null."); + Objects.requireNonNull(deserializer, "'deserializer' cannot be null."); + + return createIterable(response, deserializer, null); + } + + /** + * Decodes an SSE response as a blocking closeable iterable until an inclusive terminal event is emitted. + */ + public static CloseableIterableStream> toIterable(Response response, + BiFunction deserializer, Predicate> terminalEvent) { + Objects.requireNonNull(response, "'response' cannot be null."); + Objects.requireNonNull(deserializer, "'deserializer' cannot be null."); + Objects.requireNonNull(terminalEvent, "'terminalEvent' cannot be null."); + + return createIterable(response, deserializer, terminalEvent); + } + + private static CloseableIterableStream> createIterable(Response response, + BiFunction deserializer, Predicate> terminalEvent) { + ServerSentEventStreamResponse streamResponse = ServerSentEventStreamResponse.fromResponse(response); + if (streamResponse.getStatusCode() == 204) { + return new CloseableIterableStream<>(Collections.emptyList(), () -> { + }); + } + + SyncSseIterable iterable = new SyncSseIterable<>(streamResponse.getBody(), deserializer, terminalEvent); + return new CloseableIterableStream<>(iterable, iterable); + } + + private static Flux> decodeBody(BinaryData body, + BiFunction deserializer) { + return Flux.defer(() -> { + ServerSentEventDecoder decoder = new ServerSentEventDecoder(new StreamState()); + Flux frames = body.toFluxByteBuffer() + // Keep body termination observable when demand exactly matches the final decoded events. + .hide() + .concatMapIterable(decoder::feed, 1) + .concatWith(Flux.defer(() -> Flux.fromIterable(decoder.finish()))); + + return frames.handle((frame, sink) -> { + T data = deserializer.apply(frame.event, frame.data); + if (data != null) { + sink.next(frame.toEvent(data)); + } + }); + }); + } + + private static void checkInterrupted() { + if (Thread.currentThread().isInterrupted()) { + throw new RuntimeException("Interrupted while processing the server-sent event stream.", + new InterruptedException()); + } + } + + private static String removeOptionalSpace(String value) { + return value.startsWith(" ") ? value.substring(1) : value; + } + + private static Duration parseRetryAfter(String value) { + if (value.isEmpty()) { + return null; + } + + for (int i = 0; i < value.length(); i++) { + char character = value.charAt(i); + if (character < '0' || character > '9') { + return null; + } + } + + try { + return Duration.ofMillis(Long.parseLong(value)); + } catch (NumberFormatException ignored) { + // Ignore retry values that don't fit in a long. + return null; + } + } + + private static final class StreamState { + private String lastEventId; + private Duration retryAfter; + + private void setLastEventId(String lastEventId) { + this.lastEventId = lastEventId; + } + + private void setRetryAfter(Duration retryAfter) { + this.retryAfter = retryAfter; + } + + } + + private static final class SyncSseIterable implements Iterable>, AutoCloseable { + private final BinaryData body; + private final BiFunction deserializer; + private final Predicate> terminalEvent; + private final AtomicBoolean iteratorClaimed = new AtomicBoolean(); + private final AtomicBoolean closed = new AtomicBoolean(); + private InputStream inputStream; + + private SyncSseIterable(BinaryData body, BiFunction deserializer, + Predicate> terminalEvent) { + this.body = body; + this.deserializer = deserializer; + this.terminalEvent = terminalEvent; + } + + @Override + public Iterator> iterator() { + if (!iteratorClaimed.compareAndSet(false, true)) { + throw new IllegalStateException("This server-sent event stream supports only one iteration."); + } + + if (closed.get()) { + return Collections.emptyIterator(); + } + + try { + inputStream = body.toStream(); + } catch (RuntimeException exception) { + close(); + throw exception; + } + return new SyncSseIterator(); + } + + @Override + public void close() { + if (!closed.compareAndSet(false, true)) { + return; + } + + if (inputStream != null) { + try { + inputStream.close(); + } catch (IOException exception) { + throw new UncheckedIOException(exception); + } + } else { + try { + body.toStream().close(); + } catch (IOException exception) { + throw new UncheckedIOException(exception); + } + } + } + + private final class SyncSseIterator implements Iterator> { + private final ServerSentEventDecoder decoder = new ServerSentEventDecoder(new StreamState()); + private final byte[] readBuffer = new byte[8192]; + private final Deque> pendingEvents = new ArrayDeque<>(); + private boolean complete; + + @Override + public boolean hasNext() { + while (pendingEvents.isEmpty() && !complete && !closed.get()) { + readNextChunk(); + } + + return !pendingEvents.isEmpty(); + } + + @Override + public ServerSentEvent next() { + if (!hasNext()) { + throw LOGGER.logExceptionAsError(new NoSuchElementException()); + } + + return pendingEvents.removeFirst(); + } + + private void readNextChunk() { + try { + checkInterrupted(); + int read = inputStream.read(readBuffer); + if (read == -1) { + complete = true; + addFrames(decoder.finish()); + close(); + } else if (read > 0) { + addFrames(decoder.feed(ByteBuffer.wrap(readBuffer, 0, read))); + } + } catch (IOException exception) { + complete = true; + close(); + throw LOGGER.logThrowableAsError(new UncheckedIOException(exception)); + } catch (RuntimeException exception) { + complete = true; + close(); + throw LOGGER.logThrowableAsError(exception); + } + } + + private void addFrames(List frames) { + for (ServerSentEventFrame frame : frames) { + checkInterrupted(); + T data = deserializer.apply(frame.event, frame.data); + if (data == null) { + continue; + } + + ServerSentEvent event = frame.toEvent(data); + pendingEvents.addLast(event); + if (terminalEvent != null && terminalEvent.test(event)) { + complete = true; + close(); + return; + } + } + } + } + } + + private static final class ServerSentEventDecoder { + private final StreamState state; + private final CharsetDecoder charsetDecoder = StandardCharsets.UTF_8.newDecoder() + .onMalformedInput(CodingErrorAction.REPORT) + .onUnmappableCharacter(CodingErrorAction.REPORT); + private ByteBuffer remainingBytes = ByteBuffer.allocate(0); + private final StringBuilder line = new StringBuilder(); + private boolean pendingCarriageReturn; + private boolean firstLine = true; + private String event; + private List data; + private String comment; + + private ServerSentEventDecoder(StreamState state) { + this.state = state; + } + + private List feed(ByteBuffer source) { + return feedCharacters(decode(source.duplicate(), false)); + } + + private List feedCharacters(CharBuffer buffer) { + List events = new ArrayList<>(); + + while (buffer.hasRemaining()) { + char value = buffer.get(); + + if (pendingCarriageReturn) { + pendingCarriageReturn = false; + if (value == '\n') { + continue; + } + } + + if (value == '\n') { + processLine(consumeLine(), events); + } else if (value == '\r') { + processLine(consumeLine(), events); + pendingCarriageReturn = true; + } else { + line.append(value); + } + } + + return events; + } + + private List finish() { + // The SSE parsing algorithm discards an event that wasn't terminated by a blank line. + decode(ByteBuffer.allocate(0), true); + return Collections.emptyList(); + } + + private CharBuffer decode(ByteBuffer source, boolean endOfInput) { + ByteBuffer input = ByteBuffer.allocate(remainingBytes.remaining() + source.remaining()); + input.put(remainingBytes.duplicate()); + input.put(source); + input.flip(); + CharBuffer output = CharBuffer.allocate((int) (input.remaining() * charsetDecoder.maxCharsPerByte()) + 1); + try { + CoderResult result = charsetDecoder.decode(input, output, endOfInput); + if (result.isError()) { + result.throwException(); + } + if (endOfInput) { + result = charsetDecoder.flush(output); + if (result.isError()) { + result.throwException(); + } + } + } catch (CharacterCodingException exception) { + throw new IllegalStateException("Failed to decode the server-sent event stream.", exception); + } + remainingBytes = ByteBuffer.allocate(input.remaining()); + remainingBytes.put(input).flip(); + output.flip(); + return output; + } + + private String consumeLine() { + String decodedLine = line.toString(); + line.setLength(0); + + if (firstLine) { + firstLine = false; + if (!decodedLine.isEmpty() && decodedLine.charAt(0) == '\uFEFF') { + return decodedLine.substring(1); + } + } + + return decodedLine; + } + + private void processLine(String line, List events) { + if (line.isEmpty()) { + ServerSentEventFrame parsedEvent = buildEvent(); + if (parsedEvent != null) { + events.add(parsedEvent); + } + return; + } + + if (line.charAt(0) == ':') { + comment = removeOptionalSpace(line.substring(1)); + return; + } + + int colonIndex = line.indexOf(':'); + String field = colonIndex < 0 ? line : line.substring(0, colonIndex); + String value = colonIndex < 0 ? "" : removeOptionalSpace(line.substring(colonIndex + 1)); + + switch (field) { + case "event": + event = value; + break; + + case "data": + if (data == null) { + data = new ArrayList<>(); + } + data.add(value); + break; + + case "id": + if (value.indexOf('\0') < 0) { + state.setLastEventId(value); + } + break; + + case "retry": + Duration parsedRetryAfter = parseRetryAfter(value); + if (parsedRetryAfter != null) { + state.setRetryAfter(parsedRetryAfter); + } + break; + + default: + break; + } + } + + private ServerSentEventFrame buildEvent() { + String currentEvent = event; + List currentData = data; + String currentComment = comment; + resetEvent(); + + if (currentData == null) { + return null; + } + + if (currentEvent == null || currentEvent.isEmpty()) { + currentEvent = DEFAULT_EVENT; + } + + return new ServerSentEventFrame(state.lastEventId, currentEvent, String.join("\n", currentData), + currentComment, state.retryAfter); + } + + private void resetEvent() { + event = null; + data = null; + comment = null; + } + } + + private static final class ServerSentEventFrame { + private final String id; + private final String event; + private final String data; + private final String comment; + private final Duration retryAfter; + + private ServerSentEventFrame(String id, String event, String data, String comment, Duration retryAfter) { + this.id = id; + this.event = event; + this.data = data; + this.comment = comment; + this.retryAfter = retryAfter; + } + + private ServerSentEvent toEvent(T data) { + return ServerSentEventHelper.create(id, event, data, comment, retryAfter); + } + } +} diff --git a/sdk/core/azure-core/src/main/java/com/azure/core/implementation/util/ServerSentEventStreamResponse.java b/sdk/core/azure-core/src/main/java/com/azure/core/implementation/util/ServerSentEventStreamResponse.java new file mode 100644 index 000000000000..681acf5f616d --- /dev/null +++ b/sdk/core/azure-core/src/main/java/com/azure/core/implementation/util/ServerSentEventStreamResponse.java @@ -0,0 +1,85 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.core.implementation.util; + +import com.azure.core.http.HttpHeaderName; +import com.azure.core.http.rest.Response; +import com.azure.core.util.BinaryData; +import com.azure.core.util.logging.ClientLogger; +import org.reactivestreams.Subscription; +import reactor.core.publisher.BaseSubscriber; + +import java.nio.ByteBuffer; +import java.util.Objects; + +/** + * Validates the response used by a logical server-sent event stream. + */ +final class ServerSentEventStreamResponse { + private static final ClientLogger LOGGER = new ClientLogger(ServerSentEventStreamResponse.class); + + private final int statusCode; + private final BinaryData body; + + ServerSentEventStreamResponse(int statusCode, BinaryData body) { + this.statusCode = statusCode; + this.body = body; + } + + /** + * Creates a stream response from a REST response. + * + * @param response The REST response. + * @return The stream response. + */ + static ServerSentEventStreamResponse fromResponse(Response response) { + Objects.requireNonNull(response, "'response' cannot be null."); + if (response.getStatusCode() != 200 && response.getStatusCode() != 204) { + closeResponse(response); + throw LOGGER.logExceptionAsError( + new IllegalStateException("Expected a server-sent event response to have status code 200 or 204.")); + } + String contentType = response.getHeaders().getValue(HttpHeaderName.CONTENT_TYPE); + if (response.getStatusCode() == 200 && !HttpUtils.isTextEventStreamContentType(contentType)) { + closeResponse(response); + throw LOGGER.logExceptionAsError(new IllegalStateException( + "Expected a successful server-sent event response to have Content-Type 'text/event-stream'.")); + } + + BinaryData body = response.getValue(); + if (response.getStatusCode() == 200) { + if (body == null) { + closeResponse(response); + throw new NullPointerException("'response.getValue()' cannot be null unless the status code is 204."); + } + } + if (response.getStatusCode() == 204) { + closeBody(body); + } + return new ServerSentEventStreamResponse(response.getStatusCode(), body); + } + + private static void closeResponse(Response response) { + closeBody(response.getValue()); + } + + private static void closeBody(BinaryData body) { + if (body != null) { + body.toFluxByteBuffer().subscribe(new BaseSubscriber() { + @Override + protected void hookOnSubscribe(Subscription subscription) { + cancel(); + } + }); + } + } + + int getStatusCode() { + return statusCode; + } + + BinaryData getBody() { + return body; + } +} diff --git a/sdk/core/azure-core/src/main/java/com/azure/core/util/CloseableIterableStream.java b/sdk/core/azure-core/src/main/java/com/azure/core/util/CloseableIterableStream.java new file mode 100644 index 000000000000..e692ffab16ee --- /dev/null +++ b/sdk/core/azure-core/src/main/java/com/azure/core/util/CloseableIterableStream.java @@ -0,0 +1,66 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.core.util; + +import com.azure.core.util.logging.ClientLogger; + +import java.util.Objects; +import java.util.stream.Stream; + +/** + * An {@link IterableStream} that owns a closeable resource. + * + *

Close this stream when iteration does not reach its natural end, for example by using it in a + * try-with-resources statement. Closing the stream releases its owned resource and is safe to call more than once.

+ * + * @param The type of values in this stream. + */ +public final class CloseableIterableStream extends IterableStream implements AutoCloseable { + private static final ClientLogger LOGGER = new ClientLogger(CloseableIterableStream.class); + + private final AutoCloseable closeable; + private boolean closed; + + /** + * Creates an instance with the given iterable and closeable resource. + * + * @param iterable The values to iterate over. + * @param closeable The resource to release when this stream is closed. + * @throws NullPointerException If {@code iterable} or {@code closeable} is {@code null}. + */ + public CloseableIterableStream(Iterable iterable, AutoCloseable closeable) { + super(iterable); + this.closeable = Objects.requireNonNull(closeable, "'closeable' cannot be null."); + } + + /** + * Gets a Java stream of values that closes this iterable stream when the returned stream is closed. + * + * @return A Java stream of values. + */ + @Override + public Stream stream() { + return super.stream().onClose(this::close); + } + + /** + * Releases the resource owned by this stream. + * + *

This method is idempotent.

+ */ + @Override + public synchronized void close() { + if (closed) { + return; + } + + closed = true; + try { + closeable.close(); + } catch (Exception exception) { + throw LOGGER + .logExceptionAsError(new IllegalStateException("Failed to close the iterable stream.", exception)); + } + } +} diff --git a/sdk/core/azure-core/src/test/java/com/azure/core/http/ServerSentEventTests.java b/sdk/core/azure-core/src/test/java/com/azure/core/http/ServerSentEventTests.java new file mode 100644 index 000000000000..c40b716c1cde --- /dev/null +++ b/sdk/core/azure-core/src/test/java/com/azure/core/http/ServerSentEventTests.java @@ -0,0 +1,25 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.core.http; + +import com.azure.core.implementation.util.ServerSentEventHelper; +import org.junit.jupiter.api.Test; + +import java.time.Duration; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +public class ServerSentEventTests { + @Test + public void helperCreatesEvent() { + ServerSentEvent event + = ServerSentEventHelper.create("42", "stockUpdate", "payload", "comment", Duration.ofSeconds(2)); + + assertEquals("42", event.getId()); + assertEquals("stockUpdate", event.getEvent()); + assertEquals("payload", event.getData()); + assertEquals("comment", event.getComment()); + assertEquals(Duration.ofSeconds(2), event.getRetryAfter()); + } +} diff --git a/sdk/core/azure-core/src/test/java/com/azure/core/implementation/http/rest/SyncRestProxyTests.java b/sdk/core/azure-core/src/test/java/com/azure/core/implementation/http/rest/SyncRestProxyTests.java index 98099311266d..d6c5a8f5ebe9 100644 --- a/sdk/core/azure-core/src/test/java/com/azure/core/implementation/http/rest/SyncRestProxyTests.java +++ b/sdk/core/azure-core/src/test/java/com/azure/core/implementation/http/rest/SyncRestProxyTests.java @@ -221,8 +221,6 @@ public void binaryDataResponseClosesOnCompletion() { assertFalse(responseBody.isReplayable()); assertEquals("hello", responseBody.toString()); assertEquals(1, responseCloseCount.get()); - assertEquals("hello", responseBody.toString()); - assertEquals(1, responseCloseCount.get()); } @Test diff --git a/sdk/core/azure-core/src/test/java/com/azure/core/implementation/util/ServerSentEventStreamTests.java b/sdk/core/azure-core/src/test/java/com/azure/core/implementation/util/ServerSentEventStreamTests.java new file mode 100644 index 000000000000..dccba0d56daa --- /dev/null +++ b/sdk/core/azure-core/src/test/java/com/azure/core/implementation/util/ServerSentEventStreamTests.java @@ -0,0 +1,833 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.core.implementation.util; + +import com.azure.core.http.HttpHeaderName; +import com.azure.core.http.HttpHeaders; +import com.azure.core.http.ServerSentEvent; +import com.azure.core.http.ServerSentEventStreams; +import com.azure.core.http.rest.ResponseBase; +import com.azure.core.util.BinaryData; +import com.azure.core.util.CloseableIterableStream; +import org.reactivestreams.Subscription; +import org.junit.jupiter.api.Test; +import reactor.core.CoreSubscriber; +import reactor.core.publisher.Flux; +import reactor.test.StepVerifier; +import reactor.util.context.Context; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.stream.Stream; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class ServerSentEventStreamTests { + @Test + public void toFluxParsesFragmentedEventMetadata() { + byte[] bytes = ("\uFEFF: comment\rid: 42\r\nevent: greeting\nretry: 2000\ndata: caf\u00e9\r\ndata: second\n\n") + .getBytes(StandardCharsets.UTF_8); + List buffers = new ArrayList<>(); + for (byte value : bytes) { + buffers.add(ByteBuffer.wrap(new byte[] { value })); + } + + TestResponse response = response(200, BinaryData.fromFlux(Flux.fromIterable(buffers), null, false).block()); + + StepVerifier.create(ServerSentEventStreams.toFlux(response, (event, data) -> data)).assertNext(event -> { + assertEquals("42", event.getId()); + assertEquals("greeting", event.getEvent()); + assertEquals("caf\u00e9\nsecond", event.getData()); + assertEquals("comment", event.getComment()); + assertEquals(Duration.ofSeconds(2), event.getRetryAfter()); + }).verifyComplete(); + + assertTrue(response.closed.get()); + } + + @Test + public void toFluxDeserializesMultipleBufferedEventsOnDemand() { + TestResponse response = response(200, BinaryData.fromString("data: one\n\ndata: two\n\ndata: three\n\n")); + AtomicInteger conversionCount = new AtomicInteger(); + + StepVerifier.create(ServerSentEventStreams.toFlux(response, (event, data) -> { + conversionCount.incrementAndGet(); + return data; + }), 0) + .thenRequest(1) + .assertNext(event -> assertEquals("one", event.getData())) + .then(() -> assertEquals(1, conversionCount.get())) + .thenRequest(1) + .assertNext(event -> assertEquals("two", event.getData())) + .then(() -> assertEquals(2, conversionCount.get())) + .thenRequest(1) + .assertNext(event -> assertEquals("three", event.getData())) + .then(() -> assertEquals(3, conversionCount.get())) + .verifyComplete(); + + assertTrue(response.closed.get()); + } + + @Test + public void toFluxNullConversionDoesNotConsumeDemand() { + TestResponse response = response(200, BinaryData.fromString("data: skip\n\ndata: deliver\n\n")); + AtomicInteger conversionCount = new AtomicInteger(); + + StepVerifier.create(ServerSentEventStreams.toFlux(response, (event, data) -> { + conversionCount.incrementAndGet(); + return "skip".equals(data) ? null : data; + }), 0) + .thenRequest(1) + .assertNext(event -> assertEquals("deliver", event.getData())) + .then(() -> assertEquals(2, conversionCount.get())) + .verifyComplete(); + + assertTrue(response.closed.get()); + } + + @Test + public void toFluxCompletesOnEofWithoutReconnecting() { + TestResponse response = response(200, BinaryData.fromString("id: 1\nretry: 0\ndata: one\n\n")); + + StepVerifier.create(ServerSentEventStreams.toFlux(response, (event, data) -> data)) + .assertNext(event -> assertEquals("one", event.getData())) + .verifyComplete(); + + assertTrue(response.closed.get()); + } + + @Test + public void toFluxReturnsEmptyForNoContent() { + TestResponse response = response(204, null); + + StepVerifier.create(ServerSentEventStreams.toFlux(response, (event, data) -> data)).verifyComplete(); + } + + @Test + public void toFluxClosesBodyOwnedNoContentResponse() { + AtomicBoolean bodyClosed = new AtomicBoolean(); + BinaryData body + = BinaryData + .fromFlux(Flux.using(() -> bodyClosed, ignored -> Flux.never(), ignored -> bodyClosed.set(true)), null, + false) + .block(); + ResponseBase response = new ResponseBase<>(null, 204, new HttpHeaders(), body, null); + + StepVerifier.create(ServerSentEventStreams.toFlux(response, (event, data) -> data)).verifyComplete(); + + assertTrue(bodyClosed.get()); + } + + @Test + public void toFluxRejectsInvalidContentTypeAndClosesResponse() { + TestResponse response = response(200, BinaryData.fromString("data: one\n\n"), "application/json"); + + assertThrows(IllegalStateException.class, + () -> ServerSentEventStreams.toFlux(response, (event, data) -> data).blockLast()); + + assertTrue(response.closed.get()); + } + + @Test + public void responseDecodersIgnoreDeclaredCharset() { + byte[] bytes = "data: caf\u00e9\n\n".getBytes(StandardCharsets.UTF_8); + List buffers = new ArrayList<>(); + for (byte value : bytes) { + buffers.add(ByteBuffer.wrap(new byte[] { value })); + } + TestResponse response = response(200, BinaryData.fromFlux(Flux.fromIterable(buffers), null, false).block(), + "text/event-stream; charset=UTF-16BE"); + + StepVerifier.create(ServerSentEventStreams.toFlux(response, (event, data) -> data)) + .assertNext(event -> assertEquals("caf\u00e9", event.getData())) + .verifyComplete(); + + assertTrue(response.closed.get()); + + TestResponse syncResponse = response(200, BinaryData.fromBytes(bytes), "text/event-stream; charset=UTF-16BE"); + List events = new ArrayList<>(); + try (CloseableIterableStream> stream + = ServerSentEventStreams.toIterable(syncResponse, (event, data) -> data)) { + stream.forEach(event -> events.add(event.getData())); + } + + assertEquals(1, events.size()); + assertEquals("caf\u00e9", events.get(0)); + assertTrue(syncResponse.closed.get()); + } + + @Test + public void toFluxCompletesForEmptyBody() { + TestResponse response = response(200, BinaryData.fromBytes(new byte[0])); + + StepVerifier.create(ServerSentEventStreams.toFlux(response, (eventName, data) -> data)).verifyComplete(); + + assertTrue(response.closed.get()); + } + + @Test + public void toFluxCompletesForBomOnlyBody() { + TestResponse response + = response(200, BinaryData.fromBytes(new byte[] { (byte) 0xEF, (byte) 0xBB, (byte) 0xBF })); + + StepVerifier.create(ServerSentEventStreams.toFlux(response, (eventName, data) -> data)).verifyComplete(); + + assertTrue(response.closed.get()); + } + + @Test + public void toFluxRejectsNonUtf8Bom() { + TestResponse response = response(200, BinaryData.fromBytes(new byte[] { (byte) 0xFF, (byte) 0xFE })); + + StepVerifier.create(ServerSentEventStreams.toFlux(response, (eventName, data) -> data)) + .expectError(IllegalStateException.class) + .verify(); + + assertTrue(response.closed.get()); + } + + @Test + public void toFluxFailsForTruncatedBomPrefixWithoutNullPointerException() { + TestResponse response = response(200, BinaryData.fromBytes(new byte[] { (byte) 0xEF })); + + StepVerifier.create(ServerSentEventStreams.toFlux(response, (eventName, data) -> data)) + .expectErrorMatches( + error -> error instanceof IllegalStateException && !(error instanceof NullPointerException)) + .verify(); + + assertTrue(response.closed.get()); + } + + @Test + public void toFluxIgnoresUnrecognizedCharset() { + TestResponse response + = response(200, BinaryData.fromString("data: caf\u00e9\n\n"), "text/event-stream; charset=not-a-charset"); + + ServerSentEvent event = ServerSentEventStreams.toFlux(response, (eventName, data) -> data).blockLast(); + + assertEquals("caf\u00e9", event.getData()); + assertTrue(response.closed.get()); + } + + @Test + public void toFluxConsumesBodyOwnedNonCloseableResponse() { + AtomicBoolean bodyClosed = new AtomicBoolean(); + BinaryData body = BinaryData.fromFlux(Flux.using(() -> bodyClosed, + ignored -> Flux.just(ByteBuffer.wrap("data: one\n\n".getBytes(StandardCharsets.UTF_8))), + ignored -> bodyClosed.set(true)), null, false).block(); + ResponseBase response = new ResponseBase<>(null, 200, + new HttpHeaders().set(HttpHeaderName.CONTENT_TYPE, "text/event-stream"), body, null); + + StepVerifier.create(ServerSentEventStreams.toFlux(response, (event, data) -> data)) + .assertNext(event -> assertEquals("one", event.getData())) + .verifyComplete(); + + assertTrue(bodyClosed.get()); + } + + @Test + public void toFluxClosesBodyOwnedNonCloseableResponseOnValidationFailure() { + AtomicBoolean bodyClosed = new AtomicBoolean(); + BinaryData body + = BinaryData + .fromFlux(Flux.using(() -> bodyClosed, ignored -> Flux.never(), ignored -> bodyClosed.set(true)), null, + false) + .block(); + ResponseBase response = new ResponseBase<>(null, 200, new HttpHeaders(), body, null); + + StepVerifier.create(ServerSentEventStreams.toFlux(response, (event, data) -> data)) + .expectErrorMessage( + "Expected a successful server-sent event response to have Content-Type " + "'text/event-stream'.") + .verify(); + + assertTrue(bodyClosed.get()); + } + + @Test + public void toFluxRejectsMissingContentTypeAndClosesResponse() { + TestResponse response = new TestResponse(200, new HttpHeaders(), BinaryData.fromString("data: one\n\n")); + + assertThrows(IllegalStateException.class, + () -> ServerSentEventStreams.toFlux(response, (event, data) -> data).blockLast()); + + assertTrue(response.closed.get()); + } + + @Test + public void toFluxRejectsNullBody() { + TestResponse response = response(200, null); + + assertThrows(NullPointerException.class, + () -> ServerSentEventStreams.toFlux(response, (event, data) -> data).blockLast()); + + } + + @Test + public void toFluxRejectsUnsupportedStatusAndClosesResponse() { + TestResponse response = response(201, BinaryData.fromString("data: one\n\n")); + + StepVerifier.create(ServerSentEventStreams.toFlux(response, (event, data) -> data)) + .expectErrorMessage("Expected a server-sent event response to have status code 200 or 204.") + .verify(); + + assertTrue(response.closed.get()); + } + + @Test + public void toFluxDoesNotClaimOrCloseResponseBeforeSubscription() { + TestResponse response = response(200, BinaryData.fromString("data: one\n\n")); + + ServerSentEventStreams.toFlux(response, (event, data) -> data); + + assertFalse(response.closed.get()); + } + + @Test + public void toFluxAllowsOnlyOneSubscription() { + TestResponse response = response(200, BinaryData.fromString("data: one\n\n")); + Flux> events = ServerSentEventStreams.toFlux(response, (event, data) -> data); + + StepVerifier.create(events).expectNextCount(1).verifyComplete(); + StepVerifier.create(events) + .expectErrorMessage("This server-sent event stream supports only one subscription.") + .verify(); + + assertTrue(response.closed.get()); + } + + @Test + public void toFluxCancellationClosesResponse() { + AtomicBoolean cancelled = new AtomicBoolean(); + BinaryData body = BinaryData.fromFlux( + Flux.concat(Flux.just(ByteBuffer.wrap("data: one\n\n".getBytes(StandardCharsets.UTF_8))), Flux.never()) + .doOnCancel(() -> cancelled.set(true)), + null, false).block(); + TestResponse response = response(200, body); + + StepVerifier.create(ServerSentEventStreams.toFlux(response, (event, data) -> data)) + .assertNext(event -> assertEquals("one", event.getData())) + .thenCancel() + .verify(); + + assertTrue(cancelled.get()); + assertTrue(response.closed.get()); + } + + @Test + public void decodeDoesNotCompleteAfterCancellationFromOnNext() { + AtomicBoolean completed = new AtomicBoolean(); + AtomicReference error = new AtomicReference<>(); + List events = new ArrayList<>(); + + ServerSentEventStream.decode(BinaryData.fromString("data: one\n\n"), (event, data) -> data) + .subscribe(new CoreSubscriber>() { + private Subscription subscription; + + @Override + public void onSubscribe(Subscription subscription) { + this.subscription = subscription; + subscription.request(1); + } + + @Override + public void onNext(ServerSentEvent event) { + events.add(event.getData()); + subscription.cancel(); + } + + @Override + public void onError(Throwable throwable) { + error.set(throwable); + } + + @Override + public void onComplete() { + completed.set(true); + } + + @Override + public Context currentContext() { + return Context.empty(); + } + }); + + assertEquals(1, events.size()); + assertFalse(completed.get()); + assertNull(error.get()); + } + + @Test + public void toIterableCompletesOnEofAndClosesResponse() { + TestResponse response = response(200, BinaryData.fromString("data: one\n\ndata: two\n\n")); + List events = new ArrayList<>(); + + try (CloseableIterableStream> stream + = ServerSentEventStreams.toIterable(response, (event, data) -> data)) { + stream.forEach(event -> events.add(event.getData())); + } + + assertEquals(2, events.size()); + assertTrue(response.closed.get()); + } + + @Test + public void toIterableEarlyCloseCancelsResponse() { + AtomicBoolean cancelled = new AtomicBoolean(); + BinaryData body = BinaryData.fromFlux( + Flux.concat(Flux.just(ByteBuffer.wrap("data: one\n\n".getBytes(StandardCharsets.UTF_8))), Flux.never()) + .doOnCancel(() -> cancelled.set(true)), + null, false).block(); + TestResponse response = response(200, body); + + try (CloseableIterableStream> stream + = ServerSentEventStreams.toIterable(response, (event, data) -> data)) { + assertEquals("one", stream.iterator().next().getData()); + } + + assertTrue(cancelled.get()); + assertTrue(response.closed.get()); + } + + @Test + public void toIterableStreamCloseAfterEarlyTerminationCancelsResponse() { + AtomicBoolean cancelled = new AtomicBoolean(); + BinaryData body = BinaryData.fromFlux( + Flux.concat(Flux.just(ByteBuffer.wrap("data: one\n\n".getBytes(StandardCharsets.UTF_8))), Flux.never()) + .doOnCancel(() -> cancelled.set(true)), + null, false).block(); + TestResponse response = response(200, body); + CloseableIterableStream> iterable + = ServerSentEventStreams.toIterable(response, (event, data) -> data); + + try (Stream> stream = iterable.stream()) { + assertEquals("one", stream.findFirst().get().getData()); + } + + assertTrue(cancelled.get()); + assertTrue(response.closed.get()); + } + + @Test + public void toIterableCloseBeforeIterationCancelsResponse() { + AtomicBoolean cancelled = new AtomicBoolean(); + BinaryData body + = BinaryData.fromFlux(Flux.never().doOnCancel(() -> cancelled.set(true)), null, false).block(); + TestResponse response = response(200, body); + + CloseableIterableStream> stream + = ServerSentEventStreams.toIterable(response, (event, data) -> data); + stream.close(); + stream.close(); + + assertTrue(cancelled.get()); + assertTrue(response.closed.get()); + assertFalse(stream.iterator().hasNext()); + } + + @Test + public void idAndRetryAreMetadataOnly() { + TestResponse response = response(200, BinaryData.fromString("id: 42\nretry: 1000\ndata: one\n\n")); + + StepVerifier.create(ServerSentEventStreams.toFlux(response, (event, data) -> data)).assertNext(event -> { + assertEquals("42", event.getId()); + assertEquals(Duration.ofSeconds(1), event.getRetryAfter()); + }).verifyComplete(); + } + + @Test + public void parserSkipsMetadataOnlyBlocksAndPersistsMetadata() { + TestResponse response + = response(200, BinaryData.fromString("id: 42\nretry: 1000\n\ndata: one\n\ndata: two\n\n")); + + StepVerifier.create(ServerSentEventStreams.toFlux(response, (event, data) -> data).collectList()) + .assertNext(events -> { + assertEquals(2, events.size()); + for (ServerSentEvent event : events) { + assertEquals("42", event.getId()); + assertEquals(Duration.ofSeconds(1), event.getRetryAfter()); + } + }) + .verifyComplete(); + } + + @Test + public void parserResetsIdAndUsesDefaultEvent() { + TestResponse response = response(200, BinaryData.fromString("id: 42\ndata: one\n\nid:\nevent:\ndata: two\n\n")); + + StepVerifier.create(ServerSentEventStreams.toFlux(response, (event, data) -> data)) + .assertNext(event -> assertEquals("42", event.getId())) + .assertNext(event -> { + assertEquals("", event.getId()); + assertEquals("message", event.getEvent()); + }) + .verifyComplete(); + } + + @Test + public void parserDiscardsUnterminatedEventAtEof() { + TestResponse response = response(200, BinaryData.fromString("event: partial\ndata: payload")); + + StepVerifier.create(ServerSentEventStreams.toFlux(response, (event, data) -> data)).verifyComplete(); + + assertTrue(response.closed.get()); + } + + @Test + public void toFluxPropagatesBodyFailureAndClosesResponse() { + IOException failure = new IOException("connection closed"); + BinaryData body = BinaryData + .fromFlux(Flux.concat(Flux.just(ByteBuffer.wrap("data: one\n\n".getBytes(StandardCharsets.UTF_8))), + Flux.error(failure)), null, false) + .block(); + TestResponse response = response(200, body); + + StepVerifier.create(ServerSentEventStreams.toFlux(response, (event, data) -> data)) + .assertNext(event -> assertEquals("one", event.getData())) + .expectErrorMatches(error -> error == failure) + .verify(); + + assertTrue(response.closed.get()); + } + + @Test + public void toFluxEmitsRequestedEventBeforeSynchronousBodyFailure() { + IOException failure = new IOException("connection closed"); + Flux source = Flux.from(subscriber -> subscriber.onSubscribe(new Subscription() { + private boolean signalled; + + @Override + public void request(long count) { + if (!signalled) { + signalled = true; + subscriber.onNext(ByteBuffer.wrap("data: one\n\n".getBytes(StandardCharsets.UTF_8))); + subscriber.onError(failure); + } + } + + @Override + public void cancel() { + } + })); + TestResponse response = response(200, BinaryData.fromFlux(source, null, false).block()); + + StepVerifier.create(ServerSentEventStreams.toFlux(response, (event, data) -> data)) + .assertNext(event -> assertEquals("one", event.getData())) + .expectErrorMatches(error -> error == failure) + .verify(); + + assertTrue(response.closed.get()); + } + + @Test + public void decodeHonorsReentrantDemandBeforeSynchronousBodyFailure() { + IOException failure = new IOException("connection closed"); + Flux source = Flux.from(subscriber -> subscriber.onSubscribe(new Subscription() { + private boolean signalled; + + @Override + public void request(long count) { + if (!signalled) { + signalled = true; + subscriber.onNext(ByteBuffer.wrap("data: one\n\ndata: two\n\n".getBytes(StandardCharsets.UTF_8))); + subscriber.onError(failure); + } + } + + @Override + public void cancel() { + } + })); + List events = new ArrayList<>(); + AtomicReference error = new AtomicReference<>(); + + ServerSentEventStream.decode(BinaryData.fromFlux(source, null, false).block(), (event, data) -> data) + .subscribe(new CoreSubscriber>() { + private Subscription subscription; + + @Override + public void onSubscribe(Subscription subscription) { + this.subscription = subscription; + subscription.request(1); + } + + @Override + public void onNext(ServerSentEvent event) { + events.add(event.getData()); + if (events.size() == 1) { + subscription.request(1); + } + } + + @Override + public void onError(Throwable throwable) { + error.set(throwable); + } + + @Override + public void onComplete() { + } + + @Override + public Context currentContext() { + return Context.empty(); + } + }); + + assertEquals(2, events.size()); + assertEquals("one", events.get(0)); + assertEquals("two", events.get(1)); + assertSame(failure, error.get()); + } + + @Test + public void toFluxPropagatesBodyFailurePublishedFromAnotherThread() { + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + for (int i = 0; i < 100; i++) { + IOException failure = new IOException("connection closed " + i); + Flux source = Flux.from(subscriber -> subscriber.onSubscribe(new Subscription() { + private final AtomicBoolean signalled = new AtomicBoolean(); + + @Override + public void request(long count) { + if (signalled.compareAndSet(false, true)) { + executor.execute(() -> subscriber.onError(failure)); + } + } + + @Override + public void cancel() { + } + })); + TestResponse response = response(200, BinaryData.fromFlux(source, null, false).block()); + + StepVerifier.create(ServerSentEventStreams.toFlux(response, (event, data) -> data)) + .expectErrorMatches(error -> error == failure) + .verify(); + + assertTrue(response.closed.get()); + } + } finally { + executor.shutdownNow(); + } + } + + @Test + public void toFluxPropagatesConverterFailureAndClosesResponse() { + RuntimeException failure = new IllegalStateException("invalid event"); + TestResponse response = response(200, BinaryData.fromString("data: invalid\n\n")); + + StepVerifier.create(ServerSentEventStreams.toFlux(response, (event, data) -> { + throw failure; + })).expectErrorMatches(error -> error == failure).verify(); + + assertTrue(response.closed.get()); + } + + @Test + public void toFluxEmitsTerminalEventAndCancelsRemainingBody() { + AtomicBoolean cancelled = new AtomicBoolean(); + AtomicReference conversionCount = new AtomicReference<>(0); + BinaryData body + = BinaryData + .fromFlux(Flux.concat( + Flux.just(ByteBuffer + .wrap("data: one\n\ndata: [DONE]\n\ndata: ignored\n\n".getBytes(StandardCharsets.UTF_8))), + Flux.never()).doOnCancel(() -> cancelled.set(true)), null, false) + .block(); + TestResponse response = response(200, body); + + StepVerifier.create(ServerSentEventStreams.toFlux(response, (event, data) -> { + conversionCount.set(conversionCount.get() + 1); + return data; + }, event -> "[DONE]".equals(event.getData()))) + .assertNext(event -> assertEquals("one", event.getData())) + .assertNext(event -> assertEquals("[DONE]", event.getData())) + .verifyComplete(); + + assertEquals(2, conversionCount.get()); + assertTrue(cancelled.get()); + assertTrue(response.closed.get()); + } + + @Test + public void toFluxCompletesOnEofBeforeTerminalEvent() { + TestResponse response = response(200, BinaryData.fromString("data: one\n\n")); + + StepVerifier.create(ServerSentEventStreams.toFlux(response, (event, data) -> data, event -> false)) + .assertNext(event -> assertEquals("one", event.getData())) + .verifyComplete(); + + assertTrue(response.closed.get()); + } + + @Test + public void toFluxCompletesOnMetadataOnlyEofBeforeTerminalEvent() { + TestResponse response = response(200, BinaryData.fromString("retry: 1000\n\n")); + + StepVerifier.create(ServerSentEventStreams.toFlux(response, (event, data) -> data, event -> false)) + .verifyComplete(); + + assertTrue(response.closed.get()); + } + + @Test + public void toFluxCancellationBeforeTerminalClosesResponse() { + AtomicBoolean cancelled = new AtomicBoolean(); + BinaryData body = BinaryData.fromFlux( + Flux.concat(Flux.just(ByteBuffer.wrap("data: one\n\n".getBytes(StandardCharsets.UTF_8))), Flux.never()) + .doOnCancel(() -> cancelled.set(true)), + null, false).block(); + TestResponse response = response(200, body); + + StepVerifier.create(ServerSentEventStreams.toFlux(response, (event, data) -> data, event -> false)) + .assertNext(event -> assertEquals("one", event.getData())) + .thenCancel() + .verify(); + + assertTrue(cancelled.get()); + assertTrue(response.closed.get()); + } + + @Test + public void toFluxNoContentDoesNotInvokeTerminalPredicate() { + AtomicBoolean predicateInvoked = new AtomicBoolean(); + TestResponse response = response(204, null); + + StepVerifier.create(ServerSentEventStreams.toFlux(response, (event, data) -> data, event -> { + predicateInvoked.set(true); + return false; + })).verifyComplete(); + + assertFalse(predicateInvoked.get()); + } + + @Test + public void toFluxPropagatesTerminalPredicateFailureAndClosesResponse() { + RuntimeException failure = new IllegalStateException("predicate failed"); + TestResponse response = response(200, BinaryData.fromString("data: one\n\n")); + + StepVerifier.create(ServerSentEventStreams.toFlux(response, (event, data) -> data, event -> { + throw failure; + })) + .assertNext(event -> assertEquals("one", event.getData())) + .expectErrorMatches(error -> error == failure) + .verify(); + + assertTrue(response.closed.get()); + } + + @Test + public void toIterableDeliversTerminalEventAndSkipsBufferedEventsAfterIt() { + TestResponse response = response(200, BinaryData.fromString("data: one\n\ndata: [DONE]\n\ndata: ignored\n\n")); + List events = new ArrayList<>(); + AtomicReference conversionCount = new AtomicReference<>(0); + + try (CloseableIterableStream> stream + = ServerSentEventStreams.toIterable(response, (event, data) -> { + conversionCount.set(conversionCount.get() + 1); + return data; + }, event -> "[DONE]".equals(event.getData()))) { + stream.forEach(event -> events.add(event.getData())); + } + + assertEquals(2, events.size()); + assertEquals("[DONE]", events.get(1)); + assertEquals(2, conversionCount.get()); + assertTrue(response.closed.get()); + } + + @Test + public void toIterableCompletesOnEofBeforeTerminalEvent() { + TestResponse response = response(200, BinaryData.fromString("data: one\n\n")); + List events = new ArrayList<>(); + + try (CloseableIterableStream> stream + = ServerSentEventStreams.toIterable(response, (event, data) -> data, event -> false)) { + stream.forEach(event -> events.add(event.getData())); + } + + assertEquals(1, events.size()); + assertEquals("one", events.get(0)); + assertTrue(response.closed.get()); + } + + @Test + public void toIterableNoContentDoesNotInvokeTerminalPredicate() { + AtomicBoolean predicateInvoked = new AtomicBoolean(); + TestResponse response = response(204, null); + + try (CloseableIterableStream> stream + = ServerSentEventStreams.toIterable(response, (event, data) -> data, event -> { + predicateInvoked.set(true); + return false; + })) { + assertFalse(stream.iterator().hasNext()); + } + + assertFalse(predicateInvoked.get()); + } + + @Test + public void toIterablePropagatesTerminalPredicateFailureAndClosesResponse() { + RuntimeException failure = new IllegalStateException("predicate failed"); + TestResponse response = response(200, BinaryData.fromString("data: one\n\n")); + + RuntimeException exception = assertThrows(RuntimeException.class, () -> { + try (CloseableIterableStream> stream + = ServerSentEventStreams.toIterable(response, (event, data) -> data, event -> { + throw failure; + })) { + stream.iterator().next(); + } + }); + + assertSame(failure, exception); + assertTrue(response.closed.get()); + } + + private static TestResponse response(int statusCode, BinaryData body) { + return response(statusCode, body, "text/event-stream"); + } + + private static TestResponse response(int statusCode, BinaryData body, String contentType) { + return new TestResponse(statusCode, new HttpHeaders().set(HttpHeaderName.CONTENT_TYPE, contentType), body); + } + + private static final class TestResponse extends ResponseBase { + private final AtomicBoolean closed; + + private TestResponse(int statusCode, HttpHeaders headers, BinaryData value) { + this(statusCode, headers, value, new AtomicBoolean()); + } + + private TestResponse(int statusCode, HttpHeaders headers, BinaryData value, AtomicBoolean closed) { + super(null, statusCode, headers, trackBody(value, closed), null); + this.closed = closed; + } + + private static BinaryData trackBody(BinaryData body, AtomicBoolean closed) { + return body == null + ? null + : BinaryData.fromFlux(body.toFluxByteBuffer() + .doOnComplete(() -> closed.set(true)) + .doOnError(ignored -> closed.set(true)) + .doOnCancel(() -> closed.set(true)), null, false).block(); + } + } +}