From 9a1bba5b67e2dcfbd024c36ae12e9e5dbe664ddc Mon Sep 17 00:00:00 2001 From: Rossen Stoyanchev Date: Fri, 29 Mar 2019 15:25:28 -0400 Subject: [PATCH 1/3] NettyDataBufferFactory.wrap(ByteBuf) calls touch() Closes gh-21960 --- .../springframework/core/io/buffer/NettyDataBufferFactory.java | 1 + 1 file changed, 1 insertion(+) diff --git a/spring-core/src/main/java/org/springframework/core/io/buffer/NettyDataBufferFactory.java b/spring-core/src/main/java/org/springframework/core/io/buffer/NettyDataBufferFactory.java index 9d9e0a3993..14931d4243 100644 --- a/spring-core/src/main/java/org/springframework/core/io/buffer/NettyDataBufferFactory.java +++ b/spring-core/src/main/java/org/springframework/core/io/buffer/NettyDataBufferFactory.java @@ -90,6 +90,7 @@ public class NettyDataBufferFactory implements DataBufferFactory { * @return the wrapped buffer */ public NettyDataBuffer wrap(ByteBuf byteBuf) { + byteBuf.touch(); return new NettyDataBuffer(byteBuf, this); } From 81f95efdbb7481a8a622648e7462b373597c762b Mon Sep 17 00:00:00 2001 From: Rossen Stoyanchev Date: Fri, 29 Mar 2019 15:43:06 -0400 Subject: [PATCH 2/3] Call onDispose before first read The cancellation callback in asynchronousReadFileChannel must be called before the first read I/O or otherwise if cancellation signals happens immediately the onDispose callback may be missed. The DefaultBufferFactory workaround however remains in place until an expected additional fix arrives with Reactor Core 3.2.9. See gh-22107 --- .../org/springframework/core/io/buffer/DataBufferUtils.java | 2 +- 1 file changed, 1 insertion(+), 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 4fa9beba58..af9015db6d 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 @@ -144,10 +144,10 @@ public abstract class DataBufferUtils { channel -> Flux.create(sink -> { ReadCompletionHandler handler = new ReadCompletionHandler(channel, sink, position, bufferFactoryToUse, bufferSize); + sink.onDispose(handler::dispose); DataBuffer dataBuffer = bufferFactoryToUse.allocateBuffer(bufferSize); ByteBuffer byteBuffer = dataBuffer.asByteBuffer(0, bufferSize); channel.read(byteBuffer, position, dataBuffer, handler); - sink.onDispose(handler::dispose); }), channel -> { // Do not close channel from here, rather wait for the current read callback From 85332c7a49dd112d3b6e71f3458edaa3f05d8b4d Mon Sep 17 00:00:00 2001 From: Rossen Stoyanchev Date: Fri, 29 Mar 2019 15:53:45 -0400 Subject: [PATCH 3/3] Polish --- .../core/io/buffer/DataBufferUtils.java | 138 ++++++++++-------- 1 file changed, 75 insertions(+), 63 deletions(-) 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 af9015db6d..359cf189c4 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 @@ -69,28 +69,29 @@ public abstract class DataBufferUtils { //--------------------------------------------------------------------- /** - * Obtain a {@link InputStream} from the given supplier, and read it into a {@code Flux} - * of {@code DataBuffer}s. Closes the input stream when the flux is terminated. + * Obtain a {@link InputStream} from the given supplier, and read it into a + * {@code Flux} of {@code DataBuffer}s. Closes the input stream when the + * Flux is terminated. * @param inputStreamSupplier the supplier for the input stream to read from - * @param dataBufferFactory the factory to create data buffers with + * @param bufferFactory the factory to create data buffers with * @param bufferSize the maximum size of the data buffers - * @return a flux of data buffers read from the given channel + * @return a Flux of data buffers read from the given channel */ public static Flux readInputStream( - Callable inputStreamSupplier, DataBufferFactory dataBufferFactory, int bufferSize) { + Callable inputStreamSupplier, DataBufferFactory bufferFactory, int bufferSize) { Assert.notNull(inputStreamSupplier, "'inputStreamSupplier' must not be null"); - - return readByteChannel(() -> Channels.newChannel(inputStreamSupplier.call()), dataBufferFactory, bufferSize); + return readByteChannel(() -> Channels.newChannel(inputStreamSupplier.call()), bufferFactory, bufferSize); } /** - * Obtain a {@link ReadableByteChannel} from the given supplier, and read it into a - * {@code Flux} of {@code DataBuffer}s. Closes the channel when the flux is terminated. + * Obtain a {@link ReadableByteChannel} from the given supplier, and read + * it into a {@code Flux} of {@code DataBuffer}s. Closes the channel when + * the Flux is terminated. * @param channelSupplier the supplier for the channel to read from * @param bufferFactory the factory to create data buffers with * @param bufferSize the maximum size of the data buffers - * @return a flux of data buffers read from the given channel + * @return a Flux of data buffers read from the given channel */ public static Flux readByteChannel( Callable channelSupplier, DataBufferFactory bufferFactory, int bufferSize) { @@ -107,12 +108,13 @@ public abstract class DataBufferUtils { } /** - * Obtain a {@code AsynchronousFileChannel} from the given supplier, and read it into a - * {@code Flux} of {@code DataBuffer}s. Closes the channel when the flux is terminated. + * Obtain a {@code AsynchronousFileChannel} from the given supplier, and read + * it into a {@code Flux} of {@code DataBuffer}s. Closes the channel when + * the Flux is terminated. * @param channelSupplier the supplier for the channel to read from * @param bufferFactory the factory to create data buffers with * @param bufferSize the maximum size of the data buffers - * @return a flux of data buffers read from the given channel + * @return a Flux of data buffers read from the given channel */ public static Flux readAsynchronousFileChannel( Callable channelSupplier, DataBufferFactory bufferFactory, int bufferSize) { @@ -121,17 +123,18 @@ public abstract class DataBufferUtils { } /** - * Obtain a {@code AsynchronousFileChannel} from the given supplier, and read it into a - * {@code Flux} of {@code DataBuffer}s, starting at the given position. Closes the - * channel when the flux is terminated. + * Obtain a {@code AsynchronousFileChannel} from the given supplier, and + * read it into a {@code Flux} of {@code DataBuffer}s, starting at the given + * position. Closes the channel when the Flux is terminated. * @param channelSupplier the supplier for the channel to read from * @param position the position to start reading from * @param bufferFactory the factory to create data buffers with * @param bufferSize the maximum size of the data buffers - * @return a flux of data buffers read from the given channel + * @return a Flux of data buffers read from the given channel */ - public static Flux readAsynchronousFileChannel(Callable channelSupplier, - long position, DataBufferFactory bufferFactory, int bufferSize) { + public static Flux readAsynchronousFileChannel( + Callable channelSupplier, long position, + DataBufferFactory bufferFactory, int bufferSize) { Assert.notNull(channelSupplier, "'channelSupplier' must not be null"); Assert.notNull(bufferFactory, "'dataBufferFactory' must not be null"); @@ -165,14 +168,12 @@ public abstract class DataBufferUtils { * fall back to {@link #readByteChannel(Callable, DataBufferFactory, int)}. * Closes the channel when the flux is terminated. * @param resource the resource to read from - * @param dataBufferFactory the factory to create data buffers with + * @param bufferFactory the factory to create data buffers with * @param bufferSize the maximum size of the data buffers - * @return a flux of data buffers read from the given channel + * @return a Flux of data buffers read from the given channel */ - public static Flux read( - Resource resource, DataBufferFactory dataBufferFactory, int bufferSize) { - - return read(resource, 0, dataBufferFactory, bufferSize); + public static Flux read(Resource resource, DataBufferFactory bufferFactory, int bufferSize) { + return read(resource, 0, bufferFactory, bufferSize); } /** @@ -185,26 +186,25 @@ public abstract class DataBufferUtils { * Closes the channel when the flux is terminated. * @param resource the resource to read from * @param position the position to start reading from - * @param dataBufferFactory the factory to create data buffers with + * @param bufferFactory the factory to create data buffers with * @param bufferSize the maximum size of the data buffers - * @return a flux of data buffers read from the given channel + * @return a Flux of data buffers read from the given channel */ public static Flux read( - Resource resource, long position, DataBufferFactory dataBufferFactory, int bufferSize) { + Resource resource, long position, DataBufferFactory bufferFactory, int bufferSize) { try { if (resource.isFile()) { File file = resource.getFile(); return readAsynchronousFileChannel( () -> AsynchronousFileChannel.open(file.toPath(), StandardOpenOption.READ), - position, dataBufferFactory, bufferSize); + position, bufferFactory, bufferSize); } } catch (IOException ignore) { // fallback to resource.readableChannel(), below } - - Flux result = readByteChannel(resource::readableChannel, dataBufferFactory, bufferSize); + Flux result = readByteChannel(resource::readableChannel, bufferFactory, bufferSize); return position == 0 ? result : skipUntilByteCount(result, position); } @@ -214,16 +214,19 @@ public abstract class DataBufferUtils { //--------------------------------------------------------------------- /** - * Write the given stream of {@link DataBuffer DataBuffers} to the given {@code OutputStream}. Does - * not close the output stream when the flux is terminated, and does - * not {@linkplain #release(DataBuffer) release} the data buffers in the - * source. If releasing is required, then subscribe to the returned {@code Flux} with a - * {@link #releaseConsumer()}. - *

Note that the writing process does not start until the returned {@code Flux} is subscribed to. + * Write the given stream of {@link DataBuffer DataBuffers} to the given + * {@code OutputStream}. Does not close the output stream + * when the flux is terminated, and does not + * {@linkplain #release(DataBuffer) release} the data buffers in the source. + * If releasing is required, then subscribe to the returned {@code Flux} + * with a {@link #releaseConsumer()}. + *

Note that the writing process does not start until the returned + * {@code Flux} is subscribed to. * @param source the stream of data buffers to be written * @param outputStream the output stream to write to - * @return a flux containing the same buffers as in {@code source}, that starts the writing - * process when subscribed to, and that publishes any writing errors and the completion signal + * @return a Flux containing the same buffers as in {@code source}, that + * starts the writing process when subscribed to, and that publishes any + * writing errors and the completion signal */ public static Flux write(Publisher source, OutputStream outputStream) { Assert.notNull(source, "'source' must not be null"); @@ -234,16 +237,19 @@ public abstract class DataBufferUtils { } /** - * Write the given stream of {@link DataBuffer DataBuffers} to the given {@code WritableByteChannel}. Does - * not close the channel when the flux is terminated, and does - * not {@linkplain #release(DataBuffer) release} the data buffers in the - * source. If releasing is required, then subscribe to the returned {@code Flux} with a - * {@link #releaseConsumer()}. - *

Note that the writing process does not start until the returned {@code Flux} is subscribed to. + * Write the given stream of {@link DataBuffer DataBuffers} to the given + * {@code WritableByteChannel}. Does not close the channel + * when the flux is terminated, and does not + * {@linkplain #release(DataBuffer) release} the data buffers in the source. + * If releasing is required, then subscribe to the returned {@code Flux} + * with a {@link #releaseConsumer()}. + *

Note that the writing process does not start until the returned + * {@code Flux} is subscribed to. * @param source the stream of data buffers to be written * @param channel the channel to write to - * @return a flux containing the same buffers as in {@code source}, that starts the writing - * process when subscribed to, and that publishes any writing errors and the completion signal + * @return a Flux containing the same buffers as in {@code source}, that + * starts the writing process when subscribed to, and that publishes any + * writing errors and the completion signal */ public static Flux write(Publisher source, WritableByteChannel channel) { Assert.notNull(source, "'source' must not be null"); @@ -258,16 +264,19 @@ public abstract class DataBufferUtils { } /** - * Write the given stream of {@link DataBuffer DataBuffers} to the given {@code AsynchronousFileChannel}. - * Does not close the channel when the flux is terminated, and does - * not {@linkplain #release(DataBuffer) release} the data buffers in the - * source. If releasing is required, then subscribe to the returned {@code Flux} with a - * {@link #releaseConsumer()}. - *

Note that the writing process does not start until the returned {@code Flux} is subscribed to. + * Write the given stream of {@link DataBuffer DataBuffers} to the given + * {@code AsynchronousFileChannel}. Does not close the + * channel when the flux is terminated, and does not + * {@linkplain #release(DataBuffer) release} the data buffers in the source. + * If releasing is required, then subscribe to the returned {@code Flux} + * with a {@link #releaseConsumer()}. + *

Note that the writing process does not start until the returned + * {@code Flux} is subscribed to. * @param source the stream of data buffers to be written * @param channel the channel to write to - * @return a flux containing the same buffers as in {@code source}, that starts the writing - * process when subscribed to, and that publishes any writing errors and the completion signal + * @return a Flux containing the same buffers as in {@code source}, that + * starts the writing process when subscribed to, and that publishes any + * writing errors and the completion signal * @since 5.0.10 */ public static Flux write(Publisher source, AsynchronousFileChannel channel) { @@ -275,17 +284,20 @@ public abstract class DataBufferUtils { } /** - * Write the given stream of {@link DataBuffer DataBuffers} to the given {@code AsynchronousFileChannel}. - * Does not close the channel when the flux is terminated, and does - * not {@linkplain #release(DataBuffer) release} the data buffers in the - * source. If releasing is required, then subscribe to the returned {@code Flux} with a + * Write the given stream of {@link DataBuffer DataBuffers} to the given + * {@code AsynchronousFileChannel}. Does not close the channel + * when the flux is terminated, and does not + * {@linkplain #release(DataBuffer) release} the data buffers in the source. + * If releasing is required, then subscribe to the returned {@code Flux} with a * {@link #releaseConsumer()}. - *

Note that the writing process does not start until the returned {@code Flux} is subscribed to. + *

Note that the writing process does not start until the returned + * {@code Flux} is subscribed to. * @param source the stream of data buffers to be written * @param channel the channel to write to - * @param position the file position at which the write is to begin; must be non-negative - * @return a flux containing the same buffers as in {@code source}, that starts the writing - * process when subscribed to, and that publishes any writing errors and the completion signal + * @param position file position write write is to begin; must be non-negative + * @return a flux containing the same buffers as in {@code source}, that + * starts the writing process when subscribed to, and that publishes any + * writing errors and the completion signal */ public static Flux write( Publisher source, AsynchronousFileChannel channel, long position) {