Merge remote-tracking branch 'upstream/master' into format

Author: SentryMan

avaje-jex/src/main/java/io/avaje/jex/http/sse/SseClient.java

@@ -1,18 +1,20 @@
 package io.avaje.jex.http.sse;
 
+import java.io.Closeable;
+import java.util.function.Consumer;
+
 import io.avaje.jex.http.Context;
 import io.avaje.jex.http.ExchangeHandler;
 import io.avaje.jex.spi.JsonService;
-import java.io.Closeable;
-import java.util.function.Consumer;
 
 /**
- * A client for Server-Sent Events (SSE). This class handles the setup of the SSE connection, sending events and
- * comments to the client, and managing the lifecycle of the connection. It ensures proper headers are set and provides
- * methods for sending various types of data.
+ * A client for Server-Sent Events (SSE). This class handles the setup of the SSE connection,
+ * sending events and comments to the client, and managing the lifecycle of the connection. It
+ * ensures proper headers are set and provides methods for sending various types of data.
  *
- * <p>This class implements {@link Closeable} to allow for proper resource management. The connection is automatically
- * closed if the client disconnects or if an error occurs during event emission.
+ * <p>This class implements {@link Closeable} to allow for proper resource management. The
+ * connection is automatically closed if the client disconnects or if an error occurs during event
+ * emission.
  */
 public interface SseClient extends Closeable {
 
@@ -33,14 +35,23 @@ static ExchangeHandler handler(Consumer<SseClient> consumer) {
   Context ctx();
 
   /**
-   * By blocking the SSE connection, you can share this client outside the handler to notify it from other sources.
-   * Keep in mind that this function will block the handler until the SSE client is released by another thread.
+   * By blocking the SSE connection, you can share this client outside the handler to notify it from
+   * other sources. Keep in mind that this function will block the handler until the SSE client is
+   * released by another thread.
    */
   void keepAlive();
 
   /**
-   * Attempt to send a comment. If the {@link Emitter} fails to emit (remote client has disconnected), the
-   * {@link #close()} function will be called instead.
+   * Add a callback that will be called either when connection is closed through {@link #close()},
+   * or when the {@link Emitter} is detected as closed.
+   *
+   * @param task task to run
+   */
+  void onClose(Runnable task);
+
+  /**
+   * Attempt to send a comment. If the {@link Emitter} fails to emit (remote client has
+   * disconnected), the {@link #close()} function will be called instead.
    */
   void sendComment(String comment);
 
@@ -51,19 +62,19 @@ static ExchangeHandler handler(Consumer<SseClient> consumer) {
   void sendEvent(String event, Object data);
 
   /**
-   * Attempt to send an event. If the {@link Emitter} fails to emit (remote client has disconnected), the
-   * {@link #close()} function will be called instead.
+   * Attempt to send an event. If the {@link Emitter} fails to emit (remote client has
+   * disconnected), the {@link #close()} function will be called instead.
    *
    * @param event The name of the event.
-   * @param data The data to send in the event. This can be a String, an InputStream, or any object that can be
-   *     serialized to JSON using the configured {@link JsonService}.
+   * @param data The data to send in the event. This can be a String, an InputStream, or any object
+   *     that can be serialized to JSON using the configured {@link JsonService}.
    * @param id The ID of the event.
    */
   void sendEvent(String event, Object data, String id);
 
   /**
-   * Returns true if {@link #close()} has been called. This can either be by the user, or by Jex upon detecting that
-   * the {@link Emitter} is closed.
+   * Returns true if {@link #close()} has been called. This can either be by the user, or by Jex
+   * upon detecting that the {@link Emitter} is closed.
    */
   boolean terminated();
 }

avaje-jex/src/main/java/io/avaje/jex/http/sse/SseClientImpl.java

@@ -19,16 +19,24 @@ final class SseClientImpl implements SseClient {
   private final JsonService jsonService;
   private final Context ctx;
   private CompletableFuture<?> blockingFuture;
+  private Runnable closeCallback = () -> {};
 
   SseClientImpl(Context ctx) {
     this.emitter = new Emitter(ctx.exchange().getResponseBody());
     jsonService = ctx.jsonService();
     this.ctx = ctx;
   }
 
+  @Override
+  public void onClose(Runnable task) {
+    this.closeCallback = task;
+  }
+
   @Override
   public void close() {
-    if (terminated.getAndSet(true) && blockingFuture != null) {
+    if (terminated.getAndSet(true)) return;
+    closeCallback.run();
+    if (blockingFuture != null) {
       blockingFuture.complete(null);
     }
   }

avaje-jex/src/main/java/io/avaje/jex/spi/JsonService.java

@@ -15,7 +15,8 @@ public non-sealed interface JsonService extends JexExtension {
   /**
    * **Writes a Java Object as JSON to an OutputStream**
    *
-   * <p>Serializes a Java object into JSON format and writes the resulting JSON to the specified output stream.
+   * <p>Serializes a Java object into JSON format and writes the resulting JSON to the specified
+   * output stream.
    *
    * @param bean the Java object to be serialized
    * @param os the output stream to write the JSON data to
@@ -25,7 +26,8 @@ public non-sealed interface JsonService extends JexExtension {
   /**
    * **Writes a Java Object as a JSON string**
    *
-   * <p>Serializes a Java object into JSON string format and writes the resulting JSON to the specified output stream.
+   * <p>Serializes a Java object into JSON string format and writes the resulting JSON to the
+   * specified output stream.
    *
    * @param bean the Java object to be serialized
    * @return the serialized JSON string
@@ -35,7 +37,8 @@ public non-sealed interface JsonService extends JexExtension {
   /**
    * **Reads JSON from an InputStream**
    *
-   * <p>Reads a JSON-formatted input stream and deserializes it into a Java object of the specified type.
+   * <p>Reads a JSON-formatted input stream and deserializes it into a Java object of the specified
+   * type.
    *
    * @param type the Class object of the desired type
    * @param is the input stream containing the JSON data
@@ -46,7 +49,8 @@ public non-sealed interface JsonService extends JexExtension {
   /**
    * **Reads JSON from an InputStream**
    *
-   * <p>Reads a JSON-formatted input stream and deserializes it into a Java object of the specified type.
+   * <p>Reads a JSON-formatted input stream and deserializes it into a Java object of the specified
+   * type.
    *
    * @param type the Type object of the desired type
    * @param is the input stream containing the JSON data
@@ -55,11 +59,14 @@ public non-sealed interface JsonService extends JexExtension {
   <T> T fromJson(Type type, InputStream is);
 
   /**
-   * Serializes a stream of Java objects into a JSON-Stream format, using the {@code x-json-stream} media type. Each
-   * object in the stream is serialized as a separate JSON object, and the objects are separated by newlines.
+   * Serializes a stream of Java objects into a JSON-Stream format, using the {@code x-json-stream}
+   * media type. Each object in the stream is serialized as a separate JSON object, and the objects
+   * are separated by newlines.
    *
    * @param iterator the stream of objects to be serialized
    * @param os the output stream to write the JSON-Stream data to
    */
-  <E> void toJsonStream(Iterator<E> iterator, OutputStream os);
+  default <E> void toJsonStream(Iterator<E> iterator, OutputStream os) {
+    throw new UnsupportedOperationException("toJsonStream is unimplemented in this JsonService");
+  }
 }