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.

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.
+ *
+ * <p>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();
+
+}

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);
+
+}

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}.
+ *
+ * <p>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> T readInternal(Function<ByteBuffer, T> 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> T writeInternal(Function<ByteBuffer, T> 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));
+		}
+	}
+}

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.
+ *
+ * <p>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";
+	}
+
+}

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();
+	}
+}

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 + ")";
+	}
+}

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<DataBuffer> queue;
+
+	private InputStream currentStream;
+
+	/**
+	 * Creates a new {@code ByteArrayPublisherInputStream} based on the given publisher.
+	 * @param publisher the publisher to use
+	 */
+	public DataBufferPublisherInputStream(Publisher<DataBuffer> 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<DataBuffer> 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();
+	}
+
+
+}

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<Byte> 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<DataBuffer> publisher) {
+		return new DataBufferPublisherInputStream(publisher);
+	}
+
+}

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);
+
+	}
+
+
+}