Spring/spring-framework
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Polishing

Browse Source
This commit is contained in:
Arjen Poutsma
2016-05-03 16:33:56 +02:00
parent b2d9e7c28d
commit b71f143dfd
1 changed files with 18 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
spring-web-reactive/src/main/java/org/springframework/core/io/buffer/support/DataBufferUtils.java
View File

@@ -16,7 +16,6 @@
package org.springframework.core.io.buffer.support; package org.springframework.core.io.buffer.support;
import java.io.Closeable;
import java.io.IOException; import java.io.IOException;
import java.io.InputStream; import java.io.InputStream;
import java.nio.ByteBuffer; import java.nio.ByteBuffer;
@@ -44,33 +43,16 @@ import org.springframework.util.Assert;
*/ */
public abstract class DataBufferUtils { public abstract class DataBufferUtils {
private static final Consumer<? extends Closeable> CLOSE_CONSUMER = closeable -> { private static final Consumer<ReadableByteChannel> CLOSE_CONSUMER = channel -> {
try { try {
if (closeable != null) { if (channel != null) {
closeable.close(); channel.close();
} }
} }
catch (IOException ignored) { catch (IOException ignored) {
} }
}; };
/**
* Reads the given {@code ReadableByteChannel} into a {@code Flux} of
* {@code DataBuffer}s. Closes the channel when the flux is terminated.
* @param channel the channel to read from
* @param allocator the allocator 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> read(ReadableByteChannel channel,
DataBufferAllocator allocator, int bufferSize) {
Assert.notNull(channel, "'channel' must not be null");
Assert.notNull(allocator, "'allocator' must not be null");
return Flux.create(new ReadableByteChannelConsumer(allocator, bufferSize),
subscriber -> channel, closeConsumer());
}
/** /**
* Reads the given {@code InputStream} into a {@code Flux} of * Reads the given {@code InputStream} into a {@code Flux} of
* {@code DataBuffer}s. Closes the stream when the flux inputStream terminated. * {@code DataBuffer}s. Closes the stream when the flux inputStream terminated.
@@ -88,9 +70,21 @@ public abstract class DataBufferUtils {
return read(channel, allocator, bufferSize); return read(channel, allocator, bufferSize);
} }
@SuppressWarnings("unchecked") /**
private static <T extends Closeable> Consumer<T> closeConsumer() { * Reads the given {@code ReadableByteChannel} into a {@code Flux} of
return (Consumer<T>) CLOSE_CONSUMER; * {@code DataBuffer}s. Closes the channel when the flux is terminated.
* @param channel the channel to read from
* @param allocator the allocator 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> read(ReadableByteChannel channel,
DataBufferAllocator allocator, int bufferSize) {
Assert.notNull(channel, "'channel' must not be null");
Assert.notNull(allocator, "'allocator' must not be null");
return Flux.create(new ReadableByteChannelConsumer(allocator, bufferSize),
subscriber -> channel, CLOSE_CONSUMER);
} }
/** /**