Add a heartbeat executor for SSE emitters

spring-webmvc/src/main/java/org/springframework/web/servlet/mvc/method/annotation/DefaultSseEmitterHeartbeatExecutor.java

@@ -0,0 +1,125 @@
+/*
+ * Copyright 2002-2025 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.web.servlet.mvc.method.annotation;
+
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledFuture;
+
+import org.jspecify.annotations.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.SmartLifecycle;
+import org.springframework.http.MediaType;
+import org.springframework.scheduling.TaskScheduler;
+
+/**
+ * @author Réda Housni Alaoui
+ */
+public class DefaultSseEmitterHeartbeatExecutor implements SmartLifecycle, SseEmitterHeartbeatExecutor {
+
+	private static final Logger LOGGER = LoggerFactory.getLogger(DefaultSseEmitterHeartbeatExecutor.class);
+
+	private final TaskScheduler taskScheduler;
+	private final Set<SseEmitter> emitters = ConcurrentHashMap.newKeySet();
+
+	private final Object lifecycleMonitor = new Object();
+
+	private Duration period = Duration.ofSeconds(5);
+	private String eventName = "ping";
+	private String eventObject = "ping";
+
+	private volatile boolean running;
+	@Nullable
+	private volatile ScheduledFuture<?> taskFuture;
+
+	public DefaultSseEmitterHeartbeatExecutor(TaskScheduler taskScheduler) {
+		this.taskScheduler = taskScheduler;
+	}
+
+	public void setPeriod(Duration period) {
+		this.period = period;
+	}
+
+	public void setEventName(String eventName) {
+		this.eventName = eventName;
+	}
+
+	public void setEventObject(String eventObject) {
+		this.eventObject = eventObject;
+	}
+
+	@Override
+	public void start() {
+		synchronized (lifecycleMonitor) {
+			taskFuture = taskScheduler.scheduleAtFixedRate(this::ping, period);
+			running = true;
+		}
+	}
+
+	@Override
+	public void register(SseEmitter emitter) {
+		Runnable closeCallback = () -> emitters.remove(emitter);
+		emitter.onCompletion(closeCallback);
+		emitter.onError(t -> closeCallback.run());
+		emitter.onTimeout(closeCallback);
+
+		emitters.add(emitter);
+	}
+
+	@Override
+	public void stop() {
+		synchronized (lifecycleMonitor) {
+			ScheduledFuture<?> future = taskFuture;
+			if (future != null) {
+				future.cancel(true);
+			}
+			emitters.clear();
+			running = false;
+		}
+	}
+
+	@Override
+	public boolean isRunning() {
+		return running;
+	}
+
+	boolean isRegistered(SseEmitter emitter) {
+		return emitters.contains(emitter);
+	}
+
+	private void ping() {
+		LOGGER.atDebug().log(() -> "Pinging %s emitter(s)".formatted(emitters.size()));
+
+		for (SseEmitter emitter : emitters) {
+			if (Thread.currentThread().isInterrupted()) {
+				return;
+			}
+			LOGGER.trace("Pinging {}", emitter);
+			SseEmitter.SseEventBuilder eventBuilder = SseEmitter.event().name(eventName).data(eventObject, MediaType.TEXT_PLAIN);
+			try {
+				emitter.send(eventBuilder);
+			} catch (IOException | RuntimeException e) {
+				// According to SseEmitter's Javadoc, the container itself will call SseEmitter#completeWithError
+				LOGGER.debug(e.getMessage());
+			}
+		}
+	}
+}

spring-webmvc/src/main/java/org/springframework/web/servlet/mvc/method/annotation/RequestMappingHandlerAdapter.java

@@ -123,6 +123,7 @@
  * @author Rossen Stoyanchev
  * @author Juergen Hoeller
  * @author Sebastien Deleuze
+ * @author Réda Housni Alaoui
  * @since 3.1
  * @see HandlerMethodArgumentResolver
  * @see HandlerMethodReturnValueHandler
@@ -201,6 +202,8 @@ public class RequestMappingHandlerAdapter extends AbstractHandlerMethodAdapter
 
 	private final Map<ControllerAdviceBean, Set<Method>> modelAttributeAdviceCache = new LinkedHashMap<>();
 
+	@Nullable
+	private SseEmitterHeartbeatExecutor sseEmitterHeartbeatExecutor;
 
 	/**
 	 * Provide resolvers for custom argument types. Custom resolvers are ordered
@@ -526,6 +529,13 @@ public void setParameterNameDiscoverer(ParameterNameDiscoverer parameterNameDisc
 		this.parameterNameDiscoverer = parameterNameDiscoverer;
 	}
 
+	/**
+	 * Set the {@link SseEmitterHeartbeatExecutor} that will be used to periodically prob the SSE connection health
+	 */
+	public void setSseEmitterHeartbeatExecutor(@Nullable SseEmitterHeartbeatExecutor sseEmitterHeartbeatExecutor) {
+		this.sseEmitterHeartbeatExecutor = sseEmitterHeartbeatExecutor;
+	}
+
 	/**
 	 * A {@link ConfigurableBeanFactory} is expected for resolving expressions
 	 * in method argument default values.
@@ -735,7 +745,7 @@ private List<HandlerMethodReturnValueHandler> getDefaultReturnValueHandlers() {
 		handlers.add(new ViewMethodReturnValueHandler());
 		handlers.add(new ResponseBodyEmitterReturnValueHandler(getMessageConverters(),
 				this.reactiveAdapterRegistry, this.taskExecutor, this.contentNegotiationManager,
-				initViewResolvers(), initLocaleResolver()));
+				initViewResolvers(), initLocaleResolver(), this.sseEmitterHeartbeatExecutor));
 		handlers.add(new StreamingResponseBodyReturnValueHandler());
 		handlers.add(new HttpEntityMethodProcessor(getMessageConverters(),
 				this.contentNegotiationManager, this.requestResponseBodyAdvice, this.errorResponseInterceptors));

spring-webmvc/src/main/java/org/springframework/web/servlet/mvc/method/annotation/ResponseBodyEmitterReturnValueHandler.java

@@ -25,6 +25,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Locale;
+import java.util.Optional;
 import java.util.Set;
 import java.util.function.Consumer;
 
@@ -89,6 +90,7 @@
  * </ul>
  *
  * @author Rossen Stoyanchev
+ * @author Réda Housni Alaoui
  * @since 4.2
  */
 public class ResponseBodyEmitterReturnValueHandler implements HandlerMethodReturnValueHandler {
@@ -101,6 +103,8 @@ public class ResponseBodyEmitterReturnValueHandler implements HandlerMethodRetur
 
 	private final LocaleResolver localeResolver;
 
+	@Nullable
+	private final SseEmitterHeartbeatExecutor sseEmitterHeartbeatExecutor;
 
 	/**
 	 * Simple constructor with reactive type support based on a default instance of
@@ -143,11 +147,32 @@ public ResponseBodyEmitterReturnValueHandler(
 			ReactiveAdapterRegistry registry, TaskExecutor executor, ContentNegotiationManager manager,
 			List<ViewResolver> viewResolvers, @Nullable LocaleResolver localeResolver) {
 
+		this(messageConverters, registry, executor, manager, viewResolvers, localeResolver, null);
+	}
+
+	/**
+	 * Constructor that with added arguments for view rendering.
+	 * @param messageConverters converters to write emitted objects with
+	 * @param registry for reactive return value type support
+	 * @param executor for blocking I/O writes of items emitted from reactive types
+	 * @param manager for detecting streaming media types
+	 * @param viewResolvers resolvers for fragment stream rendering
+	 * @param localeResolver the {@link LocaleResolver} for fragment stream rendering
+	 * @param sseEmitterHeartbeatExecutor for sending periodic events to SSE clients
+	 * @since 6.2
+	 */
+	public ResponseBodyEmitterReturnValueHandler(
+			List<HttpMessageConverter<?>> messageConverters,
+			ReactiveAdapterRegistry registry, TaskExecutor executor, ContentNegotiationManager manager,
+			List<ViewResolver> viewResolvers, @Nullable LocaleResolver localeResolver,
+			@Nullable SseEmitterHeartbeatExecutor sseEmitterHeartbeatExecutor) {
+
 		Assert.notEmpty(messageConverters, "HttpMessageConverter List must not be empty");
 		this.sseMessageConverters = initSseConverters(messageConverters);
 		this.reactiveHandler = new ReactiveTypeHandler(registry, executor, manager, null);
 		this.viewResolvers = viewResolvers;
 		this.localeResolver = (localeResolver != null ? localeResolver : new AcceptHeaderLocaleResolver());
+		this.sseEmitterHeartbeatExecutor = sseEmitterHeartbeatExecutor;
 	}
 
 	private static List<HttpMessageConverter<?>> initSseConverters(List<HttpMessageConverter<?>> converters) {
@@ -239,6 +264,9 @@ public void handleReturnValue(@Nullable Object returnValue, MethodParameter retu
 		}
 
 		emitter.initialize(emitterHandler);
+		if (emitter instanceof SseEmitter sseEmitter) {
+			Optional.ofNullable(sseEmitterHeartbeatExecutor).ifPresent(handler -> handler.register(sseEmitter));
+		}
 	}
 
 

spring-webmvc/src/main/java/org/springframework/web/servlet/mvc/method/annotation/SseEmitterHeartbeatExecutor.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright 2002-2025 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.web.servlet.mvc.method.annotation;
+
+/**
+ * @author Réda Housni Alaoui
+ */
+public interface SseEmitterHeartbeatExecutor {
+
+	void register(SseEmitter emitter);
+}