This is an automated email from the ASF dual-hosted git repository. yuluo-yx pushed a commit to branch 0720-yuluo/fix in repository https://gitbox.apache.org/repos/asf/hertzbeat.git
commit 7c0bc62c8cc689be31516035585c9bab0fce1728 Author: yuluo-yx <[email protected]> AuthorDate: Mon Jul 20 21:50:09 2026 +0800 fix(log): send initial SSE event and periodic heartbeat 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]
