Mainly for internal use within the framework; consider Netty's + * {@link io.netty.buffer.ByteBuf} for a more comprehensive byte buffer. + * + * @author Arjen Poutsma + */ +public interface DataBuffer { + + /** + * Gets the byte at the specified index. + * @param index the index + * @return the byte at the specified index + * @throws IndexOutOfBoundsException if the given index is out of bounds + */ + byte get(int index); + + /** + * Returns the number of bytes that can be read from this data buffer. + * @return the readable byte count + */ + int readableByteCount(); + + /** + * Reads a single byte from the current reading position of this data buffer. + * @return the byte at this buffer's current reading position + */ + byte read(); + + /** + * Reads this buffer's data into the specified destination, starting at the current + * reading position of this buffer. + * + * @param destination the array into which the bytes are to be written + * @return this buffer + */ + DataBuffer read(byte[] destination); + + /** + * Reads at most {@code length} bytes of this buffer into the specified destination, + * starting at the current reading position of this buffer. + * @param destination the array into which the bytes are to be written + * @param offset the index within {@code destination} of the first byte to be written + * @param length the maximum number of bytes to be written in {@code destination} + * @return this buffer + */ + DataBuffer read(byte[] destination, int offset, int length); + + /** + * Write a single byte into this buffer at the current writing position. + * @param b the byte to be written + * @return this buffer + */ + DataBuffer write(byte b); + + /** + * Writes the given source into this buffer, startin at the current writing position + * of this buffer. + * @param source the bytes to be written into this buffer + * @return this buffer + */ + DataBuffer write(byte[] source); + + /** + * Writes at most {@code length} bytes of the given source into this buffer, starting + * at the current writing position of this buffer. + * @param source the bytes to be written into this buffer + * @param offset the index withing {@code source} to start writing from + * @param length the maximum number of bytes to be written from {@code source} + * @return this buffer + */ + DataBuffer write(byte[] source, int offset, int length); + + /** + * Writes one or more {@link DataBuffer} to this buffer, starting at the current + * writing position. + * @param buffers the byte buffers to write into this buffer + * @return this buffer + */ + DataBuffer write(DataBuffer... buffers); + + /** + * Writes one or more {@link ByteBuffer} to this buffer, starting at the current + * writing position. + * @param buffers the byte buffers to write into this buffer + * @return this buffer + */ + DataBuffer write(ByteBuffer... buffers); + + /** + * Exposes 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 position(s) of this data buffer. + * @return this data buffer as a byte buffer + */ + ByteBuffer asByteBuffer(); + + /** + * Exposes this buffer's data as an {@link InputStream}. Both data and position are + * shared between the returned stream and this data buffer. + * @return this data buffer as an input stream + */ + InputStream asInputStream(); + + /** + * Exposes this buffer's data as an {@link OutputStream}. Both data and position are + * shared between the returned stream and this data buffer. + * @return this data buffer as an output stream + */ + OutputStream asOutputStream(); + +} diff --git a/spring-web-reactive/src/main/java/org/springframework/core/io/buffer/DataBufferAllocator.java b/spring-web-reactive/src/main/java/org/springframework/core/io/buffer/DataBufferAllocator.java new file mode 100644 index 0000000000..e4586100f6 --- /dev/null +++ b/spring-web-reactive/src/main/java/org/springframework/core/io/buffer/DataBufferAllocator.java @@ -0,0 +1,66 @@ +/* + * Copyright 2002-2016 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.core.io.buffer; + +import java.nio.ByteBuffer; + +/** + * A factory for {@link DataBuffer}s, allowing for allocation of heap-based and direct + * data buffers. + * + * @author Arjen Poutsma + * @see DataBuffer + */ +public interface DataBufferAllocator { + + /** + * Allocates a data buffer of a default initial capacity. Depending on the underlying + * implementation and its configuration, this will be heap-based or direct buffer. + * @return the allocated buffer + */ + DataBuffer allocateBuffer(); + + /** + * Allocates a data buffer of the given initial capacity. Depending on the underlying + * implementation and its configuration, this will be heap-based or direct buffer. + * @param initialCapacity the initial capacity of the buffer to allocateBuffer + * @return the allocated buffer + */ + DataBuffer allocateBuffer(int initialCapacity); + + /** + * Allocates a data buffer of the given initial capacity on the heap. + * @param initialCapacity the initial capacity of the buffer to allocate + * @return the allocated buffer + */ + DataBuffer allocateHeapBuffer(int initialCapacity); + + /** + * Allocates a direct data buffer of the given initial capacity. + * @param initialCapacity the initial capacity of the buffer to allocate + * @return the allocated buffer + */ + DataBuffer allocateDirectBuffer(int initialCapacity); + + /** + * Wraps the given {@link ByteBuffer} in a {@code DataBuffer}. + * @param byteBuffer the NIO byte buffer to wrap + * @return the wrapped buffer + */ + DataBuffer wrap(ByteBuffer byteBuffer); + +} diff --git a/spring-web-reactive/src/main/java/org/springframework/core/io/buffer/DefaultDataBuffer.java b/spring-web-reactive/src/main/java/org/springframework/core/io/buffer/DefaultDataBuffer.java new file mode 100644 index 0000000000..5795c2a74b --- /dev/null +++ b/spring-web-reactive/src/main/java/org/springframework/core/io/buffer/DefaultDataBuffer.java @@ -0,0 +1,288 @@ +/* + * Copyright 2002-2016 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.core.io.buffer; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.nio.ByteBuffer; +import java.util.Arrays; +import java.util.function.Function; + +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. Typically constructed + * using the {@link DefaultDataBufferAllocator}. + * + *
This class is rather limited; consider using Netty's
+ * {@link io.netty.buffer.ByteBuf} and {@link NettyDataBuffer} for a more comprehensive byte buffer.
+
+ * @author Arjen Poutsma
+ * @see DefaultDataBufferAllocator
+ */
+public class DefaultDataBuffer implements DataBuffer {
+
+ private ByteBuffer byteBuffer;
+
+ private int readPosition;
+
+ private int writePosition;
+
+ /**
+ * Creates 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) {
+ this(byteBuffer, byteBuffer.position(), byteBuffer.position());
+ }
+
+ DefaultDataBuffer(ByteBuffer byteBuffer, int readPosition, int writePosition) {
+ 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'");
+
+ this.byteBuffer = byteBuffer;
+ this.readPosition = readPosition;
+ this.writePosition = writePosition;
+ }
+
+ /**
+ * Directly exposes the native {@code ByteBuffer} that this buffer is based on.
+ * @return the wrapped byte buffer
+ */
+ public ByteBuffer getNativeBuffer() {
+ return this.byteBuffer;
+ }
+
+ @Override
+ public byte get(int index) {
+ return this.byteBuffer.get(index);
+ }
+
+ @Override
+ public int readableByteCount() {
+ return this.writePosition - this.readPosition;
+ }
+
+ @Override
+ public byte read() {
+ return readInternal(ByteBuffer::get);
+ }
+
+ @Override
+ public DefaultDataBuffer read(byte[] destination) {
+ Assert.notNull(destination, "'destination' must not be null");
+
+ readInternal(b -> b.get(destination));
+
+ 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;
+ }
+
+ /**
+ * Internal read method that keeps track of the {@link #readPosition} before and after
+ * applying the given function on {@link #byteBuffer}.
+ */
+ private This class is rather limited; consider using Netty's
+ * {@link io.netty.buffer.ByteBuf} and {@link NettyDataBuffer} for a more comprehensive
+ * byte buffer.
+ * @author Arjen Poutsma
+ */
+public class DefaultDataBufferAllocator implements DataBufferAllocator {
+
+ public static final int DEFAULT_INITIAL_CAPACITY = 256;
+
+
+ private final boolean preferDirect;
+
+ /**
+ * Creates a new {@code DefaultDataBufferAllocator} with default settings.
+ */
+ public DefaultDataBufferAllocator() {
+ this(false);
+ }
+
+ /**
+ * Creates a new {@code DefaultDataBufferAllocator}, indicating whether direct buffers
+ * should be created by {@link #allocateBuffer(int)}.
+ * @param preferDirect {@code true} if direct buffers are to be preferred; {@code
+ * false} otherwise
+ */
+ public DefaultDataBufferAllocator(boolean preferDirect) {
+ this.preferDirect = preferDirect;
+ }
+
+ @Override
+ public DataBuffer allocateBuffer() {
+ return allocateBuffer(DEFAULT_INITIAL_CAPACITY);
+ }
+
+ @Override
+ public DefaultDataBuffer allocateBuffer(int initialCapacity) {
+ return preferDirect ? allocateDirectBuffer(initialCapacity) :
+ allocateHeapBuffer(initialCapacity);
+ }
+
+ @Override
+ public DefaultDataBuffer allocateHeapBuffer(int initialCapacity) {
+ return new DefaultDataBuffer(ByteBuffer.allocate(initialCapacity));
+ }
+
+ @Override
+ public DefaultDataBuffer allocateDirectBuffer(int initialCapacity) {
+ return new DefaultDataBuffer(ByteBuffer.allocateDirect(initialCapacity));
+ }
+
+ @Override
+ public DataBuffer wrap(ByteBuffer byteBuffer) {
+ ByteBuffer sliced = byteBuffer.slice();
+ return new DefaultDataBuffer(sliced, 0, byteBuffer.remaining());
+ }
+
+ @Override
+ public String toString() {
+ return "DefaultDataBufferFactory";
+ }
+
+}
diff --git a/spring-web-reactive/src/main/java/org/springframework/core/io/buffer/NettyDataBuffer.java b/spring-web-reactive/src/main/java/org/springframework/core/io/buffer/NettyDataBuffer.java
new file mode 100644
index 0000000000..f2a4bcf841
--- /dev/null
+++ b/spring-web-reactive/src/main/java/org/springframework/core/io/buffer/NettyDataBuffer.java
@@ -0,0 +1,199 @@
+/*
+ * Copyright 2002-2016 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.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.core.io.buffer;
+
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufInputStream;
+import io.netty.buffer.ByteBufOutputStream;
+import io.netty.buffer.CompositeByteBuf;
+import io.netty.buffer.Unpooled;
+
+import org.springframework.util.Assert;
+import org.springframework.util.ObjectUtils;
+
+/**
+ * Implementation of the {@code DataBuffer} interface that wraps a Netty {@link ByteBuf}.
+ * Typically constructed using the {@link NettyDataBufferAllocator}.
+ *
+ * @author Arjen Poutsma
+ */
+public class NettyDataBuffer implements DataBuffer {
+
+ private ByteBuf byteBuf;
+
+ /**
+ * Creates a new {@code NettyDataBuffer} based on the given {@code ByteBuff}.
+ * @param byteBuf the buffer to base this buffer on
+ */
+ public NettyDataBuffer(ByteBuf byteBuf) {
+ Assert.notNull(byteBuf, "'byteBuf' must not be null");
+
+ this.byteBuf = byteBuf;
+ }
+
+ /**
+ * Directly exposes the native {@code ByteBuf} that this buffer is based on.
+ * @return the wrapped byte buffer
+ */
+ public ByteBuf getNativeBuffer() {
+ return this.byteBuf;
+ }
+
+ @Override
+ public byte get(int index) {
+ return this.byteBuf.getByte(index);
+ }
+
+ @Override
+ public int readableByteCount() {
+ return this.byteBuf.readableBytes();
+ }
+
+ @Override
+ public byte read() {
+ return this.byteBuf.readByte();
+ }
+
+ @Override
+ public NettyDataBuffer read(byte[] destination) {
+ this.byteBuf.readBytes(destination);
+ return this;
+ }
+
+ @Override
+ public NettyDataBuffer read(byte[] destination, int offset, int length) {
+ this.byteBuf.readBytes(destination, offset, length);
+ return this;
+ }
+
+ @Override
+ public NettyDataBuffer write(byte b) {
+ this.byteBuf.writeByte(b);
+ return this;
+ }
+
+ @Override
+ public NettyDataBuffer write(byte[] source) {
+ this.byteBuf.writeBytes(source);
+ return this;
+ }
+
+ @Override
+ public NettyDataBuffer write(byte[] source, int offset, int length) {
+ this.byteBuf.writeBytes(source, offset, length);
+ return this;
+ }
+
+ @Override
+ public NettyDataBuffer write(DataBuffer... buffers) {
+ if (!ObjectUtils.isEmpty(buffers)) {
+ if (buffers[0] instanceof NettyDataBuffer) {
+ NettyDataBuffer[] copy =
+ Arrays.copyOf(buffers, buffers.length, NettyDataBuffer[].class);
+
+ ByteBuf[] nativeBuffers =
+ Arrays.stream(copy).map(NettyDataBuffer::getNativeBuffer)
+ .toArray(ByteBuf[]::new);
+
+ write(nativeBuffers);
+ }
+ else {
+ ByteBuffer[] byteBuffers =
+ Arrays.stream(buffers).map(DataBuffer::asByteBuffer)
+ .toArray(ByteBuffer[]::new);
+ write(byteBuffers);
+ }
+ }
+ return this;
+ }
+
+ @Override
+ public NettyDataBuffer write(ByteBuffer... buffers) {
+ Assert.notNull(buffers, "'buffers' must not be null");
+
+ ByteBuf[] wrappedBuffers = Arrays.stream(buffers).map(Unpooled::wrappedBuffer)
+ .toArray(ByteBuf[]::new);
+
+ return write(wrappedBuffers);
+ }
+
+ /**
+ * Writes one or more Netty {@link ByteBuf}s to this buffer, starting at the current
+ * writing position.
+ * @param byteBufs the buffers to write into this buffer
+ * @return this buffer
+ */
+ public NettyDataBuffer write(ByteBuf... byteBufs) {
+ Assert.notNull(byteBufs, "'byteBufs' must not be null");
+
+ CompositeByteBuf composite =
+ new CompositeByteBuf(this.byteBuf.alloc(), this.byteBuf.isDirect(),
+ byteBufs.length + 1);
+ composite.addComponent(this.byteBuf);
+ Arrays.stream(byteBufs).forEach(composite::addComponent);
+
+ int writerIndex = this.byteBuf.readableBytes() +
+ Arrays.stream(byteBufs).mapToInt(ByteBuf::readableBytes).sum();
+ composite.writerIndex(writerIndex);
+
+ this.byteBuf = composite;
+
+ return this;
+ }
+
+ @Override
+ public ByteBuffer asByteBuffer() {
+ return this.byteBuf.nioBuffer();
+ }
+
+ @Override
+ public InputStream asInputStream() {
+ return new ByteBufInputStream(this.byteBuf);
+ }
+
+ @Override
+ public OutputStream asOutputStream() {
+ return new ByteBufOutputStream(this.byteBuf);
+ }
+
+ @Override
+ public int hashCode() {
+ return this.byteBuf.hashCode();
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ else if (obj instanceof NettyDataBuffer) {
+ NettyDataBuffer other = (NettyDataBuffer) obj;
+ return this.byteBuf.equals(other.byteBuf);
+ }
+ return false;
+ }
+
+ @Override
+ public String toString() {
+ return this.byteBuf.toString();
+ }
+}
diff --git a/spring-web-reactive/src/main/java/org/springframework/core/io/buffer/NettyDataBufferAllocator.java b/spring-web-reactive/src/main/java/org/springframework/core/io/buffer/NettyDataBufferAllocator.java
new file mode 100644
index 0000000000..6eea0f3298
--- /dev/null
+++ b/spring-web-reactive/src/main/java/org/springframework/core/io/buffer/NettyDataBufferAllocator.java
@@ -0,0 +1,94 @@
+/*
+ * Copyright 2002-2016 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.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.core.io.buffer;
+
+import java.nio.ByteBuffer;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.Unpooled;
+
+import org.springframework.util.Assert;
+
+/**
+ * Implemtation of the {@code DataBufferAllocator} interface based on a Netty
+ * {@link ByteBufAllocator}.
+ *
+ * @author Arjen Poutsma
+ * @see io.netty.buffer.PooledByteBufAllocator
+ * @see io.netty.buffer.UnpooledByteBufAllocator
+ */
+public class NettyDataBufferAllocator implements DataBufferAllocator {
+
+ private final ByteBufAllocator byteBufAllocator;
+
+ /**
+ * Creates a new {@code NettyDataBufferAllocator} based on the given allocator.
+ * @param byteBufAllocator the allocator to use
+ * @see io.netty.buffer.PooledByteBufAllocator
+ * @see io.netty.buffer.UnpooledByteBufAllocator
+ */
+ public NettyDataBufferAllocator(ByteBufAllocator byteBufAllocator) {
+ Assert.notNull(byteBufAllocator, "'byteBufAllocator' must not be null");
+
+ this.byteBufAllocator = byteBufAllocator;
+ }
+
+ @Override
+ public NettyDataBuffer allocateBuffer() {
+ ByteBuf byteBuf = this.byteBufAllocator.buffer();
+ return new NettyDataBuffer(byteBuf);
+ }
+
+ @Override
+ public NettyDataBuffer allocateBuffer(int initialCapacity) {
+ ByteBuf byteBuf = this.byteBufAllocator.buffer(initialCapacity);
+ return new NettyDataBuffer(byteBuf);
+ }
+
+ @Override
+ public NettyDataBuffer allocateHeapBuffer(int initialCapacity) {
+ ByteBuf byteBuf = this.byteBufAllocator.heapBuffer(initialCapacity);
+ return new NettyDataBuffer(byteBuf);
+ }
+
+ @Override
+ public NettyDataBuffer allocateDirectBuffer(int initialCapacity) {
+ ByteBuf byteBuf = this.byteBufAllocator.directBuffer(initialCapacity);
+ return new NettyDataBuffer(byteBuf);
+ }
+
+ @Override
+ public NettyDataBuffer wrap(ByteBuffer byteBuffer) {
+ ByteBuf byteBuf = Unpooled.wrappedBuffer(byteBuffer);
+ return new NettyDataBuffer(byteBuf);
+ }
+
+ /**
+ * Wraps the given Netty {@link ByteBuf} in a {@code NettyDataBuffer}.
+ * @param byteBuf the Netty byte buffer to wrap
+ * @return the wrapped buffer
+ */
+ public NettyDataBuffer wrap(ByteBuf byteBuf) {
+ return new NettyDataBuffer(byteBuf);
+ }
+
+ @Override
+ public String toString() {
+ return "NettyDataBufferAllocator (" + this.byteBufAllocator + ")";
+ }
+}
diff --git a/spring-web-reactive/src/main/java/org/springframework/core/io/buffer/support/DataBufferPublisherInputStream.java b/spring-web-reactive/src/main/java/org/springframework/core/io/buffer/support/DataBufferPublisherInputStream.java
new file mode 100644
index 0000000000..abcd0ddfea
--- /dev/null
+++ b/spring-web-reactive/src/main/java/org/springframework/core/io/buffer/support/DataBufferPublisherInputStream.java
@@ -0,0 +1,151 @@
+/*
+ * Copyright 2002-2016 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.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.core.io.buffer.support;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscription;
+import reactor.rx.Stream;
+
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.util.Assert;
+
+/**
+ * @author Arjen Poutsma
+ */
+class DataBufferPublisherInputStream extends InputStream {
+
+ private final AtomicBoolean completed = new AtomicBoolean();
+
+ private final BlockingQueue