Fix memory leak when canceling read from AsynchronousFileChannel

This commit fixes a memory leak that occurs when reading from a
AsynchronousFileChannel, and cancelling the subscription.

Issue: SPR-17419

Author: poutsma

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

@@ -139,12 +139,14 @@ public static Flux<DataBuffer> readAsynchronousFileChannel(Callable<Asynchronous
 		DataBuffer dataBuffer = dataBufferFactory.allocateBuffer(bufferSize);
 		ByteBuffer byteBuffer = dataBuffer.asByteBuffer(0, bufferSize);
 
+
 		Flux<DataBuffer> result = Flux.using(channelSupplier,
 				channel -> Flux.create(sink -> {
-					CompletionHandler<Integer, DataBuffer> completionHandler =
+					AsynchronousFileChannelReadCompletionHandler completionHandler =
 							new AsynchronousFileChannelReadCompletionHandler(channel,
 									sink, position, dataBufferFactory, bufferSize);
 					channel.read(byteBuffer, position, dataBuffer, completionHandler);
+					sink.onDispose(completionHandler::dispose);
 				}),
 				DataBufferUtils::closeChannel);
 
@@ -545,6 +547,10 @@ public void failed(Throwable exc, DataBuffer dataBuffer) {
 			release(dataBuffer);
 			this.sink.error(exc);
 		}
+
+		public void dispose() {
+			this.disposed.set(true);
+		}
 	}
 
 

spring-core/src/test/java/org/springframework/core/io/buffer/DataBufferUtilsTests.java

@@ -40,6 +40,7 @@
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
+import org.springframework.core.io.ByteArrayResource;
 import org.springframework.core.io.ClassPathResource;
 import org.springframework.core.io.Resource;
 import org.springframework.core.io.buffer.support.DataBufferTestUtils;
@@ -142,6 +143,36 @@ public void readResourcePosition() throws Exception {
 				.verify(Duration.ofSeconds(5));
 	}
 
+	@Test
+	public void readResourcePositionAndTakeUntil() throws Exception {
+		Resource resource = new ClassPathResource("DataBufferUtilsTests.txt", getClass());
+		Flux<DataBuffer> flux = DataBufferUtils.read(resource, 3, this.bufferFactory, 3);
+
+		flux = DataBufferUtils.takeUntilByteCount(flux, 5);
+
+
+		StepVerifier.create(flux)
+				.consumeNextWith(stringConsumer("bar"))
+				.consumeNextWith(stringConsumer("ba"))
+				.expectComplete()
+				.verify(Duration.ofSeconds(5));
+	}
+
+	@Test
+	public void readByteArrayResourcePositionAndTakeUntil() throws Exception {
+		Resource resource = new ByteArrayResource("foobarbazqux" .getBytes());
+		Flux<DataBuffer> flux = DataBufferUtils.read(resource, 3, this.bufferFactory, 3);
+
+		flux = DataBufferUtils.takeUntilByteCount(flux, 5);
+
+
+		StepVerifier.create(flux)
+				.consumeNextWith(stringConsumer("bar"))
+				.consumeNextWith(stringConsumer("ba"))
+				.expectComplete()
+				.verify(Duration.ofSeconds(5));
+	}
+
 	@Test
 	public void writeOutputStream() throws Exception {
 		DataBuffer foo = stringBuffer("foo");