Polishing contribution

This commit adds further fixes in the same area, since there were
similar bugs in the WriteCompletionHandler:
* databuffers were not always emitted when fully read in the onNext hook
* on completion, the iterator was closed too early, before it was fully
  read
* on completion, writing the next bytebuffers from the iterator would
  always reuse the first one and not update the attachment

Closes gh-36714
This commit is contained in:
Brian Clozel
2026-06-04 12:09:52 +02:00
parent 244f6ec390
commit f161a419bd
2 changed files with 29 additions and 2 deletions
@@ -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);
@@ -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<DataBuffer> 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;