Fix memory leak for ServerSentEventHttpMessageWriter

This commit fixes a memory leak in ServerSentEventHttpMessageWriter
that occurs when the input stream contains an error. Test added as well.

Issue: SPR-17419

Author: poutsma

spring-web/src/main/java/org/springframework/http/codec/ServerSentEventHttpMessageWriter.java

@@ -180,8 +180,8 @@ private <T> Flux<DataBuffer> encodeData(@Nullable T data, ResolvableType valueTy
 	private Mono<DataBuffer> encodeText(CharSequence text, MediaType mediaType, DataBufferFactory bufferFactory) {
 		Assert.notNull(mediaType.getCharset(), "Expected MediaType with charset");
 		byte[] bytes = text.toString().getBytes(mediaType.getCharset());
-		DataBuffer buffer = bufferFactory.allocateBuffer(bytes.length).write(bytes);
-		return Mono.just(buffer);
+		return Mono.defer(() ->
+				Mono.just(bufferFactory.allocateBuffer(bytes.length).write(bytes)));
 	}
 
 	@Override

spring-web/src/test/java/org/springframework/http/codec/ServerSentEventHttpMessageWriterTests.java

@@ -37,7 +37,7 @@
 import org.springframework.mock.http.server.reactive.test.MockServerHttpResponse;
 
 import static org.junit.Assert.*;
-import static org.springframework.core.ResolvableType.*;
+import static org.springframework.core.ResolvableType.forClass;
 
 /**
  * Unit tests for {@link ServerSentEventHttpMessageWriter}.
@@ -81,6 +81,24 @@ public void writeServerSentEvent() {
 				.verify();
 	}
 
+	@Test
+	@SuppressWarnings("rawtypes")
+	public void writeServerSentEventError() {
+		ServerSentEvent<?> event = ServerSentEvent.builder().data("bar").id("c42").event("foo")
+				.comment("bla\nbla bla\nbla bla bla").retry(Duration.ofMillis(123L)).build();
+
+		Flux<ServerSentEvent> source = Flux.concat(
+				Flux.just(event),
+				Flux.error(new RuntimeException()));
+		MockServerHttpResponse outputMessage = new MockServerHttpResponse();
+
+		Mono<Void> result = this.messageWriter.write(source, forClass(ServerSentEvent.class),
+						MediaType.TEXT_EVENT_STREAM, outputMessage, HINTS);
+
+		StepVerifier.create(result)
+				.verifyError(RuntimeException.class);
+	}
+
 	@Test
 	public void writeString() {
 		Flux<String> source = Flux.just("foo", "bar");
@@ -172,8 +190,11 @@ private <T> void testWrite(Publisher<T> source, MockServerHttpResponse response,
 	private <T> void testWrite(
 			Publisher<T> source, MediaType mediaType, MockServerHttpResponse response, Class<T> clazz) {
 
-		this.messageWriter.write(source, forClass(clazz), mediaType, response, HINTS)
-				.block(Duration.ofMillis(5000));
+		Mono<Void> result =
+				this.messageWriter.write(source, forClass(clazz), mediaType, response, HINTS);
+
+		StepVerifier.create(result)
+				.verifyComplete();
 	}
 
 }