add(EventSourceClient) : add a callback when the connection is closed

- add the method in the interface
- implement doing nothing by default

Author: bordigoni

src/main/java/io/streamdata/demo/Main.java

@@ -30,6 +30,7 @@ public static void main(String... args) throws URISyntaxException, InterruptedEx
                     .onSnapshot(data -> logger.info("INITIAL SNAPSHOT {}", data))
                     .onPatch(patch -> logger.info("PATCH {} SNAPSHOT UPDATED {}", patch, eventSource.getCurrentSnapshot()))
                     .onOpen(() -> logger.info("And we are... live!"))
+                    .onClose(() -> logger.info("Bye now!"))
                     .open();
 
             Thread.sleep(10000);

src/main/java/io/streamdata/sdk/EventSourceClient.java

@@ -39,6 +39,14 @@ public interface EventSourceClient {
      */
     EventSourceClient onOpen(Runnable onOpen);
 
+    /**
+     * Sets a optionnal callback to be called after the event source has been closed
+     *
+     * @param onClose the callback
+     * @return this client instance for nice fluent api call
+     */
+    EventSourceClient onClose(Runnable onClose);
+
     /**
      * Sets a callback to be called after streamdata sends after the first time the API is polled.
      * <b>This callback must be set before calling {@link #open()}</b>

src/main/java/io/streamdata/sdk/impl/EventSourceClientImpl.java

@@ -36,6 +36,7 @@ public class EventSourceClientImpl implements EventSourceClient {
 
     // The polling URL
     private Runnable onOpenCallback;
+    private Runnable onCloseCallback;
     private Consumer<JsonNode> onDataCallback;
     private Consumer<JsonNode> onPatchCallback;
     private Consumer<String> onErrorCallback = err -> LOGGER.error("A streamdata error has been sent from SSE : {}", err);
@@ -103,6 +104,12 @@ public EventSourceClient onOpen(Runnable callback) {
         return this;
     }
 
+    @Override
+    public EventSourceClient onClose(Runnable callback) {
+        this.onCloseCallback = callback;
+        return this;
+    }
+
     @Override
     public EventSourceClient onSnapshot(Consumer<JsonNode> snaphot) {
         this.onDataCallback = snaphot;
@@ -131,6 +138,9 @@ public EventSourceClient onException(Consumer<Throwable> callback) {
     public void close() {
         if (this.eventSource != null) {
             this.eventSource.close();
+            if (this.onCloseCallback != null) {
+                this.onCloseCallback.run();
+            }
             this.eventSource = null;
         }
     }