diff --git a/spring-core/src/main/java/org/springframework/core/io/buffer/DataBuffer.java b/spring-core/src/main/java/org/springframework/core/io/buffer/DataBuffer.java
index b76408efd0..61da1bfb4f 100644
--- a/spring-core/src/main/java/org/springframework/core/io/buffer/DataBuffer.java
+++ b/spring-core/src/main/java/org/springframework/core/io/buffer/DataBuffer.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -24,8 +24,29 @@ import java.util.function.IntPredicate;
/**
* Basic abstraction over byte buffers.
*
+ * {@code DataBuffer}s has a separate {@linkplain #readPosition() read} and
+ * {@linkplain #writePosition() write} position, as opposed to {@code ByteBuffer}'s single
+ * {@linkplain ByteBuffer#position() position}. As such, the {@code DataBuffer} does not require
+ * a {@linkplain ByteBuffer#flip() flip} to read after writing. In general, the following invariant
+ * holds for the read and write positions, and the capacity:
+ *
+ *
+ * 0 <=
+ * readPosition <=
+ * writePosition <=
+ * capacity
+ *
+ *
+ * The {@linkplain #capacity() capacity} of a {@code DataBuffer} is expanded on demand,
+ * similar to {@code StringBuilder}.
+ *
+ *
The main purpose of the {@code DataBuffer} abstraction is provide a convenient wrapper around
+ * {@link ByteBuffer}, similar to Netty's {@link io.netty.buffer.ByteBuf}, that can also be used on
+ * non-Netty platforms (i.e. Servlet).
+ *
* @author Arjen Poutsma
* @since 5.0
+ * @see DataBufferFactory
*/
public interface DataBuffer {
@@ -61,6 +82,63 @@ public interface DataBuffer {
*/
int readableByteCount();
+ /**
+ * Return the number of bytes that can be written to this data buffer.
+ * @return the writable byte count
+ * @since 5.0.1
+ */
+ int writableByteCount();
+
+ /**
+ * Return the number of bytes that this buffer can contain.
+ * @return the capacity
+ * @since 5.0.1
+ */
+ int capacity();
+
+ /**
+ * Sets the number of bytes that this buffer can contain. If the new capacity is lower than
+ * the current capacity, the contents of this buffer will be truncated. If the new capacity
+ * is higher than the current capacity, it will be expanded.
+ * @param capacity the new capacity
+ * @return this buffer
+ */
+ DataBuffer capacity(int capacity);
+
+ /**
+ * Return the position from which this buffer will read.
+ * @return the read position
+ * @since 5.0.1
+ */
+ int readPosition();
+
+ /**
+ * Set the position from which this buffer will read.
+ * @param readPosition the new read position
+ * @return this buffer
+ * @throws IndexOutOfBoundsException if {@code readPosition} is smaller than 0 or greater than
+ * {@link #writePosition()}
+ * @since 5.0.1
+ */
+ DataBuffer readPosition(int readPosition);
+
+ /**
+ * Return the position to which this buffer will write.
+ * @return the write position
+ * @since 5.0.1
+ */
+ int writePosition();
+
+ /**
+ * Set the position to which this buffer will write.
+ * @param writePosition the new write position
+ * @return this buffer
+ * @throws IndexOutOfBoundsException if {@code writePosition} is smaller than
+ * {@link #readPosition()} or greater than {@link #capacity()}
+ * @since 5.0.1
+ */
+ DataBuffer writePosition(int writePosition);
+
/**
* Read a single byte from the current reading position of this data buffer.
* @return the byte at this buffer's current reading position
@@ -146,6 +224,18 @@ public interface DataBuffer {
*/
ByteBuffer asByteBuffer();
+ /**
+ * Expose a subsequence of this buffer's bytes as a {@link ByteBuffer}. Data between this
+ * {@code DataBuffer} and the returned {@code ByteBuffer} is shared; though
+ * changes in the returned buffer's {@linkplain ByteBuffer#position() position}
+ * will not be reflected in the reading nor writing position of this data buffer.
+ * @param index the index at which to start the byte buffer
+ * @param length the length of the returned byte buffer
+ * @return this data buffer as a byte buffer
+ * @since 5.0.1
+ */
+ ByteBuffer asByteBuffer(int index, int length);
+
/**
* Expose this buffer's data as an {@link InputStream}. Both data and position are
* shared between the returned stream and this data buffer.
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 7ad441168a..655ef4749c 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
@@ -71,7 +71,6 @@ public abstract class DataBufferUtils {
DataBufferFactory dataBufferFactory, int bufferSize) {
Assert.notNull(inputStream, "InputStream must not be null");
- Assert.notNull(dataBufferFactory, "DataBufferFactory must not be null");
ReadableByteChannel channel = Channels.newChannel(inputStream);
return read(channel, dataBufferFactory, bufferSize);
@@ -90,6 +89,7 @@ public abstract class DataBufferUtils {
Assert.notNull(channel, "ReadableByteChannel 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),
@@ -125,15 +125,17 @@ public abstract class DataBufferUtils {
Assert.notNull(channel, "'channel' 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");
- ByteBuffer byteBuffer = ByteBuffer.allocate(bufferSize);
+ DataBuffer dataBuffer = dataBufferFactory.allocateBuffer(bufferSize);
+ ByteBuffer byteBuffer = dataBuffer.asByteBuffer(0, bufferSize);
return Flux.create(sink -> {
sink.onDispose(() -> closeChannel(channel));
- AsynchronousFileChannelReadCompletionHandler completionHandler =
- new AsynchronousFileChannelReadCompletionHandler(sink, position,
- dataBufferFactory, byteBuffer);
- channel.read(byteBuffer, position, channel, completionHandler);
+ CompletionHandler completionHandler =
+ new AsynchronousFileChannelReadCompletionHandler(channel, sink, position,
+ dataBufferFactory, bufferSize);
+ channel.read(byteBuffer, position, dataBuffer, completionHandler);
});
}
@@ -414,32 +416,25 @@ public abstract class DataBufferUtils {
private final DataBufferFactory dataBufferFactory;
- private final ByteBuffer byteBuffer;
+ private final int bufferSize;
- public ReadableByteChannelGenerator(DataBufferFactory dataBufferFactory, int chunkSize) {
+ public ReadableByteChannelGenerator(DataBufferFactory dataBufferFactory, int bufferSize) {
this.dataBufferFactory = dataBufferFactory;
- this.byteBuffer = ByteBuffer.allocate(chunkSize);
+ this.bufferSize = bufferSize;
}
@Override
- public ReadableByteChannel apply(ReadableByteChannel channel, SynchronousSink sub) {
+ public ReadableByteChannel apply(ReadableByteChannel channel,
+ SynchronousSink sub) {
+ boolean release = true;
+ DataBuffer dataBuffer = this.dataBufferFactory.allocateBuffer(this.bufferSize);
try {
int read;
- if ((read = channel.read(this.byteBuffer)) >= 0) {
- this.byteBuffer.flip();
- boolean release = true;
- DataBuffer dataBuffer = this.dataBufferFactory.allocateBuffer(read);
- try {
- dataBuffer.write(this.byteBuffer);
- release = false;
- sub.next(dataBuffer);
- }
- finally {
- if (release) {
- release(dataBuffer);
- }
- }
- this.byteBuffer.clear();
+ ByteBuffer byteBuffer = dataBuffer.asByteBuffer(0, dataBuffer.capacity());
+ if ((read = channel.read(byteBuffer)) >= 0) {
+ dataBuffer.writePosition(read);
+ release = false;
+ sub.next(dataBuffer);
}
else {
sub.complete();
@@ -448,62 +443,64 @@ public abstract class DataBufferUtils {
catch (IOException ex) {
sub.error(ex);
}
+ finally {
+ if (release) {
+ release(dataBuffer);
+ }
+ }
return channel;
}
}
private static class AsynchronousFileChannelReadCompletionHandler
- implements CompletionHandler {
+ implements CompletionHandler {
+
+ private final AsynchronousFileChannel channel;
private final FluxSink sink;
- private final ByteBuffer byteBuffer;
-
private final DataBufferFactory dataBufferFactory;
- private long position;
+ private final int bufferSize;
- private AsynchronousFileChannelReadCompletionHandler(FluxSink sink,
- long position, DataBufferFactory dataBufferFactory, ByteBuffer byteBuffer) {
+ private AtomicLong position;
+
+ private AsynchronousFileChannelReadCompletionHandler(
+ AsynchronousFileChannel channel, FluxSink sink,
+ long position, DataBufferFactory dataBufferFactory, int bufferSize) {
+ this.channel = channel;
this.sink = sink;
- this.position = position;
+ this.position = new AtomicLong(position);
this.dataBufferFactory = dataBufferFactory;
- this.byteBuffer = byteBuffer;
+ this.bufferSize = bufferSize;
}
@Override
- public void completed(Integer read, AsynchronousFileChannel channel) {
+ public void completed(Integer read, DataBuffer dataBuffer) {
if (read != -1) {
- this.position += read;
- this.byteBuffer.flip();
- boolean release = true;
- DataBuffer dataBuffer = this.dataBufferFactory.allocateBuffer(read);
- try {
- dataBuffer.write(this.byteBuffer);
- release = false;
- this.sink.next(dataBuffer);
- }
- finally {
- if (release) {
- release(dataBuffer);
- }
- }
- this.byteBuffer.clear();
+ long pos = this.position.addAndGet(read);
+ dataBuffer.writePosition(read);
+ this.sink.next(dataBuffer);
if (!this.sink.isCancelled()) {
- channel.read(this.byteBuffer, this.position, channel, this);
+ DataBuffer newDataBuffer =
+ this.dataBufferFactory.allocateBuffer(this.bufferSize);
+ ByteBuffer newByteBuffer = newDataBuffer.asByteBuffer(0, this.bufferSize);
+ this.channel.read(newByteBuffer, pos, newDataBuffer, this);
}
}
else {
+ release(dataBuffer);
+ closeChannel(this.channel);
this.sink.complete();
- closeChannel(channel);
}
}
@Override
- public void failed(Throwable exc, AsynchronousFileChannel channel) {
+ public void failed(Throwable exc, DataBuffer dataBuffer) {
+ release(dataBuffer);
+ closeChannel(this.channel);
this.sink.error(exc);
- closeChannel(channel);
}
}
@@ -558,9 +555,9 @@ public abstract class DataBufferUtils {
@Override
public void completed(Integer written, ByteBuffer byteBuffer) {
- this.position.addAndGet(written);
+ long pos = this.position.addAndGet(written);
if (byteBuffer.hasRemaining()) {
- this.channel.write(byteBuffer, this.position.get(), byteBuffer, this);
+ this.channel.write(byteBuffer, pos, byteBuffer, this);
return;
}
diff --git a/spring-core/src/main/java/org/springframework/core/io/buffer/DefaultDataBuffer.java b/spring-core/src/main/java/org/springframework/core/io/buffer/DefaultDataBuffer.java
index 29a260f830..f902b005cb 100644
--- a/spring-core/src/main/java/org/springframework/core/io/buffer/DefaultDataBuffer.java
+++ b/spring-core/src/main/java/org/springframework/core/io/buffer/DefaultDataBuffer.java
@@ -22,17 +22,20 @@ import java.io.OutputStream;
import java.nio.Buffer;
import java.nio.ByteBuffer;
import java.util.Arrays;
-import java.util.function.Function;
import java.util.function.IntPredicate;
import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;
+
/**
* Default implementation of the {@link DataBuffer} interface that uses a
- * {@link ByteBuffer} internally, with separate read and write positions.
+ * {@link ByteBuffer} internally. with separate read and write positions.
* Constructed using the {@link DefaultDataBufferFactory}.
*
+ * Inspired by Netty's {@code ByteBuf}. Introduced so that non-Netty runtimes (i.e. Servlet)
+ * do not require Netty on the classpath.
+ *
* @author Arjen Poutsma
* @author Juergen Hoeller
* @since 5.0
@@ -53,32 +56,29 @@ public class DefaultDataBuffer implements DataBuffer {
private int writePosition;
+ private int capacity;
- /**
- * Create a new {@code DefaultDataBuffer} based on the given
- * {@code ByteBuffer}. Both reading and writing position of this buffer are
- * based on the current {@linkplain
- * ByteBuffer#position() position} of the given buffer.
- * @param byteBuffer the buffer to base this buffer on
- */
- DefaultDataBuffer(ByteBuffer byteBuffer, DefaultDataBufferFactory dataBufferFactory) {
- this(byteBuffer, byteBuffer.position(), byteBuffer.position(), dataBufferFactory);
+ private DefaultDataBuffer(DefaultDataBufferFactory dataBufferFactory, ByteBuffer byteBuffer) {
+ Assert.notNull(dataBufferFactory, "'dataBufferFactory' must not be null");
+ Assert.notNull(byteBuffer, "'byteBuffer' must not be null");
+
+ this.dataBufferFactory = dataBufferFactory;
+ ByteBuffer slice = byteBuffer.slice();
+ this.byteBuffer = slice;
+ this.capacity = slice.remaining();
}
- DefaultDataBuffer(ByteBuffer byteBuffer, int readPosition, int writePosition,
- DefaultDataBufferFactory dataBufferFactory) {
+ static DefaultDataBuffer fromFilledByteBuffer(DefaultDataBufferFactory dataBufferFactory,
+ ByteBuffer byteBuffer) {
- Assert.notNull(byteBuffer, "'byteBuffer' must not be null");
- Assert.isTrue(readPosition >= 0, "'readPosition' must be 0 or higher");
- Assert.isTrue(writePosition >= 0, "'writePosition' must be 0 or higher");
- Assert.isTrue(readPosition <= writePosition,
- "'readPosition' must be smaller than or equal to 'writePosition'");
- Assert.notNull(dataBufferFactory, "'dataBufferFactory' must not be null");
+ DefaultDataBuffer dataBuffer = new DefaultDataBuffer(dataBufferFactory, byteBuffer);
+ dataBuffer.writePosition(byteBuffer.remaining());
+ return dataBuffer;
+ }
- this.byteBuffer = byteBuffer;
- this.readPosition = readPosition;
- this.writePosition = writePosition;
- this.dataBufferFactory = dataBufferFactory;
+ static DefaultDataBuffer fromEmptyByteBuffer(DefaultDataBufferFactory dataBufferFactory,
+ ByteBuffer byteBuffer) {
+ return new DefaultDataBuffer(dataBufferFactory, byteBuffer);
}
@@ -90,6 +90,11 @@ public class DefaultDataBuffer implements DataBuffer {
return this.byteBuffer;
}
+ private void setNativeBuffer(ByteBuffer byteBuffer) {
+ this.byteBuffer = byteBuffer;
+ this.capacity = byteBuffer.remaining();
+ }
+
@Override
public DefaultDataBufferFactory factory() {
return this.dataBufferFactory;
@@ -132,60 +137,151 @@ public class DefaultDataBuffer implements DataBuffer {
return this.writePosition - this.readPosition;
}
+ @Override
+ public int writableByteCount() {
+ return this.capacity - this.writePosition;
+ }
+
+ @Override
+ public int readPosition() {
+ return this.readPosition;
+ }
+
+ @Override
+ public DataBuffer readPosition(int readPosition) {
+ assertIndex(readPosition >= 0, "'readPosition' %d must be >= 0", readPosition);
+ assertIndex(readPosition <= this.writePosition, "'readPosition' %d must be <= %d",
+ readPosition, this.writePosition);
+
+ this.readPosition = readPosition;
+ return this;
+ }
+
+ @Override
+ public int writePosition() {
+ return this.writePosition;
+ }
+
+ @Override
+ public DataBuffer writePosition(int writePosition) {
+ assertIndex(writePosition >= this.readPosition, "'writePosition' %d must be >= %d",
+ writePosition, this.readPosition);
+ assertIndex(writePosition <= this.capacity, "'writePosition' %d must be <= %d",
+ writePosition, this.capacity);
+
+ this.writePosition = writePosition;
+ return this;
+ }
+
+ @Override
+ public int capacity() {
+ return this.capacity;
+ }
+
+ @Override
+ public DataBuffer capacity(int newCapacity) {
+ Assert.isTrue(newCapacity > 0,
+ String.format("'newCapacity' %d must be higher than 0", newCapacity));
+
+ int readPosition = readPosition();
+ int writePosition = writePosition();
+ int oldCapacity = capacity();
+
+ if (newCapacity > oldCapacity) {
+ ByteBuffer oldBuffer = this.byteBuffer;
+ ByteBuffer newBuffer = allocate(newCapacity, oldBuffer.isDirect());
+ ((Buffer) oldBuffer).position(0).limit(oldBuffer.capacity());
+ ((Buffer) newBuffer).position(0).limit(oldBuffer.capacity());
+ newBuffer.put(oldBuffer);
+ newBuffer.clear();
+ setNativeBuffer(newBuffer);
+ }
+ else if (newCapacity < oldCapacity) {
+ ByteBuffer oldBuffer = this.byteBuffer;
+ ByteBuffer newBuffer = allocate(newCapacity, oldBuffer.isDirect());
+ if (readPosition < newCapacity) {
+ if (writePosition > newCapacity) {
+ writePosition = newCapacity;
+ writePosition(writePosition);
+ }
+ ((Buffer) oldBuffer).position(readPosition).limit(writePosition);
+ ((Buffer) newBuffer).position(readPosition).limit(writePosition);
+ newBuffer.put(oldBuffer);
+ newBuffer.clear();
+ }
+ else {
+ readPosition(newCapacity);
+ writePosition(newCapacity);
+ }
+ setNativeBuffer(newBuffer);
+ }
+ return this;
+ }
+
+ private static ByteBuffer allocate(int capacity, boolean direct) {
+ return direct ? ByteBuffer.allocateDirect(capacity) : ByteBuffer.allocate(capacity);
+ }
+
@Override
public byte read() {
- return readInternal(ByteBuffer::get);
+ assertIndex(this.readPosition <= this.writePosition - 1, "readPosition %d must be <= %d",
+ this.readPosition, this.writePosition - 1);
+ int pos = this.readPosition;
+ byte b = this.byteBuffer.get(pos);
+ this.readPosition = pos + 1;
+ return b;
}
@Override
public DefaultDataBuffer read(byte[] destination) {
Assert.notNull(destination, "'destination' must not be null");
- readInternal(b -> b.get(destination));
+ read(destination, 0, destination.length);
return this;
}
@Override
public DefaultDataBuffer read(byte[] destination, int offset, int length) {
Assert.notNull(destination, "'destination' must not be null");
- readInternal(b -> b.get(destination, offset, length));
- return this;
- }
+ assertIndex(this.readPosition <= this.writePosition - length,
+ "readPosition %d and length %d should be smaller than writePosition %d",
+ this.readPosition, length, this.writePosition);
- /**
- * Internal read method that keeps track of the {@link #readPosition} before and after
- * applying the given function on {@link #byteBuffer}.
- */
- private T readInternal(Function function) {
- // Explicit cast for compatibility with covariant return type on JDK 9's ByteBuffer
- ((Buffer) this.byteBuffer).position(this.readPosition);
- try {
- return function.apply(this.byteBuffer);
- }
- finally {
- this.readPosition = this.byteBuffer.position();
- }
+ ByteBuffer tmp = this.byteBuffer.duplicate();
+ int limit = this.readPosition + length;
+ ((Buffer) tmp).clear().position(this.readPosition).limit(limit);
+ tmp.get(destination, offset, length);
+
+ this.readPosition += length;
+ return this;
}
@Override
public DefaultDataBuffer write(byte b) {
- ensureExtraCapacity(1);
- writeInternal(buffer -> buffer.put(b));
+ ensureCapacity(1);
+ int pos = this.writePosition;
+ this.byteBuffer.put(pos, b);
+ this.writePosition = pos + 1;
return this;
}
@Override
public DefaultDataBuffer write(byte[] source) {
Assert.notNull(source, "'source' must not be null");
- ensureExtraCapacity(source.length);
- writeInternal(buffer -> buffer.put(source));
+ write(source, 0, source.length);
return this;
}
@Override
public DefaultDataBuffer write(byte[] source, int offset, int length) {
Assert.notNull(source, "'source' must not be null");
- ensureExtraCapacity(length);
- writeInternal(buffer -> buffer.put(source, offset, length));
+ ensureCapacity(length);
+
+ ByteBuffer tmp = this.byteBuffer.duplicate();
+ int limit = this.writePosition + length;
+ ((Buffer) tmp).clear().position(this.writePosition).limit(limit);
+ tmp.put(source, offset, length);
+
+ this.writePosition += length;
return this;
}
@@ -203,30 +299,24 @@ public class DefaultDataBuffer implements DataBuffer {
@Override
public DefaultDataBuffer write(ByteBuffer... byteBuffers) {
Assert.notEmpty(byteBuffers, "'byteBuffers' must not be empty");
- int extraCapacity = Arrays.stream(byteBuffers).mapToInt(ByteBuffer::remaining).sum();
- ensureExtraCapacity(extraCapacity);
- Arrays.stream(byteBuffers)
- .forEach(byteBuffer -> writeInternal(buffer -> buffer.put(byteBuffer)));
+ int capacity = Arrays.stream(byteBuffers).mapToInt(ByteBuffer::remaining).sum();
+ ensureCapacity(capacity);
+ Arrays.stream(byteBuffers).forEach(this::write);
return this;
}
- /**
- * Internal write method that keeps track of the {@link #writePosition} before and
- * after applying the given function on {@link #byteBuffer}.
- */
- private T writeInternal(Function function) {
- // Explicit cast for compatibility with covariant return type on JDK 9's ByteBuffer
- ((Buffer) this.byteBuffer).position(this.writePosition);
- try {
- return function.apply(this.byteBuffer);
- }
- finally {
- this.writePosition = this.byteBuffer.position();
- }
+ private void write(ByteBuffer source) {
+ int length = source.remaining();
+ ByteBuffer tmp = this.byteBuffer.duplicate();
+ int limit = this.writePosition + source.remaining();
+ ((Buffer) tmp).clear().position(this.writePosition).limit(limit);
+ tmp.put(source);
+ this.writePosition += length;
}
@Override
public DataBuffer slice(int index, int length) {
+ checkIndex(index, length);
int oldPosition = this.byteBuffer.position();
// Explicit access via Buffer base type for compatibility
// with covariant return type on JDK 9's ByteBuffer...
@@ -236,7 +326,7 @@ public class DefaultDataBuffer implements DataBuffer {
ByteBuffer slice = this.byteBuffer.slice();
// Explicit cast for compatibility with covariant return type on JDK 9's ByteBuffer
((Buffer) slice).limit(length);
- return new SlicedDefaultDataBuffer(slice, 0, length, this.dataBufferFactory);
+ return new SlicedDefaultDataBuffer(slice, this.dataBufferFactory, length);
}
finally {
buffer.position(oldPosition);
@@ -245,13 +335,20 @@ public class DefaultDataBuffer implements DataBuffer {
@Override
public ByteBuffer asByteBuffer() {
+ return asByteBuffer(this.readPosition, readableByteCount());
+ }
+
+ @Override
+ public ByteBuffer asByteBuffer(int index, int length) {
+ checkIndex(index, length);
+
ByteBuffer duplicate = this.byteBuffer.duplicate();
// Explicit access via Buffer base type for compatibility
// with covariant return type on JDK 9's ByteBuffer...
Buffer buffer = duplicate;
- buffer.position(this.readPosition);
- buffer.limit(this.writePosition);
- return duplicate;
+ buffer.position(index);
+ buffer.limit(index + length);
+ return duplicate.slice();
}
@Override
@@ -264,11 +361,12 @@ public class DefaultDataBuffer implements DataBuffer {
return new DefaultDataBufferOutputStream();
}
- private void ensureExtraCapacity(int extraCapacity) {
- int neededCapacity = calculateCapacity(this.writePosition + extraCapacity);
- if (neededCapacity > this.byteBuffer.capacity()) {
- grow(neededCapacity);
+ private void ensureCapacity(int length) {
+ if (length <= writableByteCount()) {
+ return;
}
+ int newCapacity = calculateCapacity(this.writePosition + length);
+ capacity(newCapacity);
}
/**
@@ -299,24 +397,6 @@ public class DefaultDataBuffer implements DataBuffer {
}
}
- void grow(int capacity) {
- ByteBuffer oldBuffer = this.byteBuffer;
- ByteBuffer newBuffer =
- (oldBuffer.isDirect() ? ByteBuffer.allocateDirect(capacity) :
- ByteBuffer.allocate(capacity));
-
- final int remaining = readableByteCount();
- // Explicit cast for compatibility with covariant return type on JDK 9's ByteBuffer
- ((Buffer) oldBuffer).position(this.readPosition).limit(this.writePosition);
- newBuffer.put(oldBuffer);
-
- this.byteBuffer = newBuffer;
- this.readPosition = 0;
- this.writePosition = remaining;
- // Explicit cast for compatibility with covariant return type on JDK 9's ByteBuffer
- ((Buffer) oldBuffer).clear();
- }
-
@Override
public boolean equals(Object obj) {
@@ -339,67 +419,78 @@ public class DefaultDataBuffer implements DataBuffer {
@Override
public String toString() {
- return this.byteBuffer.toString();
+ return String.format("DefaultDataBuffer (r: %d, w %d, c %d)", this.readPosition,
+ this.writePosition, this.capacity);
}
+ private void checkIndex(int index, int length) {
+ assertIndex(index >= 0, "index %d must be >= 0", index);
+ assertIndex(length >= 0, "length %d must be >= 0", index);
+ assertIndex(index <= this.capacity, "index %d must be <= %d", index, this.capacity);
+ assertIndex(length <= this.capacity, "length %d must be <= %d", index, this.capacity);
+ }
+
+ private static void assertIndex(boolean expression, String format, Object... args) {
+ if (!expression) {
+ String message = String.format(format, args);
+ throw new IndexOutOfBoundsException(message);
+ }
+ }
private class DefaultDataBufferInputStream extends InputStream {
@Override
- public int available() throws IOException {
+ public int available() {
return readableByteCount();
}
@Override
public int read() {
- return readInternal(
- buffer -> readableByteCount() > 0 ? buffer.get() & 0xFF : -1);
+ return available() > 0 ? DefaultDataBuffer.this.read() & 0xFF : -1;
}
@Override
public int read(byte[] bytes, int off, int len) throws IOException {
- return readInternal(buffer -> {
- int count = readableByteCount();
- if (count > 0) {
- int minLen = Math.min(len, count);
- buffer.get(bytes, off, minLen);
- return minLen;
- }
- else {
- return -1;
- }
- });
+ int available = available();
+ if (available > 0) {
+ len = Math.min(len, available);
+ DefaultDataBuffer.this.read(bytes, off, len);
+ return len;
+ }
+ else {
+ return -1;
+ }
}
}
+
private class DefaultDataBufferOutputStream extends OutputStream {
@Override
public void write(int b) throws IOException {
- ensureExtraCapacity(1);
- writeInternal(buffer -> buffer.put((byte) b));
+ DefaultDataBuffer.this.write((byte) b);
}
@Override
public void write(byte[] bytes, int off, int len) throws IOException {
- ensureExtraCapacity(len);
- writeInternal(buffer -> buffer.put(bytes, off, len));
+ DefaultDataBuffer.this.write(bytes, off, len);
}
}
private static class SlicedDefaultDataBuffer extends DefaultDataBuffer {
- SlicedDefaultDataBuffer(ByteBuffer byteBuffer, int readPosition,
- int writePosition, DefaultDataBufferFactory dataBufferFactory) {
- super(byteBuffer, readPosition, writePosition, dataBufferFactory);
+ SlicedDefaultDataBuffer(ByteBuffer byteBuffer, DefaultDataBufferFactory dataBufferFactory,
+ int length) {
+ super(dataBufferFactory, byteBuffer);
+ writePosition(length);
}
@Override
- void grow(int capacity) {
+ public DataBuffer capacity(int newCapacity) {
throw new UnsupportedOperationException(
- "Growing the capacity of a sliced buffer is not supported");
+ "Changing the capacity of a sliced buffer is not supported");
}
}
diff --git a/spring-core/src/main/java/org/springframework/core/io/buffer/DefaultDataBufferFactory.java b/spring-core/src/main/java/org/springframework/core/io/buffer/DefaultDataBufferFactory.java
index 96d3a8303e..600b952046 100644
--- a/spring-core/src/main/java/org/springframework/core/io/buffer/DefaultDataBufferFactory.java
+++ b/spring-core/src/main/java/org/springframework/core/io/buffer/DefaultDataBufferFactory.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -83,21 +83,23 @@ public class DefaultDataBufferFactory implements DataBufferFactory {
@Override
public DefaultDataBuffer allocateBuffer(int initialCapacity) {
- return (this.preferDirect ?
- new DefaultDataBuffer(ByteBuffer.allocateDirect(initialCapacity), this) :
- new DefaultDataBuffer(ByteBuffer.allocate(initialCapacity), this));
+ ByteBuffer byteBuffer = (this.preferDirect ?
+ ByteBuffer.allocateDirect(initialCapacity) :
+ ByteBuffer.allocate(initialCapacity));
+
+ return DefaultDataBuffer.fromEmptyByteBuffer(this, byteBuffer);
}
@Override
public DefaultDataBuffer wrap(ByteBuffer byteBuffer) {
ByteBuffer sliced = byteBuffer.slice();
- return new DefaultDataBuffer(sliced, 0, byteBuffer.remaining(), this);
+ return DefaultDataBuffer.fromFilledByteBuffer(this, sliced);
}
@Override
public DataBuffer wrap(byte[] bytes) {
ByteBuffer wrapper = ByteBuffer.wrap(bytes);
- return new DefaultDataBuffer(wrapper, 0, bytes.length, this);
+ return DefaultDataBuffer.fromFilledByteBuffer(this, wrapper);
}
@Override
diff --git a/spring-core/src/main/java/org/springframework/core/io/buffer/NettyDataBuffer.java b/spring-core/src/main/java/org/springframework/core/io/buffer/NettyDataBuffer.java
index 366768473e..be6d93fea9 100644
--- a/spring-core/src/main/java/org/springframework/core/io/buffer/NettyDataBuffer.java
+++ b/spring-core/src/main/java/org/springframework/core/io/buffer/NettyDataBuffer.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -99,6 +99,44 @@ public class NettyDataBuffer implements PooledDataBuffer {
return this.byteBuf.readableBytes();
}
+ @Override
+ public int writableByteCount() {
+ return this.byteBuf.writableBytes();
+ }
+
+ @Override
+ public int readPosition() {
+ return this.byteBuf.readerIndex();
+ }
+
+ @Override
+ public DataBuffer readPosition(int readPosition) {
+ this.byteBuf.readerIndex(readPosition);
+ return this;
+ }
+
+ @Override
+ public int writePosition() {
+ return this.byteBuf.writerIndex();
+ }
+
+ @Override
+ public DataBuffer writePosition(int writePosition) {
+ this.byteBuf.writerIndex(writePosition);
+ return this;
+ }
+
+ @Override
+ public int capacity() {
+ return this.byteBuf.capacity();
+ }
+
+ @Override
+ public DataBuffer capacity(int capacity) {
+ this.byteBuf.capacity(capacity);
+ return this;
+ }
+
@Override
public byte read() {
return this.byteBuf.readByte();
@@ -194,6 +232,11 @@ public class NettyDataBuffer implements PooledDataBuffer {
return this.byteBuf.nioBuffer();
}
+ @Override
+ public ByteBuffer asByteBuffer(int index, int length) {
+ return this.byteBuf.nioBuffer(index, length);
+ }
+
@Override
public InputStream asInputStream() {
return new ByteBufInputStream(this.byteBuf);
diff --git a/spring-core/src/test/java/org/springframework/core/io/buffer/AbstractDataBufferAllocatingTestCase.java b/spring-core/src/test/java/org/springframework/core/io/buffer/AbstractDataBufferAllocatingTestCase.java
index b4417b0cec..87404da680 100644
--- a/spring-core/src/test/java/org/springframework/core/io/buffer/AbstractDataBufferAllocatingTestCase.java
+++ b/spring-core/src/test/java/org/springframework/core/io/buffer/AbstractDataBufferAllocatingTestCase.java
@@ -80,8 +80,8 @@ public abstract class AbstractDataBufferAllocatingTestCase {
return dataBuffer -> {
String value =
DataBufferTestUtils.dumpString(dataBuffer, StandardCharsets.UTF_8);
- assertEquals(expected, value);
DataBufferUtils.release(dataBuffer);
+ assertEquals(expected, value);
};
}
diff --git a/spring-core/src/test/java/org/springframework/core/io/buffer/DataBufferTests.java b/spring-core/src/test/java/org/springframework/core/io/buffer/DataBufferTests.java
index cd0d0aeb7e..d9efd3d685 100644
--- a/spring-core/src/test/java/org/springframework/core/io/buffer/DataBufferTests.java
+++ b/spring-core/src/test/java/org/springframework/core/io/buffer/DataBufferTests.java
@@ -20,7 +20,6 @@ import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.ByteBuffer;
-import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import org.junit.Test;
@@ -33,8 +32,107 @@ import static org.junit.Assert.*;
public class DataBufferTests extends AbstractDataBufferAllocatingTestCase {
@Test
- public void writeAndRead() {
+ public void byteCountsAndPositions() {
+ DataBuffer buffer = createDataBuffer(2);
+ assertEquals(0, buffer.readPosition());
+ assertEquals(0, buffer.writePosition());
+ assertEquals(0, buffer.readableByteCount());
+ assertEquals(2, buffer.writableByteCount());
+ assertEquals(2, buffer.capacity());
+
+ buffer.write((byte) 'a');
+ assertEquals(0, buffer.readPosition());
+ assertEquals(1, buffer.writePosition());
+ assertEquals(1, buffer.readableByteCount());
+ assertEquals(1, buffer.writableByteCount());
+ assertEquals(2, buffer.capacity());
+
+ buffer.write((byte) 'b');
+ assertEquals(0, buffer.readPosition());
+ assertEquals(2, buffer.writePosition());
+ assertEquals(2, buffer.readableByteCount());
+ assertEquals(0, buffer.writableByteCount());
+ assertEquals(2, buffer.capacity());
+
+ buffer.read();
+ assertEquals(1, buffer.readPosition());
+ assertEquals(2, buffer.writePosition());
+ assertEquals(1, buffer.readableByteCount());
+ assertEquals(0, buffer.writableByteCount());
+ assertEquals(2, buffer.capacity());
+
+ buffer.read();
+ assertEquals(2, buffer.readPosition());
+ assertEquals(2, buffer.writePosition());
+ assertEquals(0, buffer.readableByteCount());
+ assertEquals(0, buffer.writableByteCount());
+ assertEquals(2, buffer.capacity());
+
+ release(buffer);
+ }
+
+ @Test
+ public void readPositionSmallerThanZero() {
+ DataBuffer buffer = createDataBuffer(1);
+ try {
+ buffer.readPosition(-1);
+ fail("IndexOutOfBoundsException expected");
+ }
+ catch (IndexOutOfBoundsException ignored) {
+ }
+ finally {
+ release(buffer);
+ }
+ }
+
+ @Test
+ public void readPositionGreaterThanWritePosition() {
+ DataBuffer buffer = createDataBuffer(1);
+ try {
+ buffer.readPosition(1);
+ fail("IndexOutOfBoundsException expected");
+ }
+ catch (IndexOutOfBoundsException ignored) {
+ }
+ finally {
+ release(buffer);
+ }
+ }
+
+ @Test
+ public void writePositionSmallerThanReadPosition() {
+ DataBuffer buffer = createDataBuffer(2);
+ try {
+ buffer.write((byte) 'a');
+ buffer.read();
+
+ buffer.writePosition(0);
+ fail("IndexOutOfBoundsException expected");
+ }
+ catch (IndexOutOfBoundsException ignored) {
+ }
+ finally {
+ release(buffer);
+ }
+ }
+
+ @Test
+ public void writePositionGreaterThanCapacity() {
+ DataBuffer buffer = createDataBuffer(1);
+ try {
+ buffer.writePosition(2);
+ fail("IndexOutOfBoundsException expected");
+ }
+ catch (IndexOutOfBoundsException ignored) {
+ }
+ finally {
+ release(buffer);
+ }
+ }
+
+ @Test
+ public void writeAndRead() {
DataBuffer buffer = createDataBuffer(5);
buffer.write(new byte[]{'a', 'b', 'c'});
@@ -54,33 +152,32 @@ public class DataBufferTests extends AbstractDataBufferAllocatingTestCase {
@Test
public void inputStream() throws IOException {
- byte[] data = new byte[]{'a', 'b', 'c', 'd', 'e'};
-
DataBuffer buffer = createDataBuffer(4);
- buffer.write(data);
-
- buffer.read(); // readIndex++
+ buffer.write(new byte[]{'a', 'b', 'c', 'd', 'e'});
+ buffer.readPosition(1);
InputStream inputStream = buffer.asInputStream();
- int available = inputStream.available();
- assertEquals(4, available);
+ assertEquals(4, inputStream.available());
int result = inputStream.read();
assertEquals('b', result);
-
- available = inputStream.available();
- assertEquals(3, available);
+ assertEquals(3, inputStream.available());
byte[] bytes = new byte[2];
int len = inputStream.read(bytes);
assertEquals(2, len);
assertArrayEquals(new byte[]{'c', 'd'}, bytes);
+ assertEquals(1, inputStream.available());
Arrays.fill(bytes, (byte) 0);
len = inputStream.read(bytes);
assertEquals(1, len);
assertArrayEquals(new byte[]{'e', (byte) 0}, bytes);
+ assertEquals(0, inputStream.available());
+
+ assertEquals(-1, inputStream.read());
+ assertEquals(-1, inputStream.read(bytes));
release(buffer);
}
@@ -91,7 +188,8 @@ public class DataBufferTests extends AbstractDataBufferAllocatingTestCase {
buffer.write((byte) 'a');
OutputStream outputStream = buffer.asOutputStream();
- outputStream.write(new byte[]{'b', 'c', 'd'});
+ outputStream.write('b');
+ outputStream.write(new byte[]{'c', 'd'});
buffer.write((byte) 'e');
@@ -106,21 +204,60 @@ public class DataBufferTests extends AbstractDataBufferAllocatingTestCase {
public void expand() {
DataBuffer buffer = createDataBuffer(1);
buffer.write((byte) 'a');
+ assertEquals(1, buffer.capacity());
buffer.write((byte) 'b');
- byte[] result = new byte[2];
- buffer.read(result);
- assertArrayEquals(new byte[]{'a', 'b'}, result);
-
- buffer.write(new byte[]{'c', 'd'});
-
- result = new byte[2];
- buffer.read(result);
- assertArrayEquals(new byte[]{'c', 'd'}, result);
+ assertTrue(buffer.capacity() > 1);
release(buffer);
}
+ @Test
+ public void increaseCapacity() {
+ DataBuffer buffer = createDataBuffer(1);
+ assertEquals(1, buffer.capacity());
+
+ buffer.capacity(2);
+ assertEquals(2, buffer.capacity());
+
+ release(buffer);
+ }
+
+ @Test
+ public void decreaseCapacityLowReadPosition() {
+ DataBuffer buffer = createDataBuffer(2);
+ buffer.writePosition(2);
+ buffer.capacity(1);
+ assertEquals(1, buffer.capacity());
+
+ release(buffer);
+ }
+
+ @Test
+ public void decreaseCapacityHighReadPosition() {
+ DataBuffer buffer = createDataBuffer(2);
+ buffer.writePosition(2);
+ buffer.readPosition(2);
+ buffer.capacity(1);
+ assertEquals(1, buffer.capacity());
+
+ release(buffer);
+ }
+
+ @Test
+ public void capacityLessThanZero() {
+ DataBuffer buffer = createDataBuffer(1);
+ try {
+ buffer.capacity(-1);
+ fail("IllegalArgumentException expected");
+ }
+ catch (IllegalArgumentException ignored) {
+ }
+ finally {
+ release(buffer);
+ }
+ }
+
@Test
public void writeByteBuffer() {
DataBuffer buffer1 = createDataBuffer(1);
@@ -176,16 +313,74 @@ public class DataBufferTests extends AbstractDataBufferAllocatingTestCase {
buffer.read(); // skip a
ByteBuffer result = buffer.asByteBuffer();
+ assertEquals(2, result.capacity());
buffer.write((byte) 'd');
assertEquals(2, result.remaining());
+
byte[] resultBytes = new byte[2];
- buffer.read(resultBytes);
+ result.get(resultBytes);
assertArrayEquals(new byte[]{'b', 'c'}, resultBytes);
release(buffer);
}
+ @Test
+ public void asByteBufferIndexLength() {
+ DataBuffer buffer = createDataBuffer(3);
+ buffer.write(new byte[]{'a', 'b'});
+
+ ByteBuffer result = buffer.asByteBuffer(1, 2);
+ assertEquals(2, result.capacity());
+
+ buffer.write((byte) 'c');
+ assertEquals(2, result.remaining());
+
+ byte[] resultBytes = new byte[2];
+ result.get(resultBytes);
+ assertArrayEquals(new byte[]{'b', 'c'}, resultBytes);
+
+ release(buffer);
+ }
+
+ @Test
+ public void byteBufferContainsDataBufferChanges() {
+ DataBuffer dataBuffer = createDataBuffer(1);
+ ByteBuffer byteBuffer = dataBuffer.asByteBuffer(0, 1);
+
+ dataBuffer.write((byte) 'a');
+
+ assertEquals(1, byteBuffer.limit());
+ byte b = byteBuffer.get();
+ assertEquals('a', b);
+
+ release(dataBuffer);
+ }
+
+ @Test
+ public void dataBufferContainsByteBufferChanges() {
+ DataBuffer dataBuffer = createDataBuffer(1);
+ ByteBuffer byteBuffer = dataBuffer.asByteBuffer(0, 1);
+
+ byteBuffer.put((byte) 'a');
+ dataBuffer.writePosition(1);
+
+ byte b = dataBuffer.read();
+ assertEquals('a', b);
+
+ release(dataBuffer);
+ }
+
+ @Test
+ public void emptyAsByteBuffer() {
+ DataBuffer buffer = createDataBuffer(1);
+
+ ByteBuffer result = buffer.asByteBuffer();
+ assertEquals(0, result.capacity());
+
+ release(buffer);
+ }
+
@Test
public void indexOf() {
DataBuffer buffer = createDataBuffer(3);
@@ -244,7 +439,7 @@ public class DataBufferTests extends AbstractDataBufferAllocatingTestCase {
assertEquals(2, slice.readableByteCount());
try {
slice.write((byte) 0);
- fail("IndexOutOfBoundsException expected");
+ fail("Exception expected");
}
catch (Exception ignored) {
}
@@ -267,22 +462,4 @@ public class DataBufferTests extends AbstractDataBufferAllocatingTestCase {
}
- @Test
- public void growDataBuffer() {
- DataBuffer buffer = stringBuffer("Hello World!");
-
- byte[] bytes = new byte[5];
- buffer.read(bytes);
- assertArrayEquals("Hello".getBytes(StandardCharsets.UTF_8), bytes);
-
- buffer.write("!!".getBytes(StandardCharsets.UTF_8));
-
- bytes = new byte[9];
- buffer.read(bytes);
- assertArrayEquals(" World!!!".getBytes(StandardCharsets.UTF_8), bytes);
-
- release(buffer);
- }
-
-
}
\ No newline at end of file
diff --git a/spring-core/src/test/java/org/springframework/core/io/buffer/DataBufferUtilsTests.java b/spring-core/src/test/java/org/springframework/core/io/buffer/DataBufferUtilsTests.java
index 7b3b1a35a7..6b11ff09a6 100644
--- a/spring-core/src/test/java/org/springframework/core/io/buffer/DataBufferUtilsTests.java
+++ b/spring-core/src/test/java/org/springframework/core/io/buffer/DataBufferUtilsTests.java
@@ -19,8 +19,10 @@ package org.springframework.core.io.buffer;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.URI;
+import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousFileChannel;
import java.nio.channels.FileChannel;
+import java.nio.channels.ReadableByteChannel;
import java.nio.channels.WritableByteChannel;
import java.nio.file.Files;
import java.nio.file.Path;
@@ -30,6 +32,7 @@ import java.time.Duration;
import java.util.stream.Collectors;
import org.junit.Test;
+import org.mockito.stubbing.Answer;
import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;
@@ -37,6 +40,8 @@ import org.springframework.core.io.ClassPathResource;
import org.springframework.core.io.Resource;
import static org.junit.Assert.*;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.*;
/**
* @author Arjen Poutsma
@@ -291,4 +296,56 @@ public class DataBufferUtilsTests extends AbstractDataBufferAllocatingTestCase {
// AbstractDataBufferAllocatingTestCase.LeakDetector will assert the release of the buffers
}
+ @Test
+ public void SPR16070() throws Exception {
+ ReadableByteChannel channel = mock(ReadableByteChannel.class);
+ when(channel.read(any()))
+ .thenAnswer(putByte(1))
+ .thenAnswer(putByte(2))
+ .thenAnswer(putByte(3))
+ .thenReturn(-1);
+
+ Flux read = DataBufferUtils.read(channel, this.bufferFactory, 1);
+
+ StepVerifier.create(
+ read.reduce(DataBuffer::write)
+ .map(this::dataBufferToBytes)
+ .map(this::encodeHexString)
+ )
+ .expectNext("010203")
+ .verifyComplete();
+
+ }
+
+ private Answer putByte(int b) {
+ return invocation -> {
+ ByteBuffer buffer = invocation.getArgument(0);
+ buffer.put((byte) b);
+ return 1;
+ };
+ }
+
+ private byte[] dataBufferToBytes(DataBuffer buffer) {
+ try {
+ int byteCount = buffer.readableByteCount();
+ byte[] bytes = new byte[byteCount];
+ buffer.read(bytes);
+ return bytes;
+ }
+ finally {
+ release(buffer);
+ }
+ }
+
+ private String encodeHexString(byte[] data) {
+ StringBuilder builder = new StringBuilder();
+ for (byte b : data) {
+ builder.append((0xF0 & b) >>> 4);
+ builder.append(0x0F & b);
+ }
+ return builder.toString();
+ }
+
+
+
}
diff --git a/spring-web/src/main/java/org/springframework/http/server/reactive/TomcatHttpHandlerAdapter.java b/spring-web/src/main/java/org/springframework/http/server/reactive/TomcatHttpHandlerAdapter.java
index a42e9f02b1..9c028a045e 100644
--- a/spring-web/src/main/java/org/springframework/http/server/reactive/TomcatHttpHandlerAdapter.java
+++ b/spring-web/src/main/java/org/springframework/http/server/reactive/TomcatHttpHandlerAdapter.java
@@ -30,6 +30,7 @@ import org.apache.catalina.connector.CoyoteOutputStream;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferFactory;
+import org.springframework.core.io.buffer.DataBufferUtils;
/**
* {@link ServletHttpHandlerAdapter} extension that uses Tomcat APIs for reading
@@ -67,21 +68,32 @@ public class TomcatHttpHandlerAdapter extends ServletHttpHandlerAdapter {
@Override
protected DataBuffer readFromInputStream() throws IOException {
- DataBuffer buffer = getDataBufferFactory().allocateBuffer(getBufferSize());
- ByteBuffer byteBuffer = buffer.asByteBuffer();
- byteBuffer.limit(byteBuffer.capacity());
+ boolean release = true;
+ int capacity = getBufferSize();
+ DataBuffer dataBuffer = getDataBufferFactory().allocateBuffer(capacity);
+ try {
+ ByteBuffer byteBuffer = dataBuffer.asByteBuffer(0, capacity);
- ServletRequest request = getNativeRequest();
- int read = ((CoyoteInputStream) request.getInputStream()).read(byteBuffer);
- if (logger.isTraceEnabled()) {
- logger.trace("read:" + read);
+ ServletRequest request = getNativeRequest();
+ int read = ((CoyoteInputStream) request.getInputStream()).read(byteBuffer);
+ if (logger.isTraceEnabled()) {
+ logger.trace("read:" + read);
+ }
+
+ if (read > 0) {
+ dataBuffer.writePosition(read);
+ release = false;
+ return dataBuffer;
+ }
+ else {
+ return null;
+ }
}
-
- if (read > 0) {
- return getDataBufferFactory().wrap(byteBuffer);
+ finally {
+ if (release) {
+ DataBufferUtils.release(dataBuffer);
+ }
}
-
- return null;
}
}