feat(sse): preserve frames and classify consumer failures

Author: loong10k

src/main/java/io/github/easy4j/hermes/api/HermesSseClient.java

@@ -4,7 +4,10 @@
 import io.github.easy4j.hermes.HermesHttpClientConfig;
 import io.github.easy4j.hermes.HermesOkHttpClientFactory;
 import io.github.easy4j.hermes.api.model.ChatRequest;
+import io.github.easy4j.hermes.api.sse.EndpointEventDecoder;
+import io.github.easy4j.hermes.api.sse.SseConsumerException;
 import io.github.easy4j.hermes.api.sse.SseEvent;
+import io.github.easy4j.hermes.api.sse.SseFrame;
 import io.github.easy4j.hermes.api.sse.SseQueueSubscription;
 import io.github.easy4j.hermes.api.sse.SseSubscription;
 import io.github.easy4j.hermes.exception.HermesHttpException;
@@ -74,6 +77,8 @@ public class HermesSseClient implements AutoCloseable {
      * SSE 事件反序列化使用的 ObjectMapper。
      */
     private final ObjectMapper mapper;
+    /** 将保留的原始帧转换为当前兼容事件视图。 */
+    private final EndpointEventDecoder<SseEvent> eventDecoder;
     /**
      * 执行 HTTP 请求的 OkHttpClient。
      */
@@ -115,6 +120,10 @@ public HermesSseClient(HermesHttpClientConfig config, ObjectMapper objectMapper,
         this.config = Objects.requireNonNull(config, "config");
         this.mapper = Objects.isNull(objectMapper) ? new ObjectMapper()
                 .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false) : objectMapper;
+        this.eventDecoder = frame -> {
+            this.mapper.readTree(frame.getData());
+            return SseEvent.fromFrame(frame);
+        };
         this.ownsHttpClient = Objects.isNull(httpClient);
         this.ownsDispatcher = Objects.nonNull(httpClient);
         OkHttpClient baseClient = this.ownsHttpClient ? HermesOkHttpClientFactory.create(config) : httpClient;
@@ -277,8 +286,10 @@ public void onEvent(EventSource eventSource, String id, String type, String data
                     if (data == null || data.isEmpty()) {
                         return;
                     }
+                    SseFrame frame = new SseFrame(id, type, data, System.currentTimeMillis());
+                    SseEvent event;
                     try {
-                        mapper.readTree(data);
+                        event = eventDecoder.decode(frame);
                     } catch (Exception error) {
                         if (config.getDebug().allows(HttpLogLevel.BODY)) {
                             log.debug("Hermes SSE parse failed: label={}, data={}", label, truncate(data), error);
@@ -289,18 +300,22 @@ public void onEvent(EventSource eventSource, String id, String type, String data
                         return;
                     }
 
-                    SseEvent event = new SseEvent();
-                    event.setEvent(type);
-                    event.setData(data);
                     try {
                         consumer.accept(event);
                     } catch (Exception error) {
+                        SseConsumerException failure = new SseConsumerException(frame, error);
                         if (config.getDebug().allows(HttpLogLevel.BODY)) {
                             log.debug("Hermes SSE consumer failed: label={}, data={}", label, truncate(data), error);
                         } else {
                             debug(HttpLogLevel.BASIC, "Hermes SSE consumer failed: label={}, dataLength={}, error={}",
                                     label, data.length(), error.getMessage());
                         }
+                        terminalSignal.set(true);
+                        try {
+                            onError.accept(failure);
+                        } finally {
+                            finish(subscription);
+                        }
                     }
                 }