Package org.springframework.core.io.buffer

# Class DataBufferUtils

java.lang.Object
org.springframework.core.io.buffer.DataBufferUtils

public abstract class DataBufferUtils extends Object
Utility class for working with `DataBuffers`.

Since:
  5.0
Author:
  Arjen Poutsma, Brian Clozel

## Nested Class Summary

Nested Classes
Modifier and Type
Class
Description
`static interface`
`DataBufferUtils.Matcher`
Contract to find delimiter(s) against one or more data buffers that can be passed one at a time to the `DataBufferUtils.Matcher.match(DataBuffer)` method.

## Constructor Summary

Constructors
Constructor
Description
`DataBufferUtils()`

## Method Summary

Modifier and Type
Method
Description
`static reactor.core.publisher.Mono<DataBuffer>`
`join(Publisher<? extends DataBuffer> dataBuffers)`
Return a new `DataBuffer` composed of joining together the given `dataBuffers` elements.
`static reactor.core.publisher.Mono<DataBuffer>`
`join(Publisher<? extends DataBuffer> buffers, int maxByteCount)`
Variant of `join(Publisher)` that behaves the same way up until the specified max number of bytes to buffer.
`static DataBufferUtils.Matcher`
`matcher(byte[] delimiter)`
Return a `DataBufferUtils.Matcher` for the given delimiter.
`static DataBufferUtils.Matcher`
`matcher(byte[]... delimiters)`
Return a `DataBufferUtils.Matcher` for the given delimiters.
`static Publisher<DataBuffer>`
`outputStreamPublisher(Consumer<OutputStream> consumer, DataBufferFactory bufferFactory, Executor executor)`
Create a new `Publisher<DataBuffer>` based on bytes written to a `OutputStream`.
`static Publisher<DataBuffer>`
`outputStreamPublisher(Consumer<OutputStream> consumer, DataBufferFactory bufferFactory, Executor executor, int chunkSize)`
Variant of `outputStreamPublisher(Consumer, DataBufferFactory, Executor)` providing control over the chunk sizes to be produced by the publisher.
`static reactor.core.publisher.Flux<DataBuffer>`
`read(Path path, DataBufferFactory bufferFactory, int bufferSize, OpenOption... options)`
Read bytes from the given file `Path` into a `Flux` of `DataBuffer`s.
`static reactor.core.publisher.Flux<DataBuffer>`
`read(Resource resource, long position, DataBufferFactory bufferFactory, int bufferSize)`
Read the given `Resource` into a `Flux` of `DataBuffer`s starting at the given position.
`static reactor.core.publisher.Flux<DataBuffer>`
`read(Resource resource, DataBufferFactory bufferFactory, int bufferSize)`
Read the given `Resource` into a `Flux` of `DataBuffer`s.
`static reactor.core.publisher.Flux<DataBuffer>`
`readAsynchronousFileChannel(Callable<AsynchronousFileChannel> channelSupplier, long position, DataBufferFactory bufferFactory, int bufferSize)`
Obtain an `AsynchronousFileChannel` from the given supplier, and read it into a `Flux` of `DataBuffer`s, starting at the given position.
`static reactor.core.publisher.Flux<DataBuffer>`
`readAsynchronousFileChannel(Callable<AsynchronousFileChannel> channelSupplier, DataBufferFactory bufferFactory, int bufferSize)`
Obtain a `AsynchronousFileChannel` from the given supplier, and read it into a `Flux` of `DataBuffer`s.
`static reactor.core.publisher.Flux<DataBuffer>`
`readByteChannel(Callable<ReadableByteChannel> channelSupplier, DataBufferFactory bufferFactory, int bufferSize)`
Obtain a `ReadableByteChannel` from the given supplier, and read it into a `Flux` of `DataBuffer`s.
`static reactor.core.publisher.Flux<DataBuffer>`
`readInputStream(Callable<InputStream> inputStreamSupplier, DataBufferFactory bufferFactory, int bufferSize)`
Obtain an `InputStream` from the given supplier, and read it into a `Flux` of `DataBuffer`s.
`static boolean`
`release(DataBuffer dataBuffer)`
Release the given data buffer.
`static Consumer<DataBuffer>`
`releaseConsumer()`
Return a consumer that calls `release(DataBuffer)` on all passed data buffers.
`static <T extends DataBuffer>
T`
`retain(T dataBuffer)`
Retain the given data buffer, if it is a `PooledDataBuffer`.
`static <T extends DataBuffer>
reactor.core.publisher.Flux<T>`
`skipUntilByteCount(Publisher<T> publisher, long maxByteCount)`
Skip buffers from the given `Publisher` until the total byte count reaches the given maximum byte count, or until the publisher is complete.
`static <T extends DataBuffer>
InputStream`
`subscriberInputStream(Publisher<T> publisher, int demand)`
Subscribe to given `Publisher` of `DataBuffer`s, and return an `InputStream` to consume the byte content with.
`static <T extends DataBuffer>
reactor.core.publisher.Flux<T>`
`takeUntilByteCount(Publisher<T> publisher, long maxByteCount)`
Relay buffers from the given `Publisher` until the total byte count reaches the given maximum byte count, or until the publisher is complete.
`static <T extends DataBuffer>
T`
`touch(T dataBuffer, Object hint)`
Associate the given hint with the data buffer if it is a pooled buffer and supports leak tracking.
`static reactor.core.publisher.Flux<DataBuffer>`
`write(Publisher<? extends DataBuffer> source, AsynchronousFileChannel channel, long position)`
Write the given stream of `DataBuffers` to the given `AsynchronousFileChannel`.
`static reactor.core.publisher.Flux<DataBuffer>`
`write(Publisher<DataBuffer> source, OutputStream outputStream)`
Write the given stream of `DataBuffers` to the given `OutputStream`.
`static reactor.core.publisher.Flux<DataBuffer>`
`write(Publisher<DataBuffer> source, AsynchronousFileChannel channel)`
Write the given stream of `DataBuffers` to the given `AsynchronousFileChannel`.
`static reactor.core.publisher.Flux<DataBuffer>`
`write(Publisher<DataBuffer> source, WritableByteChannel channel)`
Write the given stream of `DataBuffers` to the given `WritableByteChannel`.
`static reactor.core.publisher.Mono<Void>`
`write(Publisher<DataBuffer> source, Path destination, OpenOption... options)`
Write the given stream of `DataBuffers` to the given file `Path`.

### Methods inherited from class java.lang.Object

`clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait`

## Constructor Details

### DataBufferUtils

public DataBufferUtils()

## Method Details

### readInputStream

public static reactor.core.publisher.Flux<DataBuffer> readInputStream(Callable<InputStream> inputStreamSupplier, DataBufferFactory bufferFactory, int bufferSize)
Obtain an `InputStream` from the given supplier, and read it into a `Flux` of `DataBuffer`s. Closes the input stream when the Flux is terminated.

Parameters:
  `inputStreamSupplier` - the supplier for the input stream to read from
  `bufferFactory` - the factory to create data buffers with
  `bufferSize` - the maximum size of the data buffers
Returns:
  a Flux of data buffers read from the given channel

### readByteChannel

public static reactor.core.publisher.Flux<DataBuffer> readByteChannel(Callable<ReadableByteChannel> channelSupplier, DataBufferFactory bufferFactory, int bufferSize)
Obtain a `ReadableByteChannel` from the given supplier, and read it into a `Flux` of `DataBuffer`s. Closes the channel when the Flux is terminated.

Parameters:
  `channelSupplier` - the supplier for the channel to read from
  `bufferFactory` - the factory to create data buffers with
  `bufferSize` - the maximum size of the data buffers
Returns:
  a Flux of data buffers read from the given channel

### readAsynchronousFileChannel

public static reactor.core.publisher.Flux<DataBuffer> readAsynchronousFileChannel(Callable<AsynchronousFileChannel> channelSupplier, DataBufferFactory bufferFactory, int bufferSize)
Obtain a `AsynchronousFileChannel` from the given supplier, and read it into a `Flux` of `DataBuffer`s. Closes the channel when the Flux is terminated.

Parameters:
  `channelSupplier` - the supplier for the channel to read from
  `bufferFactory` - the factory to create data buffers with
  `bufferSize` - the maximum size of the data buffers
Returns:
  a Flux of data buffers read from the given channel

### readAsynchronousFileChannel

public static reactor.core.publisher.Flux<DataBuffer> readAsynchronousFileChannel(Callable<AsynchronousFileChannel> channelSupplier, long position, DataBufferFactory bufferFactory, int bufferSize)
Obtain an `AsynchronousFileChannel` from the given supplier, and read it into a `Flux` of `DataBuffer`s, starting at the given position. Closes the channel when the Flux is terminated.

Parameters:
  `channelSupplier` - the supplier for the channel to read from
  `position` - the position to start reading from
  `bufferFactory` - the factory to create data buffers with
  `bufferSize` - the maximum size of the data buffers
Returns:
  a Flux of data buffers read from the given channel

### read

public static reactor.core.publisher.Flux<DataBuffer> read(Path path, DataBufferFactory bufferFactory, int bufferSize, OpenOption... options)
Read bytes from the given file `Path` into a `Flux` of `DataBuffer`s. The method ensures that the file is closed when the flux is terminated.

Parameters:
  `path` - the path to read bytes from
  `bufferFactory` - the factory to create data buffers with
  `bufferSize` - the maximum size of the data buffers
Returns:
  a Flux of data buffers read from the given channel
Since:
  5.2

### read

public static reactor.core.publisher.Flux<DataBuffer> read(Resource resource, DataBufferFactory bufferFactory, int bufferSize)
Read the given `Resource` into a `Flux` of `DataBuffer`s.

If the resource is a file, it is read into an `AsynchronousFileChannel` and turned to `Flux` via `readAsynchronousFileChannel(Callable, DataBufferFactory, int)` or else fall back to `readByteChannel(Callable, DataBufferFactory, int)`. Closes the channel when the flux is terminated.

Parameters:
  `resource` - the resource to read from
  `bufferFactory` - the factory to create data buffers with
  `bufferSize` - the maximum size of the data buffers
Returns:
  a Flux of data buffers read from the given channel

### read

public static reactor.core.publisher.Flux<DataBuffer> read(Resource resource, long position, DataBufferFactory bufferFactory, int bufferSize)
Read the given `Resource` into a `Flux` of `DataBuffer`s starting at the given position.

If the resource is a file, it is read into an `AsynchronousFileChannel` and turned to `Flux` via `readAsynchronousFileChannel(Callable, DataBufferFactory, int)` or else fall back on `readByteChannel(Callable, DataBufferFactory, int)`. Closes the channel when the flux is terminated.

Parameters:
  `resource` - the resource to read from
  `position` - the position to start reading from
  `bufferFactory` - the factory to create data buffers with
  `bufferSize` - the maximum size of the data buffers
Returns:
  a Flux of data buffers read from the given channel

### write

public static reactor.core.publisher.Flux<DataBuffer> write(Publisher<DataBuffer> source, OutputStream outputStream)
Write the given stream of `DataBuffers` to the given `OutputStream`. Does not close the output stream when the flux is terminated, and does not release the data buffers in the source. If releasing is required, then subscribe to the returned `Flux` with a `releaseConsumer()`.

Note that the writing process does not start until the returned `Flux` is subscribed to.

Parameters:
  `source` - the stream of data buffers to be written
  `outputStream` - the output stream to write to
Returns:
  a Flux containing the same buffers as in `source`, that starts the writing process when subscribed to, and that publishes any writing errors and the completion signal

### write

public static reactor.core.publisher.Flux<DataBuffer> write(Publisher<DataBuffer> source, WritableByteChannel channel)
Write the given stream of `DataBuffers` to the given `WritableByteChannel`. Does not close the channel when the flux is terminated, and does not release the data buffers in the source. If releasing is required, then subscribe to the returned `Flux` with a `releaseConsumer()`.

Note that the writing process does not start until the returned `Flux` is subscribed to.

Parameters:
  `source` - the stream of data buffers to be written
  `channel` - the channel to write to
Returns:
  a Flux containing the same buffers as in `source`, that starts the writing process when subscribed to, and that publishes any writing errors and the completion signal

### write

public static reactor.core.publisher.Flux<DataBuffer> write(Publisher<DataBuffer> source, AsynchronousFileChannel channel)
Write the given stream of `DataBuffers` to the given `AsynchronousFileChannel`. Does not close the channel when the flux is terminated, and does not release the data buffers in the source. If releasing is required, then subscribe to the returned `Flux` with a `releaseConsumer()`.

Note that the writing process does not start until the returned `Flux` is subscribed to.

Parameters:
  `source` - the stream of data buffers to be written
  `channel` - the channel to write to
Returns:
  a Flux containing the same buffers as in `source`, that starts the writing process when subscribed to, and that publishes any writing errors and the completion signal
Since:
  5.0.10

### write

public static reactor.core.publisher.Flux<DataBuffer> write(Publisher<? extends DataBuffer> source, AsynchronousFileChannel channel, long position)
Write the given stream of `DataBuffers` to the given `AsynchronousFileChannel`. Does not close the channel when the flux is terminated, and does not release the data buffers in the source. If releasing is required, then subscribe to the returned `Flux` with a `releaseConsumer()`.

Note that the writing process does not start until the returned `Flux` is subscribed to.

Parameters:
  `source` - the stream of data buffers to be written
  `channel` - the channel to write to
  `position` - the file position where writing is to begin; must be non-negative
Returns:
  a flux containing the same buffers as in `source`, that starts the writing process when subscribed to, and that publishes any writing errors and the completion signal

### write

public static reactor.core.publisher.Mono<Void> write(Publisher<DataBuffer> source, Path destination, OpenOption... options)
Write the given stream of `DataBuffers` to the given file `Path`. The optional `options` parameter specifies how the file is created or opened (defaults to `CREATE`, `TRUNCATE_EXISTING`, and `WRITE`).

Parameters:
  `source` - the stream of data buffers to be written
  `destination` - the path to the file
  `options` - the options specifying how the file is opened
Returns:
  a `Mono` that indicates completion or error
Since:
  5.2

### outputStreamPublisher

public static Publisher<DataBuffer> outputStreamPublisher(Consumer<OutputStream> consumer, DataBufferFactory bufferFactory, Executor executor)
Create a new `Publisher<DataBuffer>` based on bytes written to a `OutputStream`.

- The parameter `outputStreamConsumer` is invoked once per subscription of the returned `Publisher`, when the first item is requested.
- `OutputStream.write()` invocations made by `outputStreamConsumer` are buffered until they exceed the default chunk size of 1024, or when the stream is flushed and then result in a published item if there is demand.
- If there is no demand, `OutputStream.write()` will block until there is.
- If the subscription is cancelled, `OutputStream.write()` will throw a `IOException`.
- The subscription is completed when `outputStreamHandler` completes.
- Any exceptions thrown from `outputStreamHandler` will be dispatched to the Subscriber.

Parameters:
  `consumer` - invoked when the first buffer is requested
  `executor` - used to invoke the `outputStreamHandler`
Returns:
  a `Publisher<DataBuffer>` based on bytes written by `outputStreamHandler`
Since:
  6.1

### outputStreamPublisher

public static Publisher<DataBuffer> outputStreamPublisher(Consumer<OutputStream> consumer, DataBufferFactory bufferFactory, Executor executor, int chunkSize)
Variant of `outputStreamPublisher(Consumer, DataBufferFactory, Executor)` providing control over the chunk sizes to be produced by the publisher.

Since:
  6.1

### subscriberInputStream

public static <T extends DataBuffer> InputStream subscriberInputStream(Publisher<T> publisher, int demand)
Subscribe to given `Publisher` of `DataBuffer`s, and return an `InputStream` to consume the byte content with.

Byte buffers are stored in a queue. The `demand` constructor value determines the number of buffers requested initially. When storage falls below a `(demand - (demand >> 2))` limit, a request is made to refill the queue.

The `InputStream` terminates after an onError or onComplete signal, and stored buffers are read. If the `InputStream` is closed, the `Flow.Subscription` is cancelled, and stored buffers released.

Parameters:
  `publisher` - the source of `DataBuffer`s
  `demand` - the number of buffers to request initially, and buffer internally on an ongoing basis.
Returns:
  an `InputStream` backed by the `Publisher`

### takeUntilByteCount

public static <T extends DataBuffer> reactor.core.publisher.Flux<T> takeUntilByteCount(Publisher<T> publisher, long maxByteCount)
Relay buffers from the given `Publisher` until the total byte count reaches the given maximum byte count, or until the publisher is complete.

Parameters:
  `publisher` - the publisher to filter
  `maxByteCount` - the maximum byte count
Returns:
  a flux whose maximum byte count is `maxByteCount`

### skipUntilByteCount

public static <T extends DataBuffer> reactor.core.publisher.Flux<T> skipUntilByteCount(Publisher<T> publisher, long maxByteCount)
Skip buffers from the given `Publisher` until the total byte count reaches the given maximum byte count, or until the publisher is complete.

Parameters:
  `publisher` - the publisher to filter
  `maxByteCount` - the maximum byte count
Returns:
  a flux with the remaining part of the given publisher

### retain

public static <T extends DataBuffer> T retain(T dataBuffer)
Retain the given data buffer, if it is a `PooledDataBuffer`.

Parameters:
  `dataBuffer` - the data buffer to retain
Returns:
  the retained buffer

### touch

public static <T extends DataBuffer> T touch(T dataBuffer, Object hint)
Associate the given hint with the data buffer if it is a pooled buffer and supports leak tracking.

Parameters:
  `dataBuffer` - the data buffer to attach the hint to
  `hint` - the hint to attach
Returns:
  the input buffer
Since:
  5.3.2

### release

public static boolean release(@Nullable DataBuffer dataBuffer)
Release the given data buffer. If it is a `PooledDataBuffer` and has been allocated, this method will call `PooledDataBuffer.release()`. If it is a `CloseableDataBuffer`, this method will call `CloseableDataBuffer.close()`.

Parameters:
  `dataBuffer` - the data buffer to release
Returns:
  `true` if the buffer was released; `false` otherwise.

### releaseConsumer

public static Consumer<DataBuffer> releaseConsumer()
Return a consumer that calls `release(DataBuffer)` on all passed data buffers.

### join

public static reactor.core.publisher.Mono<DataBuffer> join(Publisher<? extends DataBuffer> dataBuffers)
Return a new `DataBuffer` composed of joining together the given `dataBuffers` elements. Depending on the `DataBuffer` type, the returned buffer may be a single buffer containing all data of the provided buffers, or it may be a zero-copy, composite with references to the given buffers.

If `dataBuffers` produces an error or if there is a cancel signal, then all accumulated buffers will be released.

Note that the given data buffers do not have to be released. They will be released as part of the returned composite.

Parameters:
  `dataBuffers` - the data buffers that are to be composed
Returns:
  a buffer that is composed of the `dataBuffers` argument
Since:
  5.0.3

### join

public static reactor.core.publisher.Mono<DataBuffer> join(Publisher<? extends DataBuffer> buffers, int maxByteCount)
Variant of `join(Publisher)` that behaves the same way up until the specified max number of bytes to buffer. Once the limit is exceeded, `DataBufferLimitException` is raised.

Parameters:
  `buffers` - the data buffers that are to be composed
  `maxByteCount` - the max number of bytes to buffer, or -1 for unlimited
Returns:
  a buffer with the aggregated content, possibly an empty Mono if the max number of bytes to buffer is exceeded.
Throws:
  `DataBufferLimitException` - if maxByteCount is exceeded
Since:
  5.1.11

### matcher

public static DataBufferUtils.Matcher matcher(byte[] delimiter)
Return a `DataBufferUtils.Matcher` for the given delimiter. The matcher can be used to find the delimiters in a stream of data buffers.

Parameters:
  `delimiter` - the delimiter bytes to find
Returns:
  the matcher
Since:
  5.2

### matcher

public static DataBufferUtils.Matcher matcher(byte[]... delimiters)
Return a `DataBufferUtils.Matcher` for the given delimiters. The matcher can be used to find the delimiters in a stream of data buffers.

Parameters:
  `delimiters` - the delimiters bytes to find
Returns:
  the matcher
Since:
  5.2