Eliminate windowUntil from StringDecoder

This is a follow-up on the earlier commit
28a95e89f3 eliminating windowUntil
entirely which generates a BubblingException wrapper. This also keeps
the chain a little simpler.

See gh-24355

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

@@ -32,6 +32,7 @@ import reactor.core.publisher.Flux;
 
 import org.springframework.core.ResolvableType;
 import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferLimitException;
 import org.springframework.core.io.buffer.DataBufferUtils;
 import org.springframework.core.io.buffer.DataBufferWrapper;
 import org.springframework.core.io.buffer.DefaultDataBufferFactory;
@@ -96,22 +97,13 @@ public final class StringDecoder extends AbstractDataBufferDecoder<String> {
 		Flux<DataBuffer> inputFlux = Flux.defer(() -> {
 			DataBufferUtils.Matcher matcher = DataBufferUtils.matcher(delimiterBytes);
 
-			Flux<DataBuffer> buffers = Flux.from(input)
-					.concatMapIterable(buffer -> endFrameAfterDelimiter(buffer, matcher));
+			@SuppressWarnings("MismatchedQueryAndUpdateOfCollection")
+			LimitChecker limiter = new LimitChecker(getMaxInMemorySize());
 
-			Flux<List<DataBuffer>> delimitedBuffers;
-			if (getMaxInMemorySize() != -1) {
-				delimitedBuffers = buffers
-						.windowUntil(buffer -> buffer instanceof EndFrameBuffer)
-						.concatMap(window -> window.collect(
-								() -> new LimitedDataBufferList(getMaxInMemorySize()),
-								LimitedDataBufferList::add));
-			}
-			else {
-				delimitedBuffers = buffers.bufferUntil(buffer -> buffer instanceof EndFrameBuffer);
-			}
-
-			return delimitedBuffers
+			return Flux.from(input)
+					.concatMapIterable(buffer -> endFrameAfterDelimiter(buffer, matcher))
+					.doOnNext(limiter)
+					.bufferUntil(buffer -> buffer instanceof EndFrameBuffer)
 					.map(list -> joinAndStrip(list, this.stripDelimiter))
 					.doOnDiscard(PooledDataBuffer.class, DataBufferUtils::release);
 		});
@@ -205,7 +197,7 @@ public final class StringDecoder extends AbstractDataBufferDecoder<String> {
 		DataBuffer lastBuffer = dataBuffers.get(lastIdx);
 		if (lastBuffer instanceof EndFrameBuffer) {
 			matchingDelimiter = ((EndFrameBuffer) lastBuffer).delimiter();
-			dataBuffers = dataBuffers.subList(0, lastIdx);
+			dataBuffers.remove(lastIdx);
 		}
 
 		DataBuffer result = dataBuffers.get(0).factory().join(dataBuffers);
@@ -296,31 +288,28 @@ public final class StringDecoder extends AbstractDataBufferDecoder<String> {
 	}
 
 
-	private class ConcatMapIterableDiscardWorkaroundCache implements Consumer<DataBuffer>, Runnable {
+	private static class LimitChecker implements Consumer<DataBuffer> {
 
-		private final List<DataBuffer> buffers = new ArrayList<>();
+		@SuppressWarnings("MismatchedQueryAndUpdateOfCollection")
+		private final LimitedDataBufferList list;
 
 
-		public List<DataBuffer> addAll(List<DataBuffer> buffersToAdd) {
-			this.buffers.addAll(buffersToAdd);
-			return buffersToAdd;
-		}
-
-		@Override
-		public void accept(DataBuffer dataBuffer) {
-			this.buffers.remove(dataBuffer);
+		LimitChecker(int maxInMemorySize) {
+			this.list = new LimitedDataBufferList(maxInMemorySize);
 		}
 
 		@Override
-		public void run() {
-			this.buffers.forEach(buffer -> {
-				try {
-					DataBufferUtils.release(buffer);
-				}
-				catch (Throwable ex) {
-					// Keep going..
-				}
-			});
+		public void accept(DataBuffer buffer) {
+			if (buffer instanceof EndFrameBuffer) {
+				this.list.clear();
+			}
+			try {
+				this.list.add(buffer);
+			}
+			catch (DataBufferLimitException ex) {
+				DataBufferUtils.release(buffer);
+				throw ex;
+			}
 		}
 	}
 

spring-core/src/main/java/org/springframework/core/io/buffer/DataBufferUtils.java

@@ -547,13 +547,10 @@ public abstract class DataBufferUtils {
 			return (Mono<DataBuffer>) buffers;
 		}
 
-		// TODO: Drop doOnDiscard(LimitedDataBufferList.class, ...) (reactor-core#1924)
-
 		return Flux.from(buffers)
 				.collect(() -> new LimitedDataBufferList(maxByteCount), LimitedDataBufferList::add)
 				.filter(list -> !list.isEmpty())
 				.map(list -> list.get(0).factory().join(list))
-				.doOnDiscard(LimitedDataBufferList.class, LimitedDataBufferList::releaseAndClear)
 				.doOnDiscard(PooledDataBuffer.class, DataBufferUtils::release);
 	}
 

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

@@ -78,11 +78,11 @@ class StringDecoderTests extends AbstractDecoderTests<StringDecoder> {
 		// TODO: temporarily replace testDecodeAll with explicit decode/cancel/empty
 		// see https://github.com/reactor/reactor-core/issues/2041
 
-		testDecode(input, TYPE, step -> step.expectNext(u, e, o).verifyComplete(), null, null);
-		testDecodeCancel(input, TYPE, null, null);
-		testDecodeEmpty(TYPE, null, null);
+//		testDecode(input, TYPE, step -> step.expectNext(u, e, o).verifyComplete(), null, null);
+//		testDecodeCancel(input, TYPE, null, null);
+//		testDecodeEmpty(TYPE, null, null);
 
-		// testDecodeAll(input, TYPE, step -> step.expectNext(u, e, o).verifyComplete(), null, null);
+		testDecodeAll(input, TYPE, step -> step.expectNext(u, e, o).verifyComplete(), null, null);
 	}
 
 	@Test