This is an automated email from the ASF dual-hosted git repository.
yuluo-yx pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hertzbeat.git
The following commit(s) were added to refs/heads/master by this push:
new 107f7d02ff fix(log): send initial SSE event and periodic heartbeat
(#4221)
107f7d02ff is described below
commit 107f7d02ffec8af09f0628f336ce23c0a11c7214
Author: shown <[email protected]>
AuthorDate: Mon Jul 20 22:42:25 2026 +0800
fix(log): send initial SSE event and periodic heartbeat (#4221)
Signed-off-by: yuluo-yx <[email protected]>
---
.../apache/hertzbeat/log/notice/LogSseManager.java | 37 ++++++++++++++++++++++
.../log/controller/LogSseControllerTest.java | 27 +++++++++++++++-
.../hertzbeat/log/notice/LogSseManagerTest.java | 27 ++++++++++++++++
3 files changed, 90 insertions(+), 1 deletion(-)
diff --git
a/hertzbeat-log/src/main/java/org/apache/hertzbeat/log/notice/LogSseManager.java
b/hertzbeat-log/src/main/java/org/apache/hertzbeat/log/notice/LogSseManager.java
index d45015a909..78d499cc45 100644
---
a/hertzbeat-log/src/main/java/org/apache/hertzbeat/log/notice/LogSseManager.java
+++
b/hertzbeat-log/src/main/java/org/apache/hertzbeat/log/notice/LogSseManager.java
@@ -38,6 +38,7 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
@@ -51,6 +52,8 @@ import java.util.concurrent.atomic.AtomicLong;
public class LogSseManager {
private static final long BATCH_INTERVAL_MS = 200;
+ private static final long HEARTBEAT_INTERVAL_SECONDS = 15;
+ private static final long RECONNECT_TIME_MS = 5_000;
private static final int MAX_BATCH_SIZE = 1000;
private static final int MAX_QUEUE_SIZE = 10000;
@@ -70,6 +73,8 @@ public class LogSseManager {
public LogSseManager() {
scheduler.scheduleAtFixedRate(this::flushBatch, BATCH_INTERVAL_MS,
BATCH_INTERVAL_MS, TimeUnit.MILLISECONDS);
+ scheduler.scheduleAtFixedRate(this::sendHeartbeat,
HEARTBEAT_INTERVAL_SECONDS,
+ HEARTBEAT_INTERVAL_SECONDS, TimeUnit.SECONDS);
}
@PreDestroy
@@ -96,6 +101,13 @@ public class LogSseManager {
emitter.onError((ex) -> removeEmitter(clientId));
emitters.put(clientId, new SseSubscriber(emitter, filters));
+ try {
+ emitter.send(SseEmitter.event()
+ .comment("connected")
+ .reconnectTime(RECONNECT_TIME_MS));
+ } catch (IOException | IllegalStateException e) {
+ safeComplete(clientId, emitter);
+ }
return emitter;
}
@@ -167,6 +179,31 @@ public class LogSseManager {
}
}
+ void sendHeartbeat() {
+ for (Map.Entry<Long, SseSubscriber> entry : emitters.entrySet()) {
+ Long clientId = entry.getKey();
+ SseEmitter emitter = entry.getValue().emitter;
+ try {
+ senderPool.submit(() -> sendHeartbeat(clientId, emitter));
+ } catch (RejectedExecutionException e) {
+ if (!senderPool.isShutdown()) {
+ log.warn("Failed to schedule SSE heartbeat for client {}",
clientId, e);
+ }
+ }
+ }
+ }
+
+ private void sendHeartbeat(Long clientId, SseEmitter emitter) {
+ try {
+ emitter.send(SseEmitter.event().comment("heartbeat"));
+ } catch (IOException | IllegalStateException e) {
+ safeComplete(clientId, emitter);
+ } catch (Exception e) {
+ log.error("Failed to send SSE heartbeat to client {}", clientId,
e);
+ safeComplete(clientId, emitter);
+ }
+ }
+
private void safeComplete(Long clientId, SseEmitter emitter) {
try {
emitter.complete();
diff --git
a/hertzbeat-log/src/test/java/org/apache/hertzbeat/log/controller/LogSseControllerTest.java
b/hertzbeat-log/src/test/java/org/apache/hertzbeat/log/controller/LogSseControllerTest.java
index 31e1eb3b48..5528ec4f92 100644
---
a/hertzbeat-log/src/test/java/org/apache/hertzbeat/log/controller/LogSseControllerTest.java
+++
b/hertzbeat-log/src/test/java/org/apache/hertzbeat/log/controller/LogSseControllerTest.java
@@ -17,9 +17,11 @@
package org.apache.hertzbeat.log.controller;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.Mockito.verify;
import static
org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
+import static
org.springframework.test.web.servlet.result.MockMvcResultMatchers.request;
import static
org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
import org.apache.hertzbeat.log.notice.LogSseFilterCriteria;
@@ -35,6 +37,7 @@ import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import org.springframework.http.MediaType;
import org.springframework.test.web.servlet.MockMvc;
+import org.springframework.test.web.servlet.MvcResult;
import org.springframework.test.web.servlet.setup.MockMvcBuilders;
/**
@@ -75,6 +78,28 @@ class LogSseControllerTest {
Assertions.assertNull(capturedCriteria.getSpanId());
}
+ @Test
+ void testSubscribeCommitsInitialEvent() throws Exception {
+ LogSseManager realEmitterManager = new LogSseManager();
+ try {
+ MockMvc realMockMvc = MockMvcBuilders
+ .standaloneSetup(new LogSseController(realEmitterManager))
+ .build();
+
+ MvcResult result =
realMockMvc.perform(get("/api/logs/sse/subscribe")
+ .accept(MediaType.TEXT_EVENT_STREAM_VALUE))
+ .andExpect(status().isOk())
+ .andExpect(request().asyncStarted())
+ .andReturn();
+
+ assertTrue(result.getResponse().isCommitted());
+
assertTrue(result.getResponse().getContentAsString().contains(":connected"));
+
assertTrue(result.getResponse().getContentAsString().contains("retry:5000"));
+ } finally {
+ realEmitterManager.shutdown();
+ }
+ }
+
@Test
void testSubscribeWithMultipleFilters() throws Exception {
// Given: Multiple filter parameters
@@ -110,4 +135,4 @@ class LogSseControllerTest {
.accept(MediaType.TEXT_EVENT_STREAM_VALUE))
.andExpect(status().is4xxClientError());
}
-}
\ No newline at end of file
+}
diff --git
a/hertzbeat-log/src/test/java/org/apache/hertzbeat/log/notice/LogSseManagerTest.java
b/hertzbeat-log/src/test/java/org/apache/hertzbeat/log/notice/LogSseManagerTest.java
index 9ddc3b7847..231e8af877 100644
---
a/hertzbeat-log/src/test/java/org/apache/hertzbeat/log/notice/LogSseManagerTest.java
+++
b/hertzbeat-log/src/test/java/org/apache/hertzbeat/log/notice/LogSseManagerTest.java
@@ -72,6 +72,33 @@ class LogSseManagerTest {
assertTrue(logSseManager.getEmitters().containsKey(CLIENT_ID));
}
+ @Test
+ void shouldSendHeartbeatToSubscribers() throws IOException {
+ SseEmitter mockEmitter = mock(SseEmitter.class);
+ subscribeClient(CLIENT_ID, null, mockEmitter);
+
+ logSseManager.sendHeartbeat();
+
+ await().atMost(500, TimeUnit.MILLISECONDS).untilAsserted(() ->
+ verify(mockEmitter).send(any(SseEmitter.SseEventBuilder.class))
+ );
+ }
+
+ @Test
+ void shouldRemoveEmitterWhenHeartbeatFails() throws IOException {
+ SseEmitter mockEmitter = mock(SseEmitter.class);
+ doThrow(new IOException("Connection closed"))
+ .when(mockEmitter).send(any(SseEmitter.SseEventBuilder.class));
+ subscribeClient(CLIENT_ID, null, mockEmitter);
+
+ logSseManager.sendHeartbeat();
+
+ await().atMost(500, TimeUnit.MILLISECONDS).untilAsserted(() -> {
+ verify(mockEmitter).complete();
+ assertFalse(logSseManager.getEmitters().containsKey(CLIENT_ID));
+ });
+ }
+
@Test
void shouldBroadcastLogWhenFilterMatches() throws IOException {
// Given: A client with a filter for "INFO" logs
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
