Fix buffering issue in StringDecoder

Issue: SPR-17418

Author: poutsma

spring-core/src/main/java/org/springframework/core/codec/StringDecoder.java

@@ -33,6 +33,7 @@
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferUtils;
 import org.springframework.core.io.buffer.DefaultDataBufferFactory;
+import org.springframework.core.io.buffer.PooledDataBuffer;
 import org.springframework.core.log.LogFormatUtils;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
@@ -96,7 +97,8 @@ public Flux<String> decode(Publisher<DataBuffer> inputStream, ResolvableType ele
 		Flux<DataBuffer> inputFlux = Flux.from(inputStream)
 				.flatMap(dataBuffer -> splitOnDelimiter(dataBuffer, delimiterBytes))
 				.bufferUntil(StringDecoder::isEndFrame)
-				.flatMap(StringDecoder::joinUntilEndFrame);
+				.flatMap(StringDecoder::joinUntilEndFrame)
+				.doOnDiscard(PooledDataBuffer.class, DataBufferUtils::release);
 		return super.decode(inputFlux, elementType, mimeType, hints);
 	}
 

spring-core/src/test/java/org/springframework/core/codec/StringDecoderTests.java

@@ -176,8 +176,9 @@ public void decodeEmptyDataBuffer() {
 	@Test
 	public void decodeError() {
 		DataBuffer fooBuffer = stringBuffer("foo\n");
+		DataBuffer barBuffer = stringBuffer("bar");
 		Flux<DataBuffer> source =
-				Flux.just(fooBuffer).concatWith(Flux.error(new RuntimeException()));
+				Flux.just(fooBuffer, barBuffer).concatWith(Flux.error(new RuntimeException()));
 
 		Flux<String> output = this.decoder.decode(source,
 				ResolvableType.forClass(String.class), null, Collections.emptyMap());