From eac9e66c4688a97fd21b45875c31a85b83d748df Mon Sep 17 00:00:00 2001 From: Arjen Poutsma Date: Tue, 23 Oct 2018 16:23:12 +0200 Subject: [PATCH] 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 --- .../core/io/buffer/DataBufferUtils.java | 8 ++++- .../core/io/buffer/DataBufferUtilsTests.java | 31 +++++++++++++++++++ 2 files changed, 38 insertions(+), 1 deletion(-) diff --git a/spring-core/src/main/java/org/springframework/core/io/buffer/DataBufferUtils.java b/spring-core/src/main/java/org/springframework/core/io/buffer/DataBufferUtils.java index e048576f96..198bbfa024 100644 --- a/spring-core/src/main/java/org/springframework/core/io/buffer/DataBufferUtils.java +++ b/spring-core/src/main/java/org/springframework/core/io/buffer/DataBufferUtils.java @@ -139,12 +139,14 @@ public abstract class DataBufferUtils { DataBuffer dataBuffer = dataBufferFactory.allocateBuffer(bufferSize); ByteBuffer byteBuffer = dataBuffer.asByteBuffer(0, bufferSize); + Flux result = Flux.using(channelSupplier, channel -> Flux.create(sink -> { - CompletionHandler 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 abstract class DataBufferUtils { release(dataBuffer); this.sink.error(exc); } + + public void dispose() { + this.disposed.set(true); + } } diff --git a/spring-core/src/test/java/org/springframework/core/io/buffer/DataBufferUtilsTests.java b/spring-core/src/test/java/org/springframework/core/io/buffer/DataBufferUtilsTests.java index 8f53d733fe..ff511291db 100644 --- a/spring-core/src/test/java/org/springframework/core/io/buffer/DataBufferUtilsTests.java +++ b/spring-core/src/test/java/org/springframework/core/io/buffer/DataBufferUtilsTests.java @@ -40,6 +40,7 @@ import reactor.core.publisher.Flux; 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 class DataBufferUtilsTests extends AbstractDataBufferAllocatingTestCase { .verify(Duration.ofSeconds(5)); } + @Test + public void readResourcePositionAndTakeUntil() throws Exception { + Resource resource = new ClassPathResource("DataBufferUtilsTests.txt", getClass()); + Flux 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 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");