[DO NOT MERGE] WIP, add/backport generic server-sent events support to Azure Core

sdk/core/azure-core/CHANGELOG.md

@@ -4,6 +4,10 @@
 
 ### Features Added
 
+- Added generic `ServerSentEvent<T>`, `ServerSentEventListener<T>`, and `ServerSentEventStreams` APIs for typed,
+  incrementally decoded server-sent event streams. 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 +24,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))

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.
+ *
+ * <p>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.</p>
+ *
+ * <p>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.</p>
+ *
+ * @param <T> The type of the event data.
+ * @see <a href="https://html.spec.whatwg.org/multipage/server-sent-events.html#parsing-an-event-stream">
+ * Parsing an event stream</a>
+ */
+@Immutable
+public final class ServerSentEvent<T> {
+    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 <U> ServerSentEvent<U> 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;
+    }
+
+}

sdk/core/azure-core/src/main/java/com/azure/core/http/ServerSentEventListener.java

@@ -0,0 +1,42 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.core.http;
+
+/**
+ * A listener for receiving server-sent events.
+ *
+ * <p>Errors terminate processing, invoke {@link #onError(Throwable)} and {@link #onClose()}, and are rethrown to the
+ * synchronous service caller as unchecked exceptions.</p>
+ *
+ * <p>Generated Azure Core clients consume this listener above the HTTP transport after receiving a streaming response
+ * body. The listener isn't attached to the underlying {@link HttpRequest}.</p>
+ *
+ * @param <T> The type of the event data.
+ */
+@FunctionalInterface
+public interface ServerSentEventListener<T> {
+    /**
+     * Handles a server-sent event.
+     *
+     * @param event The server-sent event.
+     * @throws RuntimeException If an error occurs while handling the event.
+     */
+    void onEvent(ServerSentEvent<T> event);
+
+    /**
+     * Handles an error that terminates event processing.
+     *
+     * @param error The error that terminated event processing.
+     */
+    default void onError(Throwable error) {
+        // No-op by default.
+    }
+
+    /**
+     * Handles closure of the event stream.
+     */
+    default void onClose() {
+        // No-op by default.
+    }
+}

sdk/core/azure-core/src/main/java/com/azure/core/http/ServerSentEventStreams.java

@@ -0,0 +1,101 @@
+// 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 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.
+ *
+ * <p>The response body owns the physical response and closes it when consumption ends.</p>
+ *
+ * <p>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.</p>
+ *
+ * <p>Event streams are always decoded as UTF-8. A {@code charset} parameter in the response Content-Type doesn't
+ * select another encoding.</p>
+ */
+public final class ServerSentEventStreams {
+    private ServerSentEventStreams() {
+    }
+
+    /**
+     * Decodes a single server-sent event response as a {@link Flux}.
+     *
+     * <p>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.</p>
+     *
+     * @param response The streaming response.
+     * @param converter Converts an event name and data payload into the generated event type.
+     * @param <T> The type of the event data.
+     * @return A flux of decoded server-sent events.
+     */
+    public static <T> Flux<ServerSentEvent<T>> toFlux(Response<BinaryData> response,
+        BiFunction<String, String, T> converter) {
+        return ServerSentEventStream.toFlux(response, converter);
+    }
+
+    /**
+     * Decodes a single server-sent event response as a {@link Flux} until an inclusive terminal event is emitted.
+     *
+     * <p>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.</p>
+     *
+     * @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 <T> The type of the event data.
+     * @return A flux of decoded server-sent events.
+     */
+    public static <T> Flux<ServerSentEvent<T>> toFlux(Response<BinaryData> response,
+        BiFunction<String, String, T> converter, Predicate<ServerSentEvent<T>> terminalEvent) {
+        return ServerSentEventStream.toFlux(response, converter, terminalEvent);
+    }
+
+    /**
+     * Decodes a single server-sent event response and invokes a listener for each event.
+     *
+     * <p>The response body is validated as {@code text/event-stream}, decoded incrementally, and closed on EOF,
+     * failure, or interruption. A 204 response completes without events. Only HTTP 200 and 204 responses are
+     * accepted.</p>
+     *
+     * @param response The streaming response.
+     * @param converter Converts an event name and data payload into the generated event type.
+     * @param listener The listener that receives decoded events and lifecycle notifications.
+     * @param <T> The type of the event data.
+     */
+    public static <T> void listen(Response<BinaryData> response, BiFunction<String, String, T> converter,
+        ServerSentEventListener<T> listener) {
+        ServerSentEventStream.listen(response, converter, listener);
+    }
+
+    /**
+     * Decodes a single server-sent event response until an inclusive terminal event is delivered to a listener.
+     *
+     * <p>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 interruption. A 204 response completes without events
+     * and without evaluating the predicate. If the response body ends before a terminal event is delivered, this
+     * method completes normally. This method does not reconnect or replay a request.</p>
+     *
+     * @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 listener The listener that receives decoded events and lifecycle notifications.
+     * @param <T> The type of the event data.
+     */
+    public static <T> void listen(Response<BinaryData> response, BiFunction<String, String, T> converter,
+        Predicate<ServerSentEvent<T>> terminalEvent, ServerSentEventListener<T> listener) {
+        ServerSentEventStream.listen(response, converter, terminalEvent, listener);
+    }
+}

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<ServerSentEventAccessor> 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 <T> The type of the event data.
+         * @return The server-sent event.
+         */
+        <T> ServerSentEvent<T> 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 <T> The type of the event data.
+     * @return The server-sent event.
+     */
+    public static <T> ServerSentEvent<T> 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;
+    }
+}