DataBufferUtils.read should not take input stream/channel as parameter

Fixed by creating `Callable`-based variants, as explained in the JIRA
issue.

Issue: SPR-16403

Author: poutsma

spring-core/src/main/java/org/springframework/core/io/buffer/DataBufferUtils.java

@@ -28,9 +28,9 @@
 import java.nio.channels.ReadableByteChannel;
 import java.nio.channels.WritableByteChannel;
 import java.nio.file.StandardOpenOption;
+import java.util.concurrent.Callable;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
-import java.util.function.BiFunction;
 import java.util.function.BinaryOperator;
 import java.util.function.Consumer;
 
@@ -69,92 +69,182 @@ public abstract class DataBufferUtils {
 	//---------------------------------------------------------------------
 
 	/**
-	 * Read the given {@code InputStream} into a {@code Flux} of
+	 * Read the given {@code InputStream} into a <strong>read-once</strong> {@code Flux} of
 	 * {@code DataBuffer}s. Closes the input stream when the flux is terminated.
+	 * <p>The resulting {@code Flux} can only be subscribed to once. See
+	 * {@link #readInputStream(Callable, DataBufferFactory, int)} for a variant that supports
+	 * multiple subscriptions.
 	 * @param inputStream the input stream to read from
 	 * @param dataBufferFactory 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
+	 * @deprecated as of Spring 5.0.3, in favor of
+	 * {@link #readInputStream(Callable, DataBufferFactory, int)}, to be removed in Spring 5.1
 	 */
+	@Deprecated
 	public static Flux<DataBuffer> read(InputStream inputStream,
 			DataBufferFactory dataBufferFactory, int bufferSize) {
+		return readInputStream(() -> inputStream, dataBufferFactory, bufferSize);
+	}
 
-		Assert.notNull(inputStream, "InputStream must not be null");
+	/**
+	 * 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 bufferSize the maximum size of the data buffers
+	 * @return a flux of data buffers read from the given channel
+	 */
+	public static Flux<DataBuffer> readInputStream(Callable<InputStream> inputStreamSupplier,
+			DataBufferFactory dataBufferFactory, int bufferSize) {
 
-		ReadableByteChannel channel = Channels.newChannel(inputStream);
-		return read(channel, dataBufferFactory, bufferSize);
+		Assert.notNull(inputStreamSupplier, "'inputStreamSupplier' must not be null");
+
+		return readByteChannel(() -> Channels.newChannel(inputStreamSupplier.call()),
+				dataBufferFactory, bufferSize);
 	}
 
 	/**
-	 * Read the given {@code ReadableByteChannel} into a {@code Flux} of
+	 * Read the given {@code ReadableByteChannel} into a <strong>read-once</strong> {@code Flux} of
 	 * {@code DataBuffer}s. Closes the channel when the flux is terminated.
+	 * <p>The resulting {@code Flux} can only be subscribed to once. See
+	 * {@link #readByteChannel(Callable, DataBufferFactory, int)} for a variant that supports
+	 * multiple subscriptions.
 	 * @param channel the channel to read from
 	 * @param dataBufferFactory 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
+	 * @deprecated as of Spring 5.0.3, in favor of
+	 * {@link #readByteChannel(Callable, DataBufferFactory, int)}, to be removed in Spring 5.1
 	 */
+	@Deprecated
 	public static Flux<DataBuffer> read(ReadableByteChannel channel,
 			DataBufferFactory dataBufferFactory, int bufferSize) {
+		return readByteChannel(() -> channel, dataBufferFactory, bufferSize);
+	}
 
-		Assert.notNull(channel, "ReadableByteChannel must not be null");
-		Assert.notNull(dataBufferFactory, "DataBufferFactory must not be null");
+	/**
+	 * 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 dataBufferFactory 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
+	 */
+	public static Flux<DataBuffer> readByteChannel(Callable<ReadableByteChannel> channelSupplier,
+			DataBufferFactory dataBufferFactory, int bufferSize) {
+
+		Assert.notNull(channelSupplier, "'channelSupplier' must not be null");
+		Assert.notNull(dataBufferFactory, "'dataBufferFactory' must not be null");
 		Assert.isTrue(bufferSize > 0, "'bufferSize' must be > 0");
 
-		return Flux.generate(() -> channel,
-				new ReadableByteChannelGenerator(dataBufferFactory, bufferSize),
-				DataBufferUtils::closeChannel);
+		return Flux.using(channelSupplier,
+				channel -> {
+					ReadableByteChannelGenerator generator =
+							new ReadableByteChannelGenerator(channel, dataBufferFactory,
+									bufferSize);
+					return Flux.generate(generator);
+				},
+				DataBufferUtils::closeChannel
+		);
 	}
 
 	/**
-	 * Read the given {@code AsynchronousFileChannel} into a {@code Flux} of
-	 * {@code DataBuffer}s. Closes the channel when the flux is terminated.
+	 * Read the given {@code AsynchronousFileChannel} into a <strong>read-once</strong> {@code Flux}
+	 * of {@code DataBuffer}s. Closes the channel when the flux is terminated.
+	 * <p>The resulting {@code Flux} can only be subscribed to once. See
+	 * {@link #readAsynchronousFileChannel(Callable, DataBufferFactory, int)} for a variant that
+	 * supports multiple subscriptions.
 	 * @param channel the channel to read from
 	 * @param dataBufferFactory 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
+	 * @deprecated as of Spring 5.0.3, in favor of
+	 * {@link #readAsynchronousFileChannel(Callable, DataBufferFactory, int)}, to be removed in
+	 * Spring 5.1
 	 */
+	@Deprecated
 	public static Flux<DataBuffer> read(AsynchronousFileChannel channel,
 			DataBufferFactory dataBufferFactory, int bufferSize) {
-		return read(channel, 0, dataBufferFactory, bufferSize);
+		return readAsynchronousFileChannel(() -> channel, dataBufferFactory, bufferSize);
 	}
 
 	/**
-	 * Read the given {@code AsynchronousFileChannel} into a {@code Flux} of
-	 * {@code DataBuffer}s, starting at the given position. Closes the channel when the flux is
+	 * Read the given {@code AsynchronousFileChannel} into a <strong>read-once</strong> {@code Flux}
+	 * of {@code DataBuffer}s, starting at the given position. Closes the channel when the flux is
 	 * terminated.
+	 * <p>The resulting {@code Flux} can only be subscribed to once. See
+	 * {@link #readAsynchronousFileChannel(Callable, long, DataBufferFactory, int)} for a variant
+	 * that supports multiple subscriptions.
 	 * @param channel the channel to read from
 	 * @param position the position to start reading from
 	 * @param dataBufferFactory 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
+	 * @deprecated as of Spring 5.0.3, in favor of
+	 * {@link #readAsynchronousFileChannel(Callable, long, DataBufferFactory, int)}, to be removed
+	 * in Spring 5.1
 	 */
+	@Deprecated
 	public static Flux<DataBuffer> read(AsynchronousFileChannel channel,
 			long position, DataBufferFactory dataBufferFactory, int bufferSize) {
+		return readAsynchronousFileChannel(() -> channel, position, dataBufferFactory, bufferSize);
+	}
 
-		Assert.notNull(channel, "'channel' must not be null");
+	/**
+	 * 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 dataBufferFactory 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
+	 */
+	public static Flux<DataBuffer> readAsynchronousFileChannel(
+			Callable<AsynchronousFileChannel> channelSupplier,
+			DataBufferFactory dataBufferFactory, int bufferSize) {
+
+		return readAsynchronousFileChannel(channelSupplier, 0, dataBufferFactory, bufferSize);
+	}
+
+	/**
+	 * 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 dataBufferFactory 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
+	 */
+	public static Flux<DataBuffer> readAsynchronousFileChannel(
+			Callable<AsynchronousFileChannel> channelSupplier,
+			long position, DataBufferFactory dataBufferFactory, int bufferSize) {
+
+		Assert.notNull(channelSupplier, "'channelSupplier' must not be null");
 		Assert.notNull(dataBufferFactory, "'dataBufferFactory' must not be null");
 		Assert.isTrue(position >= 0, "'position' must be >= 0");
 		Assert.isTrue(bufferSize > 0, "'bufferSize' must be > 0");
 
 		DataBuffer dataBuffer = dataBufferFactory.allocateBuffer(bufferSize);
 		ByteBuffer byteBuffer = dataBuffer.asByteBuffer(0, bufferSize);
 
-		return Flux.create(sink -> {
-			sink.onDispose(() -> closeChannel(channel));
-			CompletionHandler<Integer, DataBuffer> completionHandler =
-					new AsynchronousFileChannelReadCompletionHandler(channel, sink, position,
-							dataBufferFactory, bufferSize);
-			channel.read(byteBuffer, position, dataBuffer, completionHandler);
-		});
+		return Flux.using(channelSupplier,
+				channel -> Flux.create(sink -> {
+							CompletionHandler<Integer, DataBuffer> completionHandler =
+									new AsynchronousFileChannelReadCompletionHandler(channel,
+											sink, position, dataBufferFactory, bufferSize);
+							channel.read(byteBuffer, position, dataBuffer, completionHandler);
+						}),
+				DataBufferUtils::closeChannel);
 	}
 
 	/**
 	 * Read the given {@code Resource} into a {@code Flux} of {@code DataBuffer}s.
 	 * <p>If the resource is a file, it is read into an
 	 * {@code AsynchronousFileChannel} and turned to {@code Flux} via
-	 * {@link #read(AsynchronousFileChannel, DataBufferFactory, int)} or else
-	 * fall back on {@link #read(InputStream, DataBufferFactory, int)} closes
-	 * the channel when the flux is terminated.
+	 * {@link #readAsynchronousFileChannel(Callable, DataBufferFactory, int)} or else
+	 * 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 bufferSize the maximum size of the data buffers
@@ -171,9 +261,9 @@ public static Flux<DataBuffer> read(Resource resource,
 	 * starting at the given position.
 	 * <p>If the resource is a file, it is read into an
 	 * {@code AsynchronousFileChannel} and turned to {@code Flux} via
-	 * {@link #read(AsynchronousFileChannel, DataBufferFactory, int)} or else
-	 * fall back on {@link #read(InputStream, DataBufferFactory, int)}. Closes
-	 * the channel when the flux is terminated.
+	 * {@link #readAsynchronousFileChannel(Callable, DataBufferFactory, int)} or else
+	 * fall back on {@link #readByteChannel(Callable, DataBufferFactory, int)}.
+	 * 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
@@ -186,26 +276,23 @@ public static Flux<DataBuffer> read(Resource resource, long position,
 		try {
 			if (resource.isFile()) {
 				File file = resource.getFile();
-				AsynchronousFileChannel channel =
-						AsynchronousFileChannel.open(file.toPath(), StandardOpenOption.READ);
-				return DataBufferUtils.read(channel, position, dataBufferFactory, bufferSize);
+
+				return readAsynchronousFileChannel(
+						() -> AsynchronousFileChannel.open(file.toPath(), StandardOpenOption.READ),
+						position, dataBufferFactory, bufferSize);
 			}
 		}
 		catch (IOException ignore) {
 			// fallback to resource.readableChannel(), below
 		}
 
-		try {
-			ReadableByteChannel channel = resource.readableChannel();
-			Flux<DataBuffer> in = DataBufferUtils.read(channel, dataBufferFactory, bufferSize);
-			return DataBufferUtils.skipUntilByteCount(in, position);
-		}
-		catch (IOException ex) {
-			return Flux.error(ex);
-		}
+		Flux<DataBuffer> result = readByteChannel(resource::readableChannel, dataBufferFactory, bufferSize);
+		return position == 0 ? result : skipUntilByteCount(result, position);
 	}
 
 
+
+
 	//---------------------------------------------------------------------
 	// Writing
 	//---------------------------------------------------------------------
@@ -304,7 +391,7 @@ public static Flux<DataBuffer> write(Publisher<DataBuffer> source, AsynchronousF
 
 	private static void closeChannel(@Nullable Channel channel) {
 		try {
-			if (channel != null) {
+			if (channel != null && channel.isOpen()) {
 				channel.close();
 			}
 		}
@@ -442,44 +529,49 @@ public static Mono<DataBuffer> compose(Publisher<DataBuffer> publisher) {
 
 
 	private static class ReadableByteChannelGenerator
-			implements BiFunction<ReadableByteChannel, SynchronousSink<DataBuffer>, ReadableByteChannel> {
+			implements Consumer<SynchronousSink<DataBuffer>> {
+
+		private final ReadableByteChannel channel;
 
 		private final DataBufferFactory dataBufferFactory;
 
 		private final int bufferSize;
 
-		public ReadableByteChannelGenerator(DataBufferFactory dataBufferFactory, int bufferSize) {
+
+		public ReadableByteChannelGenerator(ReadableByteChannel channel,
+				DataBufferFactory dataBufferFactory, int bufferSize) {
+
+			this.channel = channel;
 			this.dataBufferFactory = dataBufferFactory;
 			this.bufferSize = bufferSize;
 		}
 
 		@Override
-		public ReadableByteChannel apply(ReadableByteChannel channel,
-				SynchronousSink<DataBuffer> sub) {
+		public void accept(SynchronousSink<DataBuffer> sink) {
 			boolean release = true;
 			DataBuffer dataBuffer = this.dataBufferFactory.allocateBuffer(this.bufferSize);
 			try {
 				int read;
 				ByteBuffer byteBuffer = dataBuffer.asByteBuffer(0, dataBuffer.capacity());
-				if ((read = channel.read(byteBuffer)) >= 0) {
+				if ((read = this.channel.read(byteBuffer)) >= 0) {
 					dataBuffer.writePosition(read);
 					release = false;
-					sub.next(dataBuffer);
+					sink.next(dataBuffer);
 				}
 				else {
-					sub.complete();
+					sink.complete();
 				}
 			}
 			catch (IOException ex) {
-				sub.error(ex);
+				sink.error(ex);
 			}
 			finally {
 				if (release) {
 					release(dataBuffer);
 				}
 			}
-			return channel;
 		}
+
 	}
 
 
@@ -494,7 +586,10 @@ private static class AsynchronousFileChannelReadCompletionHandler
 
 		private final int bufferSize;
 
-		private AtomicLong position;
+		private final AtomicLong position;
+
+		private final AtomicBoolean disposed = new AtomicBoolean();
+
 
 		private AsynchronousFileChannelReadCompletionHandler(
 				AsynchronousFileChannel channel, FluxSink<DataBuffer> sink,
@@ -513,7 +608,7 @@ public void completed(Integer read, DataBuffer dataBuffer) {
 				dataBuffer.writePosition(read);
 				this.sink.next(dataBuffer);
 
-				if (!this.sink.isCancelled()) {
+				if (!this.disposed.get()) {
 					DataBuffer newDataBuffer =
 							this.dataBufferFactory.allocateBuffer(this.bufferSize);
 					ByteBuffer newByteBuffer = newDataBuffer.asByteBuffer(0, this.bufferSize);
@@ -522,17 +617,16 @@ public void completed(Integer read, DataBuffer dataBuffer) {
 			}
 			else {
 				release(dataBuffer);
-				closeChannel(this.channel);
 				this.sink.complete();
 			}
 		}
 
 		@Override
 		public void failed(Throwable exc, DataBuffer dataBuffer) {
 			release(dataBuffer);
-			closeChannel(this.channel);
 			this.sink.error(exc);
 		}
+
 	}