From 38ab47f8a0d147d3b13cfb601d31f9957599983e Mon Sep 17 00:00:00 2001 From: Arjen Poutsma Date: Thu, 21 Jan 2016 10:33:47 +0100 Subject: [PATCH] Added DataBuffer abstraction Added DataBuffer and DataBufferAllocator, and provided a default NIO ByteBuffer-based implementation of those, as well as a Netty ByteBuf-based version. --- .../core/io/buffer/DataBuffer.java | 136 +++++++++ .../core/io/buffer/DataBufferAllocator.java | 66 ++++ .../core/io/buffer/DefaultDataBuffer.java | 288 ++++++++++++++++++ .../io/buffer/DefaultDataBufferAllocator.java | 85 ++++++ .../core/io/buffer/NettyDataBuffer.java | 199 ++++++++++++ .../io/buffer/NettyDataBufferAllocator.java | 94 ++++++ .../DataBufferPublisherInputStream.java | 151 +++++++++ .../io/buffer/support/DataBufferUtils.java | 49 +++ .../core/io/buffer/DataBufferTests.java | 207 +++++++++++++ 9 files changed, 1275 insertions(+) create mode 100644 spring-web-reactive/src/main/java/org/springframework/core/io/buffer/DataBuffer.java create mode 100644 spring-web-reactive/src/main/java/org/springframework/core/io/buffer/DataBufferAllocator.java create mode 100644 spring-web-reactive/src/main/java/org/springframework/core/io/buffer/DefaultDataBuffer.java create mode 100644 spring-web-reactive/src/main/java/org/springframework/core/io/buffer/DefaultDataBufferAllocator.java create mode 100644 spring-web-reactive/src/main/java/org/springframework/core/io/buffer/NettyDataBuffer.java create mode 100644 spring-web-reactive/src/main/java/org/springframework/core/io/buffer/NettyDataBufferAllocator.java create mode 100644 spring-web-reactive/src/main/java/org/springframework/core/io/buffer/support/DataBufferPublisherInputStream.java create mode 100644 spring-web-reactive/src/main/java/org/springframework/core/io/buffer/support/DataBufferUtils.java create mode 100644 spring-web-reactive/src/test/java/org/springframework/core/io/buffer/DataBufferTests.java diff --git a/spring-web-reactive/src/main/java/org/springframework/core/io/buffer/DataBuffer.java b/spring-web-reactive/src/main/java/org/springframework/core/io/buffer/DataBuffer.java new file mode 100644 index 0000000000..2af308228b --- /dev/null +++ b/spring-web-reactive/src/main/java/org/springframework/core/io/buffer/DataBuffer.java @@ -0,0 +1,136 @@ +/* + * 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; + +/** + * Basic abstraction over byte buffers. + * + *

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 T readInternal(Function function) { + this.byteBuffer.position(this.readPosition); + T result = function.apply(this.byteBuffer); + this.readPosition = this.byteBuffer.position(); + return result; + } + + @Override + public DefaultDataBuffer write(byte b) { + + ensureExtraCapacity(1); + writeInternal(buffer -> buffer.put(b)); + + return this; + } + + @Override + public DefaultDataBuffer write(byte[] source) { + Assert.notNull(source, "'source' must not be null"); + + ensureExtraCapacity(source.length); + writeInternal(buffer -> buffer.put(source)); + 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)); + return this; + } + + @Override + public DataBuffer write(DataBuffer... buffers) { + if (!ObjectUtils.isEmpty(buffers)) { + ByteBuffer[] byteBuffers = + Arrays.stream(buffers).map(DataBuffer::asByteBuffer) + .toArray(ByteBuffer[]::new); + write(byteBuffers); + } + return this; + } + + @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))); + + return this; + } + + /** + * Internal write method that keeps track of the {@link #writePosition} befor eand + * after applying the given function on {@link #byteBuffer}. + */ + private T writeInternal(Function function) { + this.byteBuffer.position(this.writePosition); + T result = function.apply(this.byteBuffer); + this.writePosition = this.byteBuffer.position(); + return result; + } + + @Override + public ByteBuffer asByteBuffer() { + ByteBuffer duplicate = this.byteBuffer.duplicate(); + duplicate.position(this.readPosition); + duplicate.limit(this.writePosition); + return duplicate; + } + + @Override + public InputStream asInputStream() { + return new DefaultDataBufferInputStream(); + } + + @Override + public OutputStream asOutputStream() { + return new DefaultDataBufferOutputStream(); + } + + private void ensureExtraCapacity(int extraCapacity) { + int neededCapacity = this.writePosition + extraCapacity; + if (neededCapacity > this.byteBuffer.capacity()) { + grow(neededCapacity); + } + } + + private void grow(int minCapacity) { + ByteBuffer oldBuffer = this.byteBuffer; + ByteBuffer newBuffer = + (oldBuffer.isDirect() ? ByteBuffer.allocateDirect(minCapacity) : + ByteBuffer.allocate(minCapacity)); + + oldBuffer.position(this.readPosition); + newBuffer.put(oldBuffer); + + this.byteBuffer = newBuffer; + oldBuffer.clear(); + } + + @Override + public int hashCode() { + return this.byteBuffer.hashCode(); + } + + @Override + public boolean equals(Object obj) { + if (this == obj) { + return true; + } + else if (obj instanceof DefaultDataBuffer) { + DefaultDataBuffer other = (DefaultDataBuffer) obj; + return this.byteBuffer.equals(other.byteBuffer); + } + return false; + } + + @Override + public String toString() { + return this.byteBuffer.toString(); + } + + private class DefaultDataBufferInputStream extends InputStream { + + @Override + public int available() throws IOException { + return byteBuffer.limit() - readPosition; + } + + @Override + public int read() { + return readInternal( + buffer -> buffer.hasRemaining() ? buffer.get() & 0xFF : -1); + } + + @Override + public int read(byte[] bytes, int off, int len) throws IOException { + return readInternal(buffer -> { + if (buffer.hasRemaining()) { + int minLen = Math.min(len, buffer.remaining()); + buffer.get(bytes, off, minLen); + return minLen; + } + else { + return -1; + } + }); + } + } + + private class DefaultDataBufferOutputStream extends OutputStream { + + @Override + public void write(int b) throws IOException { + ensureExtraCapacity(1); + writeInternal(buffer -> buffer.put((byte) b)); + } + + @Override + public void write(byte[] bytes, int off, int len) throws IOException { + ensureExtraCapacity(len); + writeInternal(buffer -> buffer.put(bytes, off, len)); + } + } +} diff --git a/spring-web-reactive/src/main/java/org/springframework/core/io/buffer/DefaultDataBufferAllocator.java b/spring-web-reactive/src/main/java/org/springframework/core/io/buffer/DefaultDataBufferAllocator.java new file mode 100644 index 0000000000..0f311978f5 --- /dev/null +++ b/spring-web-reactive/src/main/java/org/springframework/core/io/buffer/DefaultDataBufferAllocator.java @@ -0,0 +1,85 @@ +/* + * 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; + +/** + * Default implementation of the {@code DataBufferAllocator} interface. + * + *

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 queue; + + private InputStream currentStream; + + /** + * Creates a new {@code ByteArrayPublisherInputStream} based on the given publisher. + * @param publisher the publisher to use + */ + public DataBufferPublisherInputStream(Publisher publisher) { + this(publisher, 1); + } + + /** + * Creates a new {@code ByteArrayPublisherInputStream} based on the given publisher. + * @param publisher the publisher to use + * @param requestSize the {@linkplain Subscription#request(long) request size} to use + * on the publisher bound to Integer MAX + */ + public DataBufferPublisherInputStream(Publisher publisher, + int requestSize) { + Assert.notNull(publisher, "'publisher' must not be null"); + + // TODO Avoid using Reactor Stream, it should not be a mandatory dependency of Spring Reactive + this.queue = Stream.from(publisher).toBlockingQueue(requestSize); + } + + @Override + public int available() throws IOException { + if (completed.get()) { + return 0; + } + InputStream is = currentStream(); + return is != null ? is.available() : 0; + } + + @Override + public int read() throws IOException { + if (completed.get()) { + return -1; + } + InputStream is = currentStream(); + while (is != null) { + int ch = is.read(); + if (ch != -1) { + return ch; + } + else { + is = currentStream(); + } + } + return -1; + } + + @Override + public int read(byte[] b, int off, int len) throws IOException { + if (completed.get()) { + return -1; + } + InputStream is = currentStream(); + if (is == null) { + return -1; + } + else if (b == null) { + throw new NullPointerException(); + } + else if (off < 0 || len < 0 || len > b.length - off) { + throw new IndexOutOfBoundsException(); + } + else if (len == 0) { + return 0; + } + do { + int n = is.read(b, off, len); + if (n > 0) { + return n; + } + else { + is = currentStream(); + } + } + while (is != null); + + return -1; + } + + private InputStream currentStream() throws IOException { + try { + if (this.currentStream != null && this.currentStream.available() > 0) { + return this.currentStream; + } + else { + // take() blocks until next or complete() then return null, + // but that's OK since this is a *blocking* InputStream + DataBuffer signal = this.queue.take(); + if (signal == null) { + this.completed.set(true); + return null; + } + this.currentStream = signal.asInputStream(); + return this.currentStream; + } + } + catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + } + catch (Throwable error) { + this.completed.set(true); + throw new IOException(error); + } + throw new IOException(); + } + + +} diff --git a/spring-web-reactive/src/main/java/org/springframework/core/io/buffer/support/DataBufferUtils.java b/spring-web-reactive/src/main/java/org/springframework/core/io/buffer/support/DataBufferUtils.java new file mode 100644 index 0000000000..4839958233 --- /dev/null +++ b/spring-web-reactive/src/main/java/org/springframework/core/io/buffer/support/DataBufferUtils.java @@ -0,0 +1,49 @@ +/* + * 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.InputStream; + +import org.reactivestreams.Publisher; +import reactor.core.publisher.Flux; + +import org.springframework.core.io.buffer.DataBuffer; +import org.springframework.util.Assert; + +/** + * @author Arjen Poutsma + */ +public abstract class DataBufferUtils { + + public static Flux toPublisher(DataBuffer buffer) { + Assert.notNull(buffer, "'buffer' must not be null"); + + byte[] bytes1 = new byte[buffer.readableByteCount()]; + buffer.read(bytes1); + + Byte[] bytes2 = new Byte[bytes1.length]; + for (int i = 0; i < bytes1.length; i++) { + bytes2[i] = bytes1[i]; + } + return Flux.fromArray(bytes2); + } + + public static InputStream toInputStream(Publisher publisher) { + return new DataBufferPublisherInputStream(publisher); + } + +} diff --git a/spring-web-reactive/src/test/java/org/springframework/core/io/buffer/DataBufferTests.java b/spring-web-reactive/src/test/java/org/springframework/core/io/buffer/DataBufferTests.java new file mode 100644 index 0000000000..8c104a6069 --- /dev/null +++ b/spring-web-reactive/src/test/java/org/springframework/core/io/buffer/DataBufferTests.java @@ -0,0 +1,207 @@ +/* + * 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 io.netty.buffer.PooledByteBufAllocator; +import io.netty.buffer.UnpooledByteBufAllocator; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; + +/** + * @author Arjen Poutsma + */ +@RunWith(Parameterized.class) +public class DataBufferTests { + + @Parameterized.Parameter(0) + public DataBufferAllocator allocator; + + @Parameterized.Parameter(1) + public boolean direct; + + @Parameterized.Parameters(name = "{0} - direct: {1}") + public static Object[][] buffers() { + + return new Object[][]{ + {new NettyDataBufferAllocator(new UnpooledByteBufAllocator(false)), true}, + {new NettyDataBufferAllocator(new UnpooledByteBufAllocator(false)), + false}, + {new NettyDataBufferAllocator(new PooledByteBufAllocator(false)), true}, + {new NettyDataBufferAllocator(new PooledByteBufAllocator(false)), false}, + {new DefaultDataBufferAllocator(), true}, + {new DefaultDataBufferAllocator(), false}}; + } + + private DataBuffer createDataBuffer(int capacity) { + return direct ? allocator.allocateDirectBuffer(capacity) : + allocator.allocateHeapBuffer(capacity); + } + + @Test + public void writeAndRead() { + + DataBuffer buffer = createDataBuffer(5); + buffer.write(new byte[]{'a', 'b', 'c'}); + + int ch = buffer.read(); + assertEquals('a', ch); + + buffer.write((byte) 'd'); + buffer.write((byte) 'e'); + + byte[] result = new byte[4]; + buffer.read(result); + + assertArrayEquals(new byte[]{'b', 'c', 'd', 'e'}, result); + } + + @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++ + + InputStream inputStream = buffer.asInputStream(); + + int available = inputStream.available(); + assertEquals(4, available); + + int result = inputStream.read(); + assertEquals('b', result); + + available = inputStream.available(); + assertEquals(3, available); + + byte[] bytes = new byte[2]; + int len = inputStream.read(bytes); + assertEquals(2, len); + assertArrayEquals(new byte[]{'c', 'd'}, bytes); + + Arrays.fill(bytes, (byte) 0); + len = inputStream.read(bytes); + assertEquals(1, len); + assertArrayEquals(new byte[]{'e', (byte) 0}, bytes); + } + + @Test + public void outputStream() throws IOException { + DataBuffer buffer = createDataBuffer(4); + buffer.write((byte) 'a'); + + OutputStream outputStream = buffer.asOutputStream(); + outputStream.write(new byte[]{'b', 'c', 'd'}); + + buffer.write((byte) 'e'); + + byte[] bytes = new byte[5]; + buffer.read(bytes); + assertArrayEquals(new byte[]{'a', 'b', 'c', 'd', 'e'}, bytes); + } + + @Test + public void expand() { + DataBuffer buffer = createDataBuffer(1); + buffer.write((byte) 'a'); + 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); + } + + @Test + public void writeByteBuffer() { + DataBuffer buffer1 = createDataBuffer(1); + buffer1.write((byte) 'a'); + ByteBuffer buffer2 = createByteBuffer(2); + buffer2.put((byte) 'b'); + buffer2.flip(); + ByteBuffer buffer3 = createByteBuffer(3); + buffer3.put((byte) 'c'); + buffer3.flip(); + + buffer1.write(buffer2, buffer3); + buffer1.write((byte) 'd'); // make sure the write index is correctly set + + assertEquals(4, buffer1.readableByteCount()); + byte[] result = new byte[4]; + buffer1.read(result); + + assertArrayEquals(new byte[]{'a', 'b', 'c', 'd'}, result); + } + + @Test + public void writeDataBuffer() { + DataBuffer buffer1 = createDataBuffer(1); + buffer1.write((byte) 'a'); + DataBuffer buffer2 = createDataBuffer(2); + buffer2.write((byte) 'b'); + DataBuffer buffer3 = createDataBuffer(3); + buffer3.write((byte) 'c'); + + buffer1.write(buffer2, buffer3); + buffer1.write((byte) 'd'); // make sure the write index is correctly set + + assertEquals(4, buffer1.readableByteCount()); + byte[] result = new byte[4]; + buffer1.read(result); + + assertArrayEquals(new byte[]{'a', 'b', 'c', 'd'}, result); + } + + private ByteBuffer createByteBuffer(int capacity) { + return direct ? ByteBuffer.allocateDirect(capacity) : + ByteBuffer.allocate(capacity); + } + + @Test + public void asByteBuffer() { + DataBuffer buffer = createDataBuffer(4); + buffer.write(new byte[]{'a', 'b', 'c'}); + buffer.read(); // skip a + + ByteBuffer result = buffer.asByteBuffer(); + + buffer.write((byte) 'd'); + assertEquals(2, result.remaining()); + byte[] resultBytes = new byte[2]; + buffer.read(resultBytes); + assertArrayEquals(new byte[]{'b', 'c'}, resultBytes); + + } + + +} \ No newline at end of file