refactor(*) : Convert error into JsonNode

- Convert error message into JsonNode
- Change signatures to handle error as a JsonNode
- Update demo code

Author: bordigoni

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

@@ -69,7 +69,7 @@ public static void main(String... args) throws URISyntaxException, InterruptedEx
                                 } else if (event.isPatch()) {
                                     logger.info("RX PATCH {} SNAPSHOT UPDATED {}", event.getPatch(), event.getSnapshot());
                                 } else if (event.isError()) {
-                                    throw new RuntimeException(event.getError());
+                                    throw new RuntimeException(event.getError().toString());
                                 }
                             }, err -> logger.error(err.getMessage(), err));
 

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

@@ -71,7 +71,7 @@ public interface EventSourceClient {
      * @param callback the callback
      * @return this client instance for nice fluent api call
      */
-    EventSourceClient onError(Consumer<String> callback);
+    EventSourceClient onError(Consumer<JsonNode> callback);
 
     /**
      * Sets a callback that is called when a exception is raised :

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

@@ -59,9 +59,9 @@ class Event {
         private EventType type;
         private JsonNode snapshot;
         private JsonNode patch;
-        private String error;
+        private JsonNode error;
 
-        private Event(EventType type, JsonNode snapshot, JsonNode patch, String error) {
+        private Event(EventType type, JsonNode snapshot, JsonNode patch, JsonNode error) {
             this.type = type;
             this.snapshot = snapshot;
             this.patch = patch;
@@ -99,7 +99,7 @@ public static Event forPatch(JsonNode snapshot, JsonNode patch) {
          * @param error the json node
          * @return an Event object
          */
-        public static Event forError(String error) {
+        public static Event forError(JsonNode error) {
             return new Event(EventType.ERROR, null, null, error);
         }
 
@@ -142,7 +142,7 @@ public boolean isPatch() {
             return type == EventType.PATCH;
         }
 
-        public String getError() {
+        public JsonNode getError() {
             return error;
         }
 

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

@@ -39,7 +39,7 @@ public class EventSourceClientImpl implements EventSourceClient {
     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);
+    private Consumer<JsonNode> onErrorCallback = err -> LOGGER.error("A streamdata error has been sent from SSE : {}", err);
     private Consumer<Throwable> onFailureCallback = t -> LOGGER.error("An error occured while processing event", t);
 
     // jackson objectMapper to parse Json content
@@ -123,7 +123,7 @@ public EventSourceClient onPatch(Consumer<JsonNode> callback) {
     }
 
     @Override
-    public EventSourceClient onError(Consumer<String> callback) {
+    public EventSourceClient onError(Consumer<JsonNode> callback) {
         this.onErrorCallback = callback;
         return this;
     }
@@ -224,8 +224,16 @@ public void onEvent(InboundEvent inboundEvent) {
                             break;
 
                         case "error":
-                            LOGGER.debug("Receiving error {} ", eventData);
-                            onErrorCallback.accept(eventData);
+                            try {
+                                LOGGER.debug("Receiving error {} ", eventData);
+
+                                // read the error
+                                JsonNode lastPatch = jsonObjectMapper.readTree(eventData);
+
+                                onErrorCallback.accept(lastPatch);
+                            } catch (IOException e) {
+                                onFailureCallback.accept(e);
+                            }
                             break;
 
                         default: