perf: avoid unused sse event buffering

Author: loong10k

src/main/java/io/github/easy4j/hermes/HermesClient.java

@@ -445,8 +445,7 @@ public StreamingChatResponse chatCompletionStream(ChatRequest request,
                                                       Map<String, String> headers,
                                                       Consumer<String> deltaConsumer) {
         Objects.requireNonNull(request, "request");
-        StreamingChatResponse stream = new StreamingChatResponse(
-                config.getHttp().getStreamEventQueueCapacity());
+        StreamingChatResponse stream = new StreamingChatResponse();
         stream.onDelta(deltaConsumer);
         SseSubscription subscription = sseClient.subscribeChat(
                 request.withStream(), headers, stream::accept, stream::finish, stream::fail);

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

@@ -22,29 +22,28 @@
  */
 public class HermesChatClient extends HermesHttpClient {
 
-    private final HermesHttpClientConfig config;
     private final HermesSseClient sseClient;
     private final boolean ownsSseClient;
 
     public HermesChatClient(HermesHttpClientConfig config) {
         super(config);
-        this.config = Objects.requireNonNull(config, "config");
+        Objects.requireNonNull(config, "config");
         this.sseClient = new HermesSseClient(config, getObjectMapper(), getOkHttpClient());
         this.ownsSseClient = true;
     }
 
     public HermesChatClient(HermesHttpClientConfig config, ObjectMapper objectMapper,
                             OkHttpClient httpClient) {
         super(config, objectMapper, httpClient);
-        this.config = Objects.requireNonNull(config, "config");
+        Objects.requireNonNull(config, "config");
         this.sseClient = new HermesSseClient(config, getObjectMapper(), getOkHttpClient());
         this.ownsSseClient = true;
     }
 
     public HermesChatClient(HermesHttpClientConfig config, ObjectMapper objectMapper,
                             OkHttpClient httpClient, HermesSseClient sseClient) {
         super(config, objectMapper, httpClient);
-        this.config = Objects.requireNonNull(config, "config");
+        Objects.requireNonNull(config, "config");
         this.sseClient = Objects.requireNonNull(sseClient, "sseClient");
         this.ownsSseClient = false;
     }
@@ -63,8 +62,7 @@ public StreamingChatResponse chatCompletionStream(ChatRequest request,
                                                        Map<String, String> headers,
                                                        Consumer<String> deltaConsumer) {
         Objects.requireNonNull(request, "request");
-        StreamingChatResponse stream = new StreamingChatResponse(
-                config.getStreamEventQueueCapacity()).onDelta(deltaConsumer);
+        StreamingChatResponse stream = new StreamingChatResponse().onDelta(deltaConsumer);
         SseSubscription subscription = sseClient.subscribeChat(
                 request.withStream(), headers, stream::accept, stream::finish, stream::fail);
         stream.onCancel(subscription::close);

src/main/java/io/github/easy4j/hermes/api/sse/StreamingChatResponse.java

@@ -3,10 +3,6 @@
  * @author <a href="https://github.com/loong10k">@Loong Wan</a>
  */
 
-import lombok.Getter;
-
-import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.Objects;
@@ -20,15 +16,9 @@ public class StreamingChatResponse extends CompletableFuture<String> {
     private final StringBuilder content = new StringBuilder();
     private Consumer<String> deltaConsumer;
     private final AtomicReference<Runnable> cancellation = new AtomicReference<>();
-    @Getter
-    private final BlockingQueue<SseEvent> eventQueue;
 
+    /** 创建不缓存原始事件的流式响应。 */
     public StreamingChatResponse() {
-        this(1_024);
-    }
-
-    public StreamingChatResponse(int eventQueueCapacity) {
-        this.eventQueue = new ArrayBlockingQueue<>(Math.max(1, eventQueueCapacity));
     }
 
     public StreamingChatResponse onDelta(Consumer<String> consumer) {
@@ -37,10 +27,6 @@ public StreamingChatResponse onDelta(Consumer<String> consumer) {
     }
 
     public void accept(SseEvent event) {
-        if (!eventQueue.offer(event)) {
-            eventQueue.poll();
-            eventQueue.offer(event);
-        }
         String delta = event.deltaText();
         if (delta != null) {
             content.append(delta);

src/test/java/io/github/easy4j/hermes/HermesSseAndModelTest.java

@@ -63,7 +63,6 @@ void shouldParseSseEventsAndAccumulateStreamingContent() throws Exception {
         stream.accept(chat);
         stream.accept(direct);
         stream.accept(event("{}"));
-        assertEquals(3, stream.getEventQueue().size());
         assertEquals("hello world", stream.getAccumulatedContent());
         stream.finish();
         assertEquals("hello world", stream.get(1, TimeUnit.SECONDS));
@@ -261,6 +260,20 @@ void shouldKeepLatestQueueEventAndExposeSubscriptionLifecycle() throws Exception
                 assertEquals(0, sse.activeSubscriptionCount());
             }
 
+            server.enqueue(new MockResponse().setResponseCode(200)
+                    .setHeader("Content-Type", "text/event-stream")
+                    .setBody("data: [DONE]\n\n"));
+            try (HermesSseClient sse = new HermesSseClient(config, null, null)) {
+                SseSubscription completed = sse.subscribeSessionEvents(
+                        "session-complete", "hello", event -> { });
+                assertNotNull(server.takeRequest(3, TimeUnit.SECONDS));
+                long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(3);
+                while (completed.isActive() && System.nanoTime() < deadline) {
+                    Thread.yield();
+                }
+                assertFalse(completed.isActive());
+            }
+
             server.enqueue(new MockResponse().setResponseCode(200)
                     .setHeader("Content-Type", "text/event-stream")
                     .setBodyDelay(3, TimeUnit.SECONDS).setBody("data: [DONE]\n\n"));
@@ -294,12 +307,6 @@ void shouldCoverChatStreamingOverloadsAndCancellationPaths() throws Exception {
             HermesOkHttpClientFactory.shutdown(client);
         }
 
-        StreamingChatResponse bounded = new StreamingChatResponse(1);
-        bounded.accept(event("{\"delta\":\"first\"}"));
-        SseEvent latest = event("{\"delta\":\"latest\"}");
-        bounded.accept(latest);
-        assertEquals(latest, bounded.getEventQueue().poll());
-
         AtomicInteger cancellations = new AtomicInteger();
         StreamingChatResponse cancelBeforeBind = new StreamingChatResponse();
         cancelBeforeBind.cancel(false);

src/test/java/io/github/easy4j/hermes/HermesSseApiShapeTest.java

@@ -33,6 +33,8 @@ void shouldExposeOnlyTheUnifiedSseTopology() {
         assertFalse(hasMethod(HermesSseClient.class, "subscribeSessionStream"));
         assertFalse(hasMethod(HermesSseClient.class, "stop"));
         assertFalse(hasMethod(HermesChatClient.class, "events"));
+        assertFalse(hasMethod(io.github.easy4j.hermes.api.sse.StreamingChatResponse.class,
+                "getEventQueue"));
         for (Field field : HermesChatClient.class.getDeclaredFields()) {
             assertFalse("eventClient".equals(field.getName()));
         }