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 c1a3a07f4c1..24a6e631f54 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 @@ -1215,6 +1215,11 @@ public abstract class DataBufferUtils { failed(ex, attachment); } } + else { + iterator.close(); + this.sink.next(dataBuffer); + request(1); + } } @Override @@ -1238,7 +1243,6 @@ public abstract class DataBufferUtils { @Override public void completed(Integer written, Attachment attachment) { DataBuffer.ByteBufferIterator iterator = attachment.iterator(); - iterator.close(); long pos = this.position.addAndGet(written); ByteBuffer byteBuffer = attachment.byteBuffer(); @@ -1248,9 +1252,11 @@ public abstract class DataBufferUtils { } else if (iterator.hasNext()) { ByteBuffer next = iterator.next(); - this.channel.write(next, pos, attachment, this); + Attachment nextAttachment = new Attachment(next, attachment.dataBuffer(), iterator); + this.channel.write(next, pos, nextAttachment, this); } else { + iterator.close(); this.sink.next(attachment.dataBuffer()); this.writing.set(false); 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 35ebdd9c3ae..6c0be595624 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 @@ -466,6 +466,27 @@ class DataBufferUtilsTests extends AbstractDataBufferAllocatingTests { assertThat(result).isEqualTo("foobarbazqux"); } + @ParameterizedDataBufferAllocatingTest + void writeAsynchronousFileChannelWithJoinedBuffer(DataBufferFactory bufferFactory) throws Exception { + super.bufferFactory = bufferFactory; + + DataBuffer foo = stringBuffer("foo"); + DataBuffer bar = stringBuffer("bar"); + DataBuffer joined = bufferFactory.join(List.of(foo, bar)); + + AsynchronousFileChannel channel = AsynchronousFileChannel.open(tempFile, StandardOpenOption.WRITE); + + Flux writeResult = DataBufferUtils.write(Flux.just(joined), channel); + StepVerifier.create(writeResult) + .consumeNextWith(stringConsumer("foobar")) + .verifyComplete(); + + String result = String.join("", Files.readAllLines(tempFile)); + + assertThat(result).isEqualTo("foobar"); + channel.close(); + } + @ParameterizedDataBufferAllocatingTest void writeAsynchronousFileChannelErrorInFlux(DataBufferFactory bufferFactory) throws Exception { super.bufferFactory = bufferFactory;