Apply doOnDiscard for streaming mode

rstoyanchev · rstoyanchev · commit 28e206a9469b · 2019-04-09T22:30:26.000-04:00
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
diff --git a/spring-web/src/main/java/org/springframework/http/codec/EncoderHttpMessageWriter.java b/spring-web/src/main/java/org/springframework/http/codec/EncoderHttpMessageWriter.java
@@ -131,8 +131,13 @@ public Mono<Void> write(Publisher<? extends T> inputStream, ResolvableType eleme
 					});
 		}
 
-		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

Original file line number	Diff line number	Diff line change
`@@ -131,8 +131,13 @@ public Mono<Void> write(Publisher<? extends T> inputStream, ResolvableType eleme`
`131`	`131`	`});`
`132`	`132`	`}`
`133`	`133`
`134`		`- return (isStreamingMediaType(contentType) ?`
`135`		`- message.writeAndFlushWith(body.map(Flux::just)) : message.writeWith(body));`
	`134`	`+ if (isStreamingMediaType(contentType)) {`
	`135`	`+ return message.writeAndFlushWith(body.map(buffer ->`
	`136`	`+ Mono.fromCallable(() -> buffer)`
	`137`	`+ .doOnDiscard(PooledDataBuffer.class, PooledDataBuffer::release)));`
	`138`	`+ }`
	`139`	`+`
	`140`	`+ return message.writeWith(body);`
`136`	`141`	`}`
`137`	`142`
`138`	`143`	`@Nullable`