* sse: channel.close() now closes sse connection gracefully

Signed-off-by: neo <1100909+neowu@users.noreply.github.com>

Author: neowu

CHANGELOG.md

@@ -6,6 +6,7 @@
 * undertow: updated to 2.3.18.Final
   > due to vulnerability of old versions, has to update to latest despite potential memory consumption is higher
 * sse: support method PUT/POST with body
+* sse: channel.close() now closes sse connection gracefully
 
 ### 9.1.5 (11/11/2024 - 01/22/2025)
 

core-ng/src/main/java/core/framework/internal/web/sse/ChannelImpl.java

@@ -63,7 +63,7 @@ void send(String data) {
         try {
             queue.add(Strings.bytes(data));
             lastSentTime = System.nanoTime();
-            exchange.getIoThread().execute(() -> writeListener.handleEvent(sink));
+            sink.getIoThread().execute(() -> writeListener.handleEvent(sink));
         } finally {
             long elapsed = watch.elapsed();
             ActionLogContext.track("sse", elapsed, 0, data.length());
@@ -79,10 +79,27 @@ public boolean isOpen() {
     @Override
     public void close() {
         LOGGER.debug("close sse connection, channel={}", id);
+        sink.getIoThread().execute(() -> {
+            try {
+                lock.lock();
+                if (closed) return;
+
+                closed = true;
+                if (queue.isEmpty()) {
+                    exchange.endExchange();
+                }
+            } finally {
+                lock.unlock();
+            }
+        });
+    }
+
+    public void shutdown() {
         try {
             lock.lock();
             if (closed) return;
 
+            LOGGER.debug("shutdown sse connection, channel={}", id);
             closed = true;
             queue.clear();
             exchange.endExchange();

core-ng/src/main/java/core/framework/internal/web/sse/ServerSentEventCloseHandler.java

@@ -36,7 +36,7 @@ public void exchangeEvent(HttpServerExchange exchange, NextListener next) {
                 actionLog.correlationIds = refIds;
                 if (!channel.groups.isEmpty()) actionLog.context("group", channel.groups.toArray());
                 context.remove(channel);
-                channel.close();
+                channel.shutdown();
             } catch (Throwable e) {
                 logManager.logError(e);
             } finally {

core-ng/src/main/java/core/framework/internal/web/sse/ServerSentEventHandler.java

@@ -121,7 +121,7 @@ public void shutdown() {
         logger.info("close sse connections");
         for (ChannelSupport<?> support : supports.values()) {
             for (var channel : support.context.all()) {
-                channel.close();
+                ((ChannelImpl<?>) channel).shutdown();
             }
         }
     }

core-ng/src/main/java/core/framework/web/sse/Channel.java

@@ -7,6 +7,7 @@ default void send(T event) {
         send(null, event);
     }
 
+    // gracefully close, queue "end exchange" into io thread
     void close();
 
     void join(String group);

core-ng/src/test/java/core/framework/internal/web/sse/ChannelImplTest.java

@@ -6,6 +6,7 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.xnio.XnioIoThread;
+import org.xnio.channels.StreamSinkChannel;
 
 import java.nio.ByteBuffer;
 
@@ -18,9 +19,10 @@ class ChannelImplTest {
 
     @BeforeEach
     void createServerSentEventChannelImpl() {
+        StreamSinkChannel sink = mock(StreamSinkChannel.class);
+        when(sink.getIoThread()).thenReturn(mock(XnioIoThread.class));
         ServerConnection connection = mock(ServerConnection.class);
-        when(connection.getIoThread()).thenReturn(mock(XnioIoThread.class));
-        channel = new ChannelImpl<>(new HttpServerExchange(connection), null, null, new ServerSentEventBuilder<>(TestEvent.class), null);
+        channel = new ChannelImpl<>(new HttpServerExchange(connection), sink, null, new ServerSentEventBuilder<>(TestEvent.class), null);
     }
 
     @Test