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 0d6ff4f21cb2..3267de380eed 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 @@ -724,6 +724,10 @@ private void read() { @Override public void completed(Integer read, Attachment attachment) { + if (!attachment.markHandled()) { + return; + } + attachment.iterator().close(); DataBuffer dataBuffer = attachment.dataBuffer(); @@ -735,8 +739,8 @@ public void completed(Integer read, Attachment attachment) { if (read == -1) { release(dataBuffer); - closeChannel(this.channel); this.state.set(State.DISPOSED); + closeChannel(this.channel); this.sink.complete(); return; } @@ -759,11 +763,15 @@ public void completed(Integer read, Attachment attachment) { @Override public void failed(Throwable ex, Attachment attachment) { + if (!attachment.markHandled()) { + return; + } + attachment.iterator().close(); release(attachment.dataBuffer()); - closeChannel(this.channel); this.state.set(State.DISPOSED); + closeChannel(this.channel); this.sink.error(ex); } @@ -771,7 +779,17 @@ private enum State { IDLE, READING, DISPOSED } - private record Attachment(DataBuffer dataBuffer, DataBuffer.ByteBufferIterator iterator) {} + private record Attachment(DataBuffer dataBuffer, DataBuffer.ByteBufferIterator iterator, + AtomicBoolean handled) { + + Attachment(DataBuffer dataBuffer, DataBuffer.ByteBufferIterator iterator) { + this(dataBuffer, iterator, new AtomicBoolean()); + } + + boolean markHandled() { + return this.handled.compareAndSet(false, 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 abff1e95b512..99c684432c02 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 @@ -22,6 +22,7 @@ import java.io.OutputStreamWriter; import java.net.URI; import java.nio.ByteBuffer; +import java.nio.channels.AsynchronousCloseException; import java.nio.channels.AsynchronousFileChannel; import java.nio.channels.CompletionHandler; import java.nio.channels.FileChannel; @@ -39,6 +40,9 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executors; import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.BiConsumer; import io.netty.buffer.ByteBuf; import io.netty.buffer.PooledByteBufAllocator; @@ -61,11 +65,14 @@ import static org.assertj.core.api.Assertions.assertThatIOException; import static org.assertj.core.api.Assertions.fail; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.BDDMockito.given; import static org.mockito.BDDMockito.willAnswer; import static org.mockito.BDDMockito.willThrow; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; /** * @author Arjen Poutsma @@ -262,6 +269,58 @@ void readAsynchronousFileChannelReadThrowsSynchronouslyFromCompletionThread() th } } + @Test + void readAsynchronousFileChannelCompletedTwiceAfterEof() throws Exception { + readAsynchronousFileChannelWithDuplicateCallbackOnClose( + (handler, attachment) -> handler.completed(-1, attachment)); + } + + @Test + void readAsynchronousFileChannelFailedAfterEof() throws Exception { + readAsynchronousFileChannelWithDuplicateCallbackOnClose( + (handler, attachment) -> handler.failed(new AsynchronousCloseException(), attachment)); + } + + private void readAsynchronousFileChannelWithDuplicateCallbackOnClose( + BiConsumer, Object> duplicateCallback) throws Exception { + + DataBuffer delegate = DefaultDataBufferFactory.sharedInstance.allocateBuffer(3); + PooledDataBuffer dataBuffer = mock(); + given(dataBuffer.writableByteBuffers()).willAnswer(invocation -> delegate.writableByteBuffers()); + given(dataBuffer.isAllocated()).willReturn(true); + + DataBufferFactory factory = mock(); + given(factory.allocateBuffer(anyInt())).willReturn(dataBuffer); + + AtomicReference> handlerRef = new AtomicReference<>(); + AtomicReference attachmentRef = new AtomicReference<>(); + + AtomicBoolean open = new AtomicBoolean(true); + AsynchronousFileChannel channel = mock(); + given(channel.isOpen()).willAnswer(invocation -> open.get()); + willAnswer(invocation -> { + open.set(false); + duplicateCallback.accept(handlerRef.get(), attachmentRef.get()); + return null; + }).given(channel).close(); + willAnswer(invocation -> { + Object attachment = invocation.getArgument(2); + CompletionHandler handler = invocation.getArgument(3); + handlerRef.set(handler); + attachmentRef.set(attachment); + handler.completed(-1, attachment); + return null; + }).given(channel).read(any(), anyLong(), any(), any()); + + Flux result = DataBufferUtils.readAsynchronousFileChannel(() -> channel, factory, 3); + + StepVerifier.create(result) + .expectComplete() + .verify(Duration.ofSeconds(3)); + + verify(dataBuffer, times(1)).release(); + } + @ParameterizedDataBufferAllocatingTest void readAsynchronousFileChannelCancel(DataBufferFactory bufferFactory) throws Exception { super.bufferFactory = bufferFactory;