Apply doOnDiscard for streaming mode

Use of Flux.just is problematic in that if the Flux is cancelled before
demand, the item may never be read, nor freed. Flux#just does not
even delegate cancellation signals.

Closes gh-22731

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

@@ -131,8 +131,13 @@ public class EncoderHttpMessageWriter<T> implements HttpMessageWriter<T> {
 					});
 		}
 
-		return (isStreamingMediaType(contentType) ?
-				message.writeAndFlushWith(body.map(Flux::just)) : message.writeWith(body));
+		if (isStreamingMediaType(contentType)) {
+			return message.writeAndFlushWith(body.map(buffer ->
+					Mono.fromCallable(() -> buffer)
+							.doOnDiscard(PooledDataBuffer.class, PooledDataBuffer::release)));
+		}
+
+		return message.writeWith(body);
 	}
 
 	@Nullable