Move spring-web-reactive classes to spring-core

spring-core/src/main/java/org/springframework/core/codec/AbstractDecoder.java

@@ -0,0 +1,61 @@
+/*
+ * 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.codec;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Mono;
+
+import org.springframework.core.ResolvableType;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.util.MimeType;
+
+/**
+ * @author Sebastien Deleuze
+ * @author Arjen Poutsma
+ */
+public abstract class AbstractDecoder<T> implements Decoder<T> {
+
+	private List<MimeType> decodableMimeTypes = Collections.emptyList();
+
+	protected AbstractDecoder(MimeType... supportedMimeTypes) {
+		this.decodableMimeTypes = Arrays.asList(supportedMimeTypes);
+	}
+
+
+	@Override
+	public List<MimeType> getDecodableMimeTypes() {
+		return this.decodableMimeTypes;
+	}
+
+	@Override
+	public boolean canDecode(ResolvableType elementType, MimeType mimeType, Object... hints) {
+		if (mimeType == null) {
+			return true;
+		}
+		return this.decodableMimeTypes.stream().
+				anyMatch(mt -> mt.isCompatibleWith(mimeType));
+	}
+
+	@Override
+	public Mono<T> decodeToMono(Publisher<DataBuffer> inputStream, ResolvableType elementType, MimeType mimeType, Object... hints) {
+		throw new UnsupportedOperationException();
+	}
+}

spring-core/src/main/java/org/springframework/core/codec/AbstractEncoder.java

@@ -0,0 +1,53 @@
+/*
+ * 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.codec;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import org.springframework.core.ResolvableType;
+import org.springframework.util.MimeType;
+
+/**
+ * @author Sebastien Deleuze
+ * @author Arjen Poutsma
+ */
+public abstract class AbstractEncoder<T> implements Encoder<T> {
+
+	private List<MimeType> encodableMimeTypes = Collections.emptyList();
+
+	protected AbstractEncoder(MimeType... supportedMimeTypes) {
+		this.encodableMimeTypes = Arrays.asList(supportedMimeTypes);
+	}
+
+
+	@Override
+	public List<MimeType> getEncodableMimeTypes() {
+		return this.encodableMimeTypes;
+	}
+
+	@Override
+	public boolean canEncode(ResolvableType elementType, MimeType mimeType, Object... hints) {
+		if (mimeType == null) {
+			return true;
+		}
+		return this.encodableMimeTypes.stream().
+				anyMatch(mt -> mt.isCompatibleWith(mimeType));
+	}
+
+}

spring-core/src/main/java/org/springframework/core/codec/AbstractSingleValueEncoder.java

@@ -0,0 +1,68 @@
+/*
+ * 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.codec;
+
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+
+import org.springframework.core.ResolvableType;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferFactory;
+import org.springframework.util.MimeType;
+
+/**
+ * Abstract base class for {@link org.springframework.core.codec.Encoder} classes that
+ * can only deal with a single value.
+ * @author Arjen Poutsma
+ */
+public abstract class AbstractSingleValueEncoder<T> extends AbstractEncoder<T> {
+
+	public AbstractSingleValueEncoder(MimeType... supportedMimeTypes) {
+		super(supportedMimeTypes);
+	}
+
+	@Override
+	public final Flux<DataBuffer> encode(Publisher<? extends T> inputStream,
+			DataBufferFactory bufferFactory, ResolvableType elementType, MimeType mimeType,
+			Object... hints) {
+		return Flux.from(inputStream).
+				take(1).
+				concatMap(t -> {
+					try {
+						return encode(t, bufferFactory, elementType, mimeType);
+					}
+					catch (Exception ex) {
+						return Flux.error(ex);
+					}
+				});
+	}
+
+	/**
+	 * Encodes {@code T} to an output {@link DataBuffer} stream.
+	 * @param t the value to process
+	 * @param dataBufferFactory a buffer factory used to create the output
+	 * @param type the stream element type to process
+	 * @param mimeType the mime type to process
+	 * @param hints Additional information about how to do decode, optional
+	 * @return the output stream
+	 * @throws Exception in case of errors
+	 */
+	protected abstract Flux<DataBuffer> encode(T t, DataBufferFactory dataBufferFactory,
+			ResolvableType type, MimeType mimeType, Object... hints) throws Exception;
+
+
+}

spring-core/src/main/java/org/springframework/core/codec/ByteBufferDecoder.java

@@ -0,0 +1,60 @@
+/*
+ * 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.codec;
+
+import java.nio.ByteBuffer;
+
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+
+import org.springframework.core.ResolvableType;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.support.DataBufferUtils;
+import org.springframework.util.MimeType;
+import org.springframework.util.MimeTypeUtils;
+
+/**
+ * @author Sebastien Deleuze
+ * @author Arjen Poutsma
+ */
+public class ByteBufferDecoder extends AbstractDecoder<ByteBuffer> {
+
+
+	public ByteBufferDecoder() {
+		super(MimeTypeUtils.ALL);
+	}
+
+
+	@Override
+	public boolean canDecode(ResolvableType elementType, MimeType mimeType, Object... hints) {
+		Class<?> clazz = elementType.getRawClass();
+		return (super.canDecode(elementType, mimeType, hints) && ByteBuffer.class.isAssignableFrom(clazz));
+	}
+
+	@Override
+	public Flux<ByteBuffer> decode(Publisher<DataBuffer> inputStream, ResolvableType elementType,
+			MimeType mimeType, Object... hints) {
+		return Flux.from(inputStream).map((dataBuffer) -> {
+			ByteBuffer copy = ByteBuffer.allocate(dataBuffer.readableByteCount());
+			copy.put(dataBuffer.asByteBuffer());
+			copy.flip();
+			DataBufferUtils.release(dataBuffer);
+			return copy;
+		});
+	}
+
+}

spring-core/src/main/java/org/springframework/core/codec/ByteBufferEncoder.java

@@ -0,0 +1,54 @@
+/*
+ * 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.codec;
+
+import java.nio.ByteBuffer;
+
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+
+import org.springframework.core.ResolvableType;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferFactory;
+import org.springframework.util.MimeType;
+import org.springframework.util.MimeTypeUtils;
+
+/**
+ * @author Sebastien Deleuze
+ */
+public class ByteBufferEncoder extends AbstractEncoder<ByteBuffer> {
+
+	public ByteBufferEncoder() {
+		super(MimeTypeUtils.ALL);
+	}
+
+
+	@Override
+	public boolean canEncode(ResolvableType elementType, MimeType mimeType, Object... hints) {
+		Class<?> clazz = elementType.getRawClass();
+		return (super.canEncode(elementType, mimeType, hints) && ByteBuffer.class.isAssignableFrom(clazz));
+	}
+
+	@Override
+	public Flux<DataBuffer> encode(Publisher<? extends ByteBuffer> inputStream,
+			DataBufferFactory bufferFactory, ResolvableType elementType, MimeType mimeType,
+			Object... hints) {
+
+		return Flux.from(inputStream).map(bufferFactory::wrap);
+	}
+
+}

spring-core/src/main/java/org/springframework/core/codec/CodecException.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2002-2015 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.codec;
+
+import org.springframework.core.NestedRuntimeException;
+
+/**
+ * Codec related exception, usually used as a wrapper for a cause exception.
+ *
+ * @author Sebastien Deleuze
+ */
+@SuppressWarnings("serial")
+public class CodecException extends NestedRuntimeException {
+
+	public CodecException(String msg, Throwable cause) {
+		super(msg, cause);
+	}
+
+	public CodecException(String msg) {
+		super(msg);
+	}
+
+}

spring-core/src/main/java/org/springframework/core/codec/Decoder.java

@@ -0,0 +1,83 @@
+/*
+ * 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.codec;
+
+import java.util.List;
+
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.core.ResolvableType;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.util.MimeType;
+
+/**
+ * Strategy for decoding a {@link DataBuffer} input stream into an output stream
+ * of elements of type {@code <T>}.
+ *
+ * @author Sebastien Deleuze
+ * @author Rossen Stoyanchev
+ * @param <T> the type of elements in the output stream
+ */
+public interface Decoder<T> {
+
+	/**
+	 * Whether the decoder supports the given target element type and the MIME
+	 * type of the source stream.
+	 *
+	 * @param elementType the target element type for the output stream
+	 * @param mimeType the mime type associated with the stream to decode
+	 * @param hints additional information about how to do decode, optional
+	 * @return {@code true} if supported, {@code false} otherwise
+	 */
+	boolean canDecode(ResolvableType elementType, MimeType mimeType, Object... hints);
+
+	/**
+	 * Decode a {@link DataBuffer} input stream into a Flux of {@code T}.
+	 *
+	 * @param inputStream the {@code DataBuffer} input stream to decode
+	 * @param elementType the expected type of elements in the output stream;
+	 * this type must have been previously passed to the {@link #canDecode}
+	 * method and it must have returned {@code true}.
+	 * @param mimeType the MIME type associated with the input stream, optional
+	 * @param hints additional information about how to do decode, optional
+	 * @return the output stream with decoded elements
+	 */
+	Flux<T> decode(Publisher<DataBuffer> inputStream, ResolvableType elementType,
+			MimeType mimeType, Object... hints);
+
+	/**
+	 * Decode a {@link DataBuffer} input stream into a Mono of {@code T}.
+	 *
+	 * @param inputStream the {@code DataBuffer} input stream to decode
+	 * @param elementType the expected type of elements in the output stream;
+	 * this type must have been previously passed to the {@link #canDecode}
+	 * method and it must have returned {@code true}.
+	 * @param mimeType the MIME type associated with the input stream, optional
+	 * @param hints additional information about how to do decode, optional
+	 * @return the output stream with the decoded element
+	 */
+	Mono<T> decodeToMono(Publisher<DataBuffer> inputStream, ResolvableType elementType,
+			MimeType mimeType, Object... hints);
+
+	/**
+	 * Return the list of MIME types this decoder supports.
+	 */
+	List<MimeType> getDecodableMimeTypes();
+
+}

spring-core/src/main/java/org/springframework/core/codec/Encoder.java

@@ -0,0 +1,71 @@
+/*
+ * 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.codec;
+
+import java.util.List;
+
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+
+import org.springframework.core.ResolvableType;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferFactory;
+import org.springframework.util.MimeType;
+
+/**
+ * Strategy to encode a stream of Objects of type {@code <T>} into an output
+ * stream of bytes.
+ *
+ * @author Sebastien Deleuze
+ * @author Rossen Stoyanchev
+ * @param <T> the type of elements in the input stream
+ */
+public interface Encoder<T> {
+
+	/**
+	 * Whether the encoder supports the given source element type and the MIME
+	 * type for the output stream.
+	 *
+	 * @param elementType the type of elements in the source stream
+	 * @param mimeType the MIME type for the output stream
+	 * @param hints additional information about how to do encode, optional
+	 * @return {@code true} if supported, {@code false} otherwise
+	 */
+	boolean canEncode(ResolvableType elementType, MimeType mimeType, Object... hints);
+
+	/**
+	 * Encode a stream of Objects of type {@code T} into a {@link DataBuffer}
+	 * output stream.
+	 *
+	 * @param inputStream the input stream of Objects to encode
+	 * @param bufferFactory for creating output stream {@code DataBuffer}'s
+	 * @param elementType the expected type of elements in the input stream;
+	 * this type must have been previously passed to the {@link #canEncode}
+	 * method and it must have returned {@code true}.
+	 * @param mimeType the MIME type for the output stream
+	 * @param hints additional information about how to do encode, optional
+	 * @return the output stream
+	 */
+	Flux<DataBuffer> encode(Publisher<? extends T> inputStream, DataBufferFactory bufferFactory,
+			ResolvableType elementType, MimeType mimeType, Object... hints);
+
+	/**
+	 * Return the list of mime types this encoder supports.
+	 */
+	List<MimeType> getEncodableMimeTypes();
+
+}

spring-core/src/main/java/org/springframework/core/codec/ResourceDecoder.java

@@ -0,0 +1,82 @@
+/*
+ * 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.codec;
+
+import java.io.ByteArrayInputStream;
+
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.core.ResolvableType;
+import org.springframework.core.io.ByteArrayResource;
+import org.springframework.core.io.InputStreamResource;
+import org.springframework.core.io.Resource;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.support.DataBufferUtils;
+import org.springframework.util.MimeType;
+import org.springframework.util.MimeTypeUtils;
+
+/**
+ * A decoder for {@link Resource}s.
+ *
+ * @author Arjen Poutsma
+ */
+public class ResourceDecoder extends AbstractDecoder<Resource> {
+
+	public ResourceDecoder() {
+		super(MimeTypeUtils.ALL);
+	}
+
+	@Override
+	public boolean canDecode(ResolvableType elementType, MimeType mimeType, Object... hints) {
+		Class<?> clazz = elementType.getRawClass();
+		return (InputStreamResource.class.equals(clazz) ||
+				clazz.isAssignableFrom(ByteArrayResource.class)) &&
+				super.canDecode(elementType, mimeType, hints);
+	}
+
+	@Override
+	public Flux<Resource> decode(Publisher<DataBuffer> inputStream, ResolvableType elementType,
+			MimeType mimeType, Object... hints) {
+		Class<?> clazz = elementType.getRawClass();
+
+		Mono<byte[]> byteArray = Flux.from(inputStream).
+				reduce(DataBuffer::write).
+				map(dataBuffer -> {
+					byte[] bytes = new byte[dataBuffer.readableByteCount()];
+					dataBuffer.read(bytes);
+					DataBufferUtils.release(dataBuffer);
+					return bytes;
+				});
+
+
+		if (InputStreamResource.class.equals(clazz)) {
+			return Flux.from(byteArray.
+					map(ByteArrayInputStream::new).
+					map(InputStreamResource::new));
+		}
+		else if (clazz.isAssignableFrom(ByteArrayResource.class)) {
+			return Flux.from(byteArray.
+					map(ByteArrayResource::new));
+		}
+		else {
+			return Flux.error(new IllegalStateException(
+					"Unsupported resource class: " + clazz));
+		}
+	}
+}

spring-core/src/main/java/org/springframework/core/codec/ResourceEncoder.java

@@ -0,0 +1,69 @@
+/*
+ * 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.codec;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+import reactor.core.publisher.Flux;
+
+import org.springframework.core.ResolvableType;
+import org.springframework.core.io.Resource;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferFactory;
+import org.springframework.core.io.buffer.support.DataBufferUtils;
+import org.springframework.util.Assert;
+import org.springframework.util.MimeType;
+import org.springframework.util.MimeTypeUtils;
+import org.springframework.util.StreamUtils;
+
+/**
+ * An encoder for {@link Resource}s.
+ * @author Arjen Poutsma
+ */
+public class ResourceEncoder extends AbstractSingleValueEncoder<Resource> {
+
+	public static final int DEFAULT_BUFFER_SIZE = StreamUtils.BUFFER_SIZE;
+
+	private final int bufferSize;
+
+	public ResourceEncoder() {
+		this(DEFAULT_BUFFER_SIZE);
+	}
+
+	public ResourceEncoder(int bufferSize) {
+		super(MimeTypeUtils.ALL);
+		Assert.isTrue(bufferSize > 0, "'bufferSize' must be larger than 0");
+		this.bufferSize = bufferSize;
+	}
+
+	@Override
+	public boolean canEncode(ResolvableType elementType, MimeType mimeType, Object... hints) {
+		Class<?> clazz = elementType.getRawClass();
+		return (super.canEncode(elementType, mimeType, hints) &&
+				Resource.class.isAssignableFrom(clazz));
+	}
+
+	@Override
+	protected Flux<DataBuffer> encode(Resource resource,
+			DataBufferFactory dataBufferFactory,
+			ResolvableType type, MimeType mimeType, Object... hints) throws IOException {
+		InputStream is = resource.getInputStream();
+		return DataBufferUtils.read(is, dataBufferFactory, bufferSize);
+	}
+
+}

spring-core/src/main/java/org/springframework/core/codec/StringDecoder.java

@@ -0,0 +1,139 @@
+/*
+ * 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.codec;
+
+import java.nio.CharBuffer;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.IntPredicate;
+
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.core.ResolvableType;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.support.DataBufferUtils;
+import org.springframework.util.MimeType;
+import org.springframework.util.MimeTypeUtils;
+
+/**
+ * Decode from a bytes stream to a String stream.
+ *
+ * <p>By default, this decoder will split the received {@link DataBuffer}s along newline
+ * characters ({@code \r\n}), but this can be changed by passing {@code false} as
+ * constructor argument.
+ *
+ * @author Sebastien Deleuze
+ * @author Brian Clozel
+ * @author Arjen Poutsma
+ * @author Mark Paluch
+ * @see StringEncoder
+ */
+public class StringDecoder extends AbstractDecoder<String> {
+
+	public static final Charset DEFAULT_CHARSET = StandardCharsets.UTF_8;
+
+	private static final IntPredicate NEWLINE_DELIMITER = b -> b == '\n' || b == '\r';
+
+
+	private final boolean splitOnNewline;
+
+
+	/**
+	 * Create a {@code StringDecoder} that decodes a bytes stream to a String stream
+	 *
+	 * <p>By default, this decoder will split along new lines.
+	 */
+	public StringDecoder() {
+		this(true);
+	}
+
+	/**
+	 * Create a {@code StringDecoder} that decodes a bytes stream to a String stream
+	 *
+	 * @param splitOnNewline whether this decoder should split the received data buffers
+	 * along newline characters
+	 */
+	public StringDecoder(boolean splitOnNewline) {
+		super(new MimeType("text", "*", DEFAULT_CHARSET), MimeTypeUtils.ALL);
+		this.splitOnNewline = splitOnNewline;
+	}
+
+
+	@Override
+	public boolean canDecode(ResolvableType elementType, MimeType mimeType, Object... hints) {
+		return super.canDecode(elementType, mimeType, hints) &&
+				String.class.equals(elementType.getRawClass());
+	}
+
+	@Override
+	public Flux<String> decode(Publisher<DataBuffer> inputStream, ResolvableType elementType,
+			MimeType mimeType, Object... hints) {
+
+		Flux<DataBuffer> inputFlux = Flux.from(inputStream);
+		if (this.splitOnNewline) {
+			inputFlux = Flux.from(inputStream).flatMap(StringDecoder::splitOnNewline);
+		}
+		return inputFlux.map(buffer ->  decodeDataBuffer(buffer, mimeType));
+	}
+
+	@Override
+	public Mono<String> decodeToMono(Publisher<DataBuffer> inputStream, ResolvableType elementType,
+			MimeType mimeType, Object... hints) {
+
+		return Flux.from(inputStream)
+				.reduce(DataBuffer::write)
+				.map(buffer -> decodeDataBuffer(buffer, mimeType));
+	}
+
+	private static Flux<DataBuffer> splitOnNewline(DataBuffer dataBuffer) {
+		List<DataBuffer> results = new ArrayList<>();
+		int startIdx = 0;
+		int endIdx;
+		final int limit = dataBuffer.readableByteCount();
+		do {
+			endIdx = dataBuffer.indexOf(NEWLINE_DELIMITER, startIdx);
+			int length = endIdx != -1 ? endIdx - startIdx + 1 : limit - startIdx;
+			DataBuffer token = dataBuffer.slice(startIdx, length);
+			results.add(DataBufferUtils.retain(token));
+			startIdx = endIdx + 1;
+		}
+		while (startIdx < limit && endIdx != -1);
+		DataBufferUtils.release(dataBuffer);
+		return Flux.fromIterable(results);
+	}
+
+	private String decodeDataBuffer(DataBuffer dataBuffer, MimeType mimeType) {
+		Charset charset = getCharset(mimeType);
+		CharBuffer charBuffer = charset.decode(dataBuffer.asByteBuffer());
+		DataBufferUtils.release(dataBuffer);
+		return charBuffer.toString();
+	}
+
+	private Charset getCharset(MimeType mimeType) {
+		if (mimeType != null && mimeType.getCharset() != null) {
+			return mimeType.getCharset();
+		}
+		else {
+			return DEFAULT_CHARSET;
+		}
+	}
+
+}

spring-core/src/main/java/org/springframework/core/codec/StringEncoder.java

@@ -0,0 +1,70 @@
+/*
+ * 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.codec;
+
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+
+import org.springframework.core.ResolvableType;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferFactory;
+import org.springframework.util.MimeType;
+
+/**
+ * Encode from a String stream to a bytes stream.
+ *
+ * @author Sebastien Deleuze
+ * @see StringDecoder
+ */
+public class StringEncoder extends AbstractEncoder<String> {
+
+	public static final Charset DEFAULT_CHARSET = StandardCharsets.UTF_8;
+
+	public StringEncoder() {
+		super(new MimeType("text", "plain", DEFAULT_CHARSET));
+	}
+
+
+	@Override
+	public boolean canEncode(ResolvableType elementType, MimeType mimeType, Object... hints) {
+		Class<?> clazz = elementType.getRawClass();
+		return (super.canEncode(elementType, mimeType, hints) && String.class.equals(clazz));
+	}
+
+	@Override
+	public Flux<DataBuffer> encode(Publisher<? extends String> inputStream,
+			DataBufferFactory bufferFactory, ResolvableType elementType, MimeType mimeType,
+			Object... hints) {
+		Charset charset;
+		if (mimeType != null && mimeType.getCharset() != null) {
+			charset = mimeType.getCharset();
+		}
+		else {
+			 charset = DEFAULT_CHARSET;
+		}
+		return Flux.from(inputStream).map(s -> {
+			byte[] bytes = s.getBytes(charset);
+			DataBuffer dataBuffer = bufferFactory.allocateBuffer(bytes.length);
+			dataBuffer.write(bytes);
+			return dataBuffer;
+		});
+	}
+
+}

spring-core/src/main/java/org/springframework/core/codec/package-info.java

@@ -0,0 +1,6 @@
+/**
+ * Provides {@link org.springframework.core.codec.Encoder} and
+ * {@link org.springframework.core.codec.Decoder} abstractions for converting
+ * to and from between a stream of bytes and a stream of Java objects.
+ */
+package org.springframework.core.codec;

spring-core/src/main/java/org/springframework/core/convert/support/MonoToCompletableFutureConverter.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright 2002-2015 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.convert.support;
+
+import java.util.LinkedHashSet;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Mono;
+
+import org.springframework.core.convert.TypeDescriptor;
+import org.springframework.core.convert.converter.GenericConverter;
+
+/**
+ * @author Sebastien Deleuze
+ */
+public class MonoToCompletableFutureConverter implements GenericConverter {
+
+	@Override
+	public Set<GenericConverter.ConvertiblePair> getConvertibleTypes() {
+		Set<GenericConverter.ConvertiblePair> pairs = new LinkedHashSet<>(2);
+		pairs.add(new GenericConverter.ConvertiblePair(Mono.class, CompletableFuture.class));
+		pairs.add(new GenericConverter.ConvertiblePair(CompletableFuture.class, Mono.class));
+		return pairs;
+	}
+
+	@Override
+	public Object convert(Object source, TypeDescriptor sourceType, TypeDescriptor targetType) {
+		if (source == null) {
+			return null;
+		}
+		else if (CompletableFuture.class.isAssignableFrom(sourceType.getType())) {
+			return Mono.fromFuture((CompletableFuture<?>) source);
+		}
+		else if (CompletableFuture.class.isAssignableFrom(targetType.getType())) {
+			return Mono.from((Publisher<?>) source).toFuture();
+		}
+		return null;
+	}
+
+}

spring-core/src/main/java/org/springframework/core/convert/support/ReactorToRxJava1Converter.java

@@ -0,0 +1,79 @@
+/*
+ * Copyright 2002-2015 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.convert.support;
+
+import java.util.LinkedHashSet;
+import java.util.Set;
+
+import org.reactivestreams.Publisher;
+import reactor.core.converter.RxJava1CompletableConverter;
+import reactor.core.converter.RxJava1ObservableConverter;
+import reactor.core.converter.RxJava1SingleConverter;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import rx.Completable;
+import rx.Observable;
+import rx.Single;
+
+import org.springframework.core.convert.TypeDescriptor;
+import org.springframework.core.convert.converter.GenericConverter;
+
+/**
+ * @author Stephane Maldini
+ * @author Sebastien Deleuze
+ */
+public final class ReactorToRxJava1Converter implements GenericConverter {
+
+	@Override
+	public Set<GenericConverter.ConvertiblePair> getConvertibleTypes() {
+		Set<GenericConverter.ConvertiblePair> pairs = new LinkedHashSet<>(6);
+		pairs.add(new GenericConverter.ConvertiblePair(Flux.class, Observable.class));
+		pairs.add(new GenericConverter.ConvertiblePair(Observable.class, Flux.class));
+		pairs.add(new GenericConverter.ConvertiblePair(Mono.class, Single.class));
+		pairs.add(new GenericConverter.ConvertiblePair(Single.class, Mono.class));
+		pairs.add(new GenericConverter.ConvertiblePair(Mono.class, Completable.class));
+		pairs.add(new GenericConverter.ConvertiblePair(Completable.class, Mono.class));
+		return pairs;
+	}
+
+	@Override
+	public Object convert(Object source, TypeDescriptor sourceType, TypeDescriptor targetType) {
+		if (source == null) {
+			return null;
+		}
+		if (Observable.class.isAssignableFrom(sourceType.getType())) {
+			return RxJava1ObservableConverter.toPublisher((Observable<?>) source);
+		}
+		else if (Observable.class.isAssignableFrom(targetType.getType())) {
+			return RxJava1ObservableConverter.fromPublisher((Publisher<?>) source);
+		}
+		else if (Single.class.isAssignableFrom(sourceType.getType())) {
+			return RxJava1SingleConverter.toPublisher((Single<?>) source);
+		}
+		else if (Single.class.isAssignableFrom(targetType.getType())) {
+			return RxJava1SingleConverter.fromPublisher((Publisher<?>) source);
+		}
+		else if (Completable.class.isAssignableFrom(sourceType.getType())) {
+			return RxJava1CompletableConverter.toPublisher((Completable) source);
+		}
+		else if (Completable.class.isAssignableFrom(targetType.getType())) {
+			return RxJava1CompletableConverter.fromPublisher((Publisher<?>) source);
+		}
+		return null;
+	}
+
+}

spring-core/src/main/java/org/springframework/core/io/buffer/DataBuffer.java

@@ -0,0 +1,163 @@
+/*
+ * 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.function.IntPredicate;
+
+/**
+ * Basic abstraction over byte buffers.
+ *
+ * @author Arjen Poutsma
+ */
+public interface DataBuffer {
+
+	/**
+	 * Returns the {@link DataBufferFactory} that created this buffer.
+	 * @return the creating buffer factory
+	 */
+	DataBufferFactory factory();
+
+	/**
+	 * Returns the index of the first byte in this buffer that matches the given
+	 * predicate.
+	 * @param predicate the predicate to match
+	 * @param fromIndex the index to start the search from
+	 * @return the index of the first byte that matches {@code predicate}; or {@code -1}
+	 * if none match
+	 */
+	int indexOf(IntPredicate predicate, int fromIndex);
+
+	/**
+	 * Returns the index of the last byte in this buffer that matches the given
+	 * predicate.
+	 * @param predicate the predicate to match
+	 * @param fromIndex the index to start the search from
+	 * @return the index of the last byte that matches {@code predicate}; or {@code -1}
+	 * if none match
+	 */
+	int lastIndexOf(IntPredicate predicate, int fromIndex);
+
+	/**
+	 * 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 {@code DataBuffer}s 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);
+
+	/**
+	 * Creates a new {@code DataBuffer} whose contents is a shared subsequence of this
+	 * data buffer's content.  Data between this data buffer and the returned buffer is
+	 * shared; though changes in the returned buffer's position will not be reflected
+	 * in the reading nor writing position of this data buffer.
+	 * @param index the index at which to start the slice
+	 * @param length the length of the slice
+	 * @return the specified slice of this data buffer
+	 */
+	DataBuffer slice(int index, int length);
+
+	/**
+	 * 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 reading nor writing position 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-core/src/main/java/org/springframework/core/io/buffer/DataBufferFactory.java

@@ -0,0 +1,52 @@
+/*
+ * 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 and wrapping of data
+ * buffers.
+ *
+ * @author Arjen Poutsma
+ * @see DataBuffer
+ */
+public interface DataBufferFactory {
+
+	/**
+	 * 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);
+
+	/**
+	 * 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-core/src/main/java/org/springframework/core/io/buffer/DefaultDataBuffer.java

@@ -0,0 +1,361 @@
+/*
+ * 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 java.util.function.IntPredicate;
+
+import org.springframework.util.Assert;
+import org.springframework.util.ObjectUtils;
+
+/**
+ * Default implementation of the {@link DataBuffer} interface that uses a {@link
+ * ByteBuffer} internally, with separate read and write positions. Constructed
+ * using the {@link DefaultDataBufferFactory}.
+ *
+ * @author Arjen Poutsma
+ * @see DefaultDataBufferFactory
+ */
+public class DefaultDataBuffer implements DataBuffer {
+
+	private final DefaultDataBufferFactory dataBufferFactory;
+
+	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, DefaultDataBufferFactory dataBufferFactory) {
+		this(byteBuffer, byteBuffer.position(), byteBuffer.position(), dataBufferFactory);
+	}
+
+	DefaultDataBuffer(ByteBuffer byteBuffer, int readPosition, int writePosition,
+			DefaultDataBufferFactory dataBufferFactory) {
+		Assert.notNull(byteBuffer, "'byteBuffer' must not be null");
+		Assert.isTrue(readPosition >= 0, "'readPosition' must be 0 or higher");
+		Assert.isTrue(writePosition >= 0, "'writePosition' must be 0 or higher");
+		Assert.isTrue(readPosition <= writePosition,
+				"'readPosition' must be smaller than or equal to 'writePosition'");
+		Assert.notNull(dataBufferFactory, "'dataBufferFactory' must not be null");
+
+		this.byteBuffer = byteBuffer;
+		this.readPosition = readPosition;
+		this.writePosition = writePosition;
+		this.dataBufferFactory = dataBufferFactory;
+	}
+
+	@Override
+	public DefaultDataBufferFactory factory() {
+		return this.dataBufferFactory;
+	}
+
+	/**
+	 * 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 int indexOf(IntPredicate predicate, int fromIndex) {
+		Assert.notNull(predicate, "'predicate' must not be null");
+
+		if (fromIndex < 0) {
+			fromIndex = 0;
+		}
+		else if (fromIndex >= this.writePosition) {
+			return -1;
+		}
+		for (int i = fromIndex; i < this.writePosition; i++) {
+			byte b = this.byteBuffer.get(i);
+			if (predicate.test(b)) {
+				return i;
+			}
+		}
+		return -1;
+	}
+
+	@Override
+	public int lastIndexOf(IntPredicate predicate, int fromIndex) {
+		Assert.notNull(predicate, "'predicate' must not be null");
+		int i = Math.min(fromIndex, this.writePosition - 1);
+		for (; i >= 0; i--) {
+			byte b = this.byteBuffer.get(i);
+			if (predicate.test(b)) {
+				return i;
+			}
+		}
+		return -1;
+	}
+
+	@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);
+		try {
+			return function.apply(this.byteBuffer);
+		}
+		finally {
+			this.readPosition = this.byteBuffer.position();
+		}
+	}
+
+	@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} before and
+	 * after applying the given function on {@link #byteBuffer}.
+	 */
+	private <T> T writeInternal(Function<ByteBuffer, T> function) {
+		this.byteBuffer.position(this.writePosition);
+		try {
+			return function.apply(this.byteBuffer);
+		}
+		finally {
+			this.writePosition = this.byteBuffer.position();
+		}
+	}
+
+	@Override
+	public DataBuffer slice(int index, int length) {
+		int oldPosition = this.byteBuffer.position();
+		try {
+			this.byteBuffer.position(index);
+			ByteBuffer slice = this.byteBuffer.slice();
+			slice.limit(length);
+			return new SlicedDefaultDataBuffer(slice, 0, length, this.dataBufferFactory);
+		}
+		finally {
+			this.byteBuffer.position(oldPosition);
+		}
+
+	}
+
+	@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);
+		}
+	}
+
+	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.readPosition == other.readPosition &&
+					this.writePosition == other.writePosition &&
+					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 readableByteCount();
+		}
+
+		@Override
+		public int read() {
+			return readInternal(
+					buffer -> readableByteCount() > 0 ? buffer.get() & 0xFF : -1);
+		}
+
+		@Override
+		public int read(byte[] bytes, int off, int len) throws IOException {
+			return readInternal(buffer -> {
+				int count = readableByteCount();
+				if (count > 0) {
+					int minLen = Math.min(len, count);
+					buffer.get(bytes, off, minLen);
+					return minLen;
+				}
+				else {
+					return -1;
+				}
+			});
+		}
+	}
+
+	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));
+		}
+	}
+
+	private static class SlicedDefaultDataBuffer extends DefaultDataBuffer {
+
+		SlicedDefaultDataBuffer(ByteBuffer byteBuffer, int readPosition,
+				int writePosition, DefaultDataBufferFactory dataBufferFactory) {
+			super(byteBuffer, readPosition, writePosition, dataBufferFactory);
+		}
+
+		@Override
+		void grow(int minCapacity) {
+			throw new UnsupportedOperationException(
+					"Growing the capacity of a sliced buffer is not supported");
+		}
+	}
+}

spring-core/src/main/java/org/springframework/core/io/buffer/DefaultDataBufferFactory.java

@@ -0,0 +1,98 @@
+/*
+ * 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 org.springframework.util.Assert;
+
+/**
+ * Default implementation of the {@code DataBufferFactory} interface. Allows for
+ * specification of the default initial capacity at construction time, as well as whether
+ * heap-based or direct buffers are to be preferred.
+ *
+ * @author Arjen Poutsma
+ */
+public class DefaultDataBufferFactory implements DataBufferFactory {
+
+	/**
+	 * The default capacity when none is specified.
+	 * @see #DefaultDataBufferFactory()
+	 * @see #DefaultDataBufferFactory(boolean)
+	 */
+	public static final int DEFAULT_INITIAL_CAPACITY = 256;
+
+
+	private final boolean preferDirect;
+
+	private final int defaultInitialCapacity;
+
+	/**
+	 * Creates a new {@code DefaultDataBufferFactory} with default settings.
+	 */
+	public DefaultDataBufferFactory() {
+		this(false);
+	}
+
+	/**
+	 * Creates a new {@code DefaultDataBufferFactory}, indicating whether direct buffers
+	 * should be created by {@link #allocateBuffer()} and {@link #allocateBuffer(int)}.
+	 * @param preferDirect {@code true} if direct buffers are to be preferred; {@code
+	 * false} otherwise
+	 */
+	public DefaultDataBufferFactory(boolean preferDirect) {
+		this(preferDirect, DEFAULT_INITIAL_CAPACITY);
+	}
+
+	/**
+	 * Creates a new {@code DefaultDataBufferFactory}, indicating whether direct buffers
+	 * should be created by {@link #allocateBuffer()} and {@link #allocateBuffer(int)},
+	 * and what the capacity is to be used for {@link #allocateBuffer()}.
+	 * @param preferDirect {@code true} if direct buffers are to be preferred; {@code
+	 * false} otherwise
+	 */
+	public DefaultDataBufferFactory(boolean preferDirect, int defaultInitialCapacity) {
+		Assert.isTrue(defaultInitialCapacity > 0,
+				"'defaultInitialCapacity' should be larger than 0");
+		this.preferDirect = preferDirect;
+		this.defaultInitialCapacity = defaultInitialCapacity;
+	}
+
+	@Override
+	public DefaultDataBuffer allocateBuffer() {
+		return allocateBuffer(this.defaultInitialCapacity);
+	}
+
+	@Override
+	public DefaultDataBuffer allocateBuffer(int initialCapacity) {
+		return this.preferDirect ?
+				new DefaultDataBuffer(ByteBuffer.allocateDirect(initialCapacity), this) :
+				new DefaultDataBuffer(ByteBuffer.allocate(initialCapacity), this);
+	}
+
+	@Override
+	public DefaultDataBuffer wrap(ByteBuffer byteBuffer) {
+		ByteBuffer sliced = byteBuffer.slice();
+		return new DefaultDataBuffer(sliced, 0, byteBuffer.remaining(), this);
+	}
+
+	@Override
+	public String toString() {
+		return "DefaultDataBufferFactory - preferDirect: " + this.preferDirect;
+	}
+
+}

spring-core/src/main/java/org/springframework/core/io/buffer/FlushingDataBuffer.java

@@ -0,0 +1,124 @@
+/*
+ * 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.function.IntPredicate;
+
+/**
+ * Empty {@link DataBuffer} that indicates to the file or the socket writing it that
+ * previously buffered data should be flushed.
+ *
+ * @author Sebastien Deleuze
+ * @see FlushingDataBuffer#INSTANCE
+ */
+public class FlushingDataBuffer implements DataBuffer {
+
+	/** Singleton instance of this class */
+	public static final FlushingDataBuffer INSTANCE = new FlushingDataBuffer();
+
+	private final DataBuffer buffer;
+
+
+	private FlushingDataBuffer() {
+		this.buffer = new DefaultDataBufferFactory().allocateBuffer(0);
+	}
+
+
+	@Override
+	public DataBufferFactory factory() {
+		return this.buffer.factory();
+	}
+
+	@Override
+	public int indexOf(IntPredicate predicate, int fromIndex) {
+		return this.buffer.indexOf(predicate, fromIndex);
+	}
+
+	@Override
+	public int lastIndexOf(IntPredicate predicate, int fromIndex) {
+		return this.buffer.lastIndexOf(predicate, fromIndex);
+	}
+
+	@Override
+	public int readableByteCount() {
+		return this.buffer.readableByteCount();
+	}
+
+	@Override
+	public byte read() {
+		return this.buffer.read();
+	}
+
+	@Override
+	public DataBuffer read(byte[] destination) {
+		return this.buffer.read(destination);
+	}
+
+	@Override
+	public DataBuffer read(byte[] destination, int offset, int length) {
+		return this.buffer.read(destination, offset, length);
+	}
+
+	@Override
+	public DataBuffer write(byte b) {
+		return this.buffer.write(b);
+	}
+
+	@Override
+	public DataBuffer write(byte[] source) {
+		return this.buffer.write(source);
+	}
+
+	@Override
+	public DataBuffer write(byte[] source, int offset, int length) {
+		return this.buffer.write(source, offset, length);
+	}
+
+	@Override
+	public DataBuffer write(DataBuffer... buffers) {
+		return this.buffer.write(buffers);
+	}
+
+	@Override
+	public DataBuffer write(ByteBuffer... buffers) {
+		return this.buffer.write(buffers);
+	}
+
+	@Override
+	public DataBuffer slice(int index, int length) {
+		return this.buffer.slice(index, length);
+	}
+
+	@Override
+	public ByteBuffer asByteBuffer() {
+		return this.buffer.asByteBuffer();
+	}
+
+	@Override
+	public InputStream asInputStream() {
+		return this.buffer.asInputStream();
+	}
+
+	@Override
+	public OutputStream asOutputStream() {
+		return this.buffer.asOutputStream();
+	}
+
+}

spring-core/src/main/java/org/springframework/core/io/buffer/NettyDataBuffer.java

@@ -0,0 +1,242 @@
+/*
+ * 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 java.util.function.IntPredicate;
+
+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 NettyDataBufferFactory}.
+ *
+ * @author Arjen Poutsma
+ */
+public class NettyDataBuffer implements PooledDataBuffer {
+
+	private final NettyDataBufferFactory dataBufferFactory;
+
+	private ByteBuf byteBuf;
+
+	/**
+	 * Creates a new {@code NettyDataBuffer} based on the given {@code ByteBuff}.
+	 * @param byteBuf the buffer to base this buffer on
+	 */
+	NettyDataBuffer(ByteBuf byteBuf, NettyDataBufferFactory dataBufferFactory) {
+		Assert.notNull(byteBuf, "'byteBuf' must not be null");
+		Assert.notNull(dataBufferFactory, "'dataBufferFactory' must not be null");
+
+		this.byteBuf = byteBuf;
+		this.dataBufferFactory = dataBufferFactory;
+	}
+
+	@Override
+	public NettyDataBufferFactory factory() {
+		return this.dataBufferFactory;
+	}
+
+	/**
+	 * 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 int indexOf(IntPredicate predicate, int fromIndex) {
+		Assert.notNull(predicate, "'predicate' must not be null");
+		if (fromIndex < 0) {
+			fromIndex = 0;
+		}
+		else if (fromIndex >= this.byteBuf.writerIndex()) {
+			return -1;
+		}
+		int length = this.byteBuf.writerIndex() - fromIndex;
+
+		return this.byteBuf.forEachByte(fromIndex, length, predicate.negate()::test);
+	}
+
+	@Override
+	public int lastIndexOf(IntPredicate predicate, int fromIndex) {
+		Assert.notNull(predicate, "'predicate' must not be null");
+		if (fromIndex < 0) {
+			return -1;
+		}
+		fromIndex = Math.min(fromIndex, this.byteBuf.writerIndex() - 1);
+
+		return this.byteBuf.forEachByteDesc(0, fromIndex, predicate.negate()::test);
+	}
+
+	@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) {
+				ByteBuf[] nativeBuffers = Arrays.stream(buffers)
+						.map(b -> ((NettyDataBuffer) b).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);
+		composite.addComponents(byteBufs);
+
+		int writerIndex = this.byteBuf.readableBytes() +
+				Arrays.stream(byteBufs).mapToInt(ByteBuf::readableBytes).sum();
+		composite.writerIndex(writerIndex);
+
+		this.byteBuf = composite;
+
+		return this;
+	}
+
+	@Override
+	public DataBuffer slice(int index, int length) {
+		ByteBuf slice = this.byteBuf.slice(index, length);
+		return new NettyDataBuffer(slice, this.dataBufferFactory);
+	}
+
+	@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 PooledDataBuffer retain() {
+		return new NettyDataBuffer(this.byteBuf.retain(), dataBufferFactory);
+	}
+
+	@Override
+	public boolean release() {
+		return this.byteBuf.release();
+	}
+
+	@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-core/src/main/java/org/springframework/core/io/buffer/NettyDataBufferFactory.java

@@ -0,0 +1,82 @@
+/*
+ * 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;
+
+/**
+ * Implementation of the {@code DataBufferFactory} interface based on a Netty
+ * {@link ByteBufAllocator}.
+ *
+ * @author Arjen Poutsma
+ * @see io.netty.buffer.PooledByteBufAllocator
+ * @see io.netty.buffer.UnpooledByteBufAllocator
+ */
+public class NettyDataBufferFactory implements DataBufferFactory {
+
+	private final ByteBufAllocator byteBufAllocator;
+
+	/**
+	 * Creates a new {@code NettyDataBufferFactory} based on the given factory.
+	 * @param byteBufAllocator the factory to use
+	 * @see io.netty.buffer.PooledByteBufAllocator
+	 * @see io.netty.buffer.UnpooledByteBufAllocator
+	 */
+	public NettyDataBufferFactory(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, this);
+	}
+
+	@Override
+	public NettyDataBuffer allocateBuffer(int initialCapacity) {
+		ByteBuf byteBuf = this.byteBufAllocator.buffer(initialCapacity);
+		return new NettyDataBuffer(byteBuf, this);
+	}
+
+	@Override
+	public NettyDataBuffer wrap(ByteBuffer byteBuffer) {
+		ByteBuf byteBuf = Unpooled.wrappedBuffer(byteBuffer);
+		return new NettyDataBuffer(byteBuf, this);
+	}
+
+	/**
+	 * 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, this);
+	}
+
+	@Override
+	public String toString() {
+		return "NettyDataBufferFactory (" + this.byteBufAllocator + ")";
+	}
+}

spring-core/src/main/java/org/springframework/core/io/buffer/PooledDataBuffer.java

@@ -0,0 +1,40 @@
+/*
+ * 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;
+
+/**
+ * Extension of {@link DataBuffer} that allows for buffer that share a memory pool.
+ * Introduces methods for reference counting.
+ *
+ * @author Arjen Poutsma
+ */
+public interface PooledDataBuffer extends DataBuffer {
+
+	/**
+	 * Increases the reference count for this buffer by one.
+	 * @return this buffer
+	 */
+	PooledDataBuffer retain();
+
+	/**
+	 * Decreases the reference count for this buffer by one, and releases it once the
+	 * count reaches zero.
+	 * @return {@code true} if the buffer was released; {@code false} otherwise.
+	 */
+	boolean release();
+
+}

spring-core/src/main/java/org/springframework/core/io/buffer/support/DataBufferUtils.java

@@ -0,0 +1,196 @@
+/*
+ * 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.nio.ByteBuffer;
+import java.nio.channels.Channels;
+import java.nio.channels.ReadableByteChannel;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BiFunction;
+import java.util.function.Consumer;
+
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.subscriber.SignalEmitter;
+
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferFactory;
+import org.springframework.core.io.buffer.PooledDataBuffer;
+import org.springframework.util.Assert;
+
+/**i
+ * Utility class for working with {@link DataBuffer}s.
+ *
+ * @author Arjen Poutsma
+ */
+public abstract class DataBufferUtils {
+
+	private static final Consumer<ReadableByteChannel> CLOSE_CONSUMER = channel -> {
+		try {
+			if (channel != null) {
+				channel.close();
+			}
+		}
+		catch (IOException ignored) {
+		}
+	};
+
+	/**
+	 * Reads the given {@code InputStream} into a {@code Flux} of
+	 * {@code DataBuffer}s. Closes the input stream when the flux is terminated.
+	 * @param inputStream the input stream to read from
+	 * @param dataBufferFactory the factory to create data buffers with
+	 * @param bufferSize the maximum size of the data buffers
+	 * @return a flux of data buffers read from the given channel
+	 */
+	public static Flux<DataBuffer> read(InputStream inputStream,
+			DataBufferFactory dataBufferFactory, int bufferSize) {
+		Assert.notNull(inputStream, "'inputStream' must not be null");
+		Assert.notNull(dataBufferFactory, "'dataBufferFactory' must not be null");
+
+		ReadableByteChannel channel = Channels.newChannel(inputStream);
+		return read(channel, dataBufferFactory, bufferSize);
+	}
+
+	/**
+	 * Reads the given {@code ReadableByteChannel} into a {@code Flux} of
+	 * {@code DataBuffer}s. Closes the channel when the flux is terminated.
+	 * @param channel the channel to read from
+	 * @param dataBufferFactory the factory to create data buffers with
+	 * @param bufferSize the maximum size of the data buffers
+	 * @return a flux of data buffers read from the given channel
+	 */
+	public static Flux<DataBuffer> read(ReadableByteChannel channel,
+			DataBufferFactory dataBufferFactory, int bufferSize) {
+		Assert.notNull(channel, "'channel' must not be null");
+		Assert.notNull(dataBufferFactory, "'dataBufferFactory' must not be null");
+
+		return Flux.generate(() -> channel,
+				new ReadableByteChannelGenerator(dataBufferFactory, bufferSize),
+				CLOSE_CONSUMER);
+	}
+
+	/**
+	 * Relays buffers from the given {@link Publisher} until the total
+	 * {@linkplain DataBuffer#readableByteCount() byte count} reaches the given maximum
+	 * byte count, or until the publisher is complete.
+	 * @param publisher the publisher to filter
+	 * @param maxByteCount the maximum byte count
+	 * @return a flux whose maximum byte count is {@code maxByteCount}
+	 */
+	public static Flux<DataBuffer> takeUntilByteCount(Publisher<DataBuffer> publisher,
+			long maxByteCount) {
+		Assert.notNull(publisher, "'publisher' must not be null");
+		Assert.isTrue(maxByteCount >= 0, "'maxByteCount' must be a positive number");
+
+		AtomicLong byteCountDown = new AtomicLong(maxByteCount);
+
+		return Flux.from(publisher).
+				takeWhile(dataBuffer -> {
+					int delta = -dataBuffer.readableByteCount();
+					long currentCount = byteCountDown.getAndAdd(delta);
+					return currentCount >= 0;
+				}).
+				map(dataBuffer -> {
+					long currentCount = byteCountDown.get();
+					if (currentCount >= 0) {
+						return dataBuffer;
+					}
+					else {
+						// last buffer
+						int size = (int) (currentCount + dataBuffer.readableByteCount());
+						return dataBuffer.slice(0, size);
+					}
+				});
+	}
+
+	/**
+	 * Retains the given data buffer, it it is a {@link PooledDataBuffer}.
+	 * @param dataBuffer the data buffer to retain
+	 * @return the retained buffer
+	 */
+	@SuppressWarnings("unchecked")
+	public static <T extends DataBuffer> T retain(T dataBuffer) {
+		if (dataBuffer instanceof PooledDataBuffer) {
+			return (T) ((PooledDataBuffer) dataBuffer).retain();
+		}
+		else {
+			return dataBuffer;
+		}
+	}
+
+	/**
+	 * Releases the given data buffer, if it is a {@link PooledDataBuffer}.
+	 * @param dataBuffer the data buffer to release
+	 * @return {@code true} if the buffer was released; {@code false} otherwise.
+	 */
+	public static boolean release(DataBuffer dataBuffer) {
+		if (dataBuffer instanceof PooledDataBuffer) {
+			return ((PooledDataBuffer) dataBuffer).release();
+		}
+		return false;
+	}
+
+	private static class ReadableByteChannelGenerator
+			implements BiFunction<ReadableByteChannel, SignalEmitter<DataBuffer>,
+						ReadableByteChannel> {
+
+		private final DataBufferFactory dataBufferFactory;
+
+		private final int chunkSize;
+
+		public ReadableByteChannelGenerator(DataBufferFactory dataBufferFactory,
+				int chunkSize) {
+			this.dataBufferFactory = dataBufferFactory;
+			this.chunkSize = chunkSize;
+		}
+
+		@Override
+		public ReadableByteChannel apply(ReadableByteChannel
+				channel, SignalEmitter<DataBuffer>	sub) {
+			try {
+				ByteBuffer byteBuffer = ByteBuffer.allocate(chunkSize);
+				int read;
+				if ((read = channel.read(byteBuffer)) > 0) {
+					byteBuffer.flip();
+					boolean release = true;
+					DataBuffer dataBuffer = this.dataBufferFactory.allocateBuffer(read);
+					try {
+						dataBuffer.write(byteBuffer);
+						release = false;
+						sub.next(dataBuffer);
+					}
+					finally {
+						if (release) {
+							release(dataBuffer);
+						}
+					}
+				}
+				else {
+					sub.complete();
+				}
+			}
+			catch (IOException ex) {
+				sub.fail(ex);
+			}
+			return channel;
+		}
+	}
+
+}

spring-core/src/main/java/org/springframework/util/MimeTypeUtils.java

@@ -16,6 +16,8 @@
 
 package org.springframework.util;
 
+import java.io.IOException;
+import java.io.InputStream;
 import java.nio.charset.Charset;
 import java.nio.charset.UnsupportedCharsetException;
 import java.util.ArrayList;
@@ -26,8 +28,13 @@ import java.util.Iterator;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Random;
+import javax.activation.FileTypeMap;
+import javax.activation.MimetypesFileTypeMap;
 
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.core.io.Resource;
 import org.springframework.util.MimeType.SpecificityComparator;
 
 /**
@@ -49,6 +56,11 @@ public abstract class MimeTypeUtils {
 
 	private static Charset US_ASCII = Charset.forName("US-ASCII");
 
+	private static final FileTypeMap fileTypeMap;
+
+	static {
+		fileTypeMap = initFileTypeMap();
+	}
 
 	/**
 	 * Public constant mime type that includes all media ranges (i.e. "*/*").
@@ -208,6 +220,31 @@ public abstract class MimeTypeUtils {
 		TEXT_XML = MimeType.valueOf(TEXT_XML_VALUE);
 	}
 
+	private static FileTypeMap initFileTypeMap() {
+		// See if we can find the extended mime.types from the context-support module...
+		Resource mappingLocation = new ClassPathResource("org/springframework/mail/javamail/mime.types");
+		if (mappingLocation.exists()) {
+			InputStream inputStream = null;
+			try {
+				inputStream = mappingLocation.getInputStream();
+				return new MimetypesFileTypeMap(inputStream);
+			}
+			catch (IOException ex) {
+				// ignore
+			}
+			finally {
+				if (inputStream != null) {
+					try {
+						inputStream.close();
+					}
+					catch (IOException ex) {
+						// ignore
+					}
+				}
+			}
+		}
+		return FileTypeMap.getDefaultFileTypeMap();
+	}
 
 	/**
 	 * Parse the given String into a single {@code MimeType}.
@@ -285,6 +322,22 @@ public abstract class MimeTypeUtils {
 		return result;
 	}
 
+	/**
+	 * Returns the {@code MimeType} of the given file name, using the Java Activation
+	 * Framework.
+	 * @param filename the filename whose mime type is to be found
+	 * @return the mime type, if any
+	 */
+	public static Optional<MimeType> getMimeType(String filename) {
+		if (filename != null) {
+			String mimeType = fileTypeMap.getContentType(filename);
+			if (StringUtils.hasText(mimeType)) {
+				return Optional.of(parseMimeType(mimeType));
+			}
+		}
+		return Optional.empty();
+	}
+
 	/**
 	 * Return a string representation of the given list of {@code MimeType} objects.
 	 * @param mimeTypes the string to parse

spring-core/src/main/java/org/springframework/util/ResourceUtils.java

@@ -18,12 +18,18 @@ package org.springframework.util;
 
 import java.io.File;
 import java.io.FileNotFoundException;
+import java.io.IOException;
 import java.net.MalformedURLException;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.net.URL;
 import java.net.URLConnection;
 
+import org.springframework.core.io.ByteArrayResource;
+import org.springframework.core.io.DescriptiveResource;
+import org.springframework.core.io.InputStreamResource;
+import org.springframework.core.io.Resource;
+
 /**
  * Utility methods for resolving resource locations to files in the
  * file system. Mainly for internal use within the framework.
@@ -296,6 +302,32 @@ public abstract class ResourceUtils {
 				url.getPath().toLowerCase().endsWith(JAR_FILE_EXTENSION));
 	}
 
+	/**
+	 * Indicates whether the given resource has a file, so that {@link
+	 * Resource#getFile()}
+	 * can be called without an {@link java.io.IOException}.
+	 * @param resource the resource to check
+	 * @return {@code true} if the given resource has a file; {@code false} otherwise
+	 * @since 5.0
+	 */
+	public static boolean hasFile(Resource resource) {
+		Assert.notNull(resource, "'resource' must not be null");
+
+		// the following Resource implementations do not support getURI/getFile
+		if (resource instanceof ByteArrayResource ||
+				resource instanceof DescriptiveResource ||
+				resource instanceof InputStreamResource) {
+			return false;
+		}
+		try {
+			URI resourceUri = resource.getURI();
+			return URL_PROTOCOL_FILE.equals(resourceUri.getScheme());
+		}
+		catch (IOException ignored) {
+		}
+		return false;
+	}
+
 	/**
 	 * Extract the URL for the actual jar file from the given URL
 	 * (which may point to a resource in a jar file or to a jar file itself).

spring-core/src/main/java/org/springframework/util/xml/AbstractXMLEventReader.java

@@ -0,0 +1,139 @@
+/*
+ * 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.util.xml;
+
+import java.util.NoSuchElementException;
+import javax.xml.stream.XMLEventReader;
+import javax.xml.stream.XMLStreamConstants;
+import javax.xml.stream.XMLStreamException;
+import javax.xml.stream.events.Characters;
+import javax.xml.stream.events.XMLEvent;
+
+import org.springframework.util.ClassUtils;
+
+/**
+ * Abstract base class for {@code XMLEventReader}s.
+ * @author Arjen Poutsma
+ */
+abstract class AbstractXMLEventReader implements XMLEventReader {
+
+	private boolean closed;
+
+	@Override
+	public Object next() {
+		try {
+			return nextEvent();
+		}
+		catch (XMLStreamException ex) {
+			throw new NoSuchElementException();
+		}
+	}
+
+	@Override
+	public void remove() {
+		throw new UnsupportedOperationException(
+				"remove not supported on " + ClassUtils.getShortName(getClass()));
+	}
+
+	@Override
+	public String getElementText() throws XMLStreamException {
+		checkIfClosed();
+		if (!peek().isStartElement()) {
+			throw new XMLStreamException("Not at START_ELEMENT");
+		}
+
+		StringBuilder builder = new StringBuilder();
+		while (true) {
+			XMLEvent event = nextEvent();
+			if (event.isEndElement()) {
+				break;
+			}
+			else if (!event.isCharacters()) {
+				throw new XMLStreamException(
+						"Unexpected event [" + event + "] in getElementText()");
+			}
+			Characters characters = event.asCharacters();
+			if (!characters.isIgnorableWhiteSpace()) {
+				builder.append(event.asCharacters().getData());
+			}
+		}
+		return builder.toString();
+	}
+
+	@Override
+	public XMLEvent nextTag() throws XMLStreamException {
+		checkIfClosed();
+		while (true) {
+			XMLEvent event = nextEvent();
+			switch (event.getEventType()) {
+				case XMLStreamConstants.START_ELEMENT:
+				case XMLStreamConstants.END_ELEMENT:
+					return event;
+				case XMLStreamConstants.END_DOCUMENT:
+					return null;
+				case XMLStreamConstants.SPACE:
+				case XMLStreamConstants.COMMENT:
+				case XMLStreamConstants.PROCESSING_INSTRUCTION:
+					continue;
+				case XMLStreamConstants.CDATA:
+				case XMLStreamConstants.CHARACTERS:
+					if (!event.asCharacters().isWhiteSpace()) {
+						throw new XMLStreamException(
+								"Non-ignorable whitespace CDATA or CHARACTERS event in nextTag()");
+					}
+					break;
+				default:
+					throw new XMLStreamException("Received event [" + event +
+							"], instead of START_ELEMENT or END_ELEMENT.");
+			}
+		}
+	}
+
+	/**
+	 * Throws an {@code IllegalArgumentException} when called.
+	 * @throws IllegalArgumentException when called.
+	 */
+	@Override
+	public Object getProperty(String name) throws IllegalArgumentException {
+		throw new IllegalArgumentException("Property not supported: [" + name + "]");
+	}
+
+	/**
+	 * Returns {@code true} if closed; {@code false} otherwise.
+	 * @see #close()
+	 */
+	protected boolean isClosed() {
+		return closed;
+	}
+
+	/**
+	 * Checks if the reader is closed, and throws a {@code XMLStreamException} if so.
+	 * @throws XMLStreamException if the reader is closed
+	 * @see #close()
+	 * @see #isClosed()
+	 */
+	protected void checkIfClosed() throws XMLStreamException {
+		if (isClosed()) {
+			throw new XMLStreamException("XMLEventReader has been closed");
+		}
+	}
+
+	@Override
+	public void close() {
+		closed = true;
+	}
+}

spring-core/src/main/java/org/springframework/util/xml/ListBasedXMLEventReader.java

@@ -0,0 +1,72 @@
+/*
+ * 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.util.xml;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.NoSuchElementException;
+import javax.xml.stream.events.XMLEvent;
+
+import org.springframework.util.Assert;
+
+/**
+ * Implementation of {@code XMLEventReader} based on a list of {@link XMLEvent}s.
+ *
+ * @author Arjen Poutsma
+ */
+class ListBasedXMLEventReader extends AbstractXMLEventReader {
+
+	private final List<XMLEvent> events;
+
+	private int cursor = 0;
+
+	public ListBasedXMLEventReader(List<XMLEvent> events) {
+		Assert.notNull(events, "'events' must not be null");
+		this.events = Collections.unmodifiableList(events);
+	}
+
+	@Override
+	public boolean hasNext() {
+		return cursor != events.size();
+	}
+
+	@Override
+	public XMLEvent nextEvent() {
+		if (cursor < events.size()) {
+			return events.get(cursor++);
+		}
+		else {
+			throw new NoSuchElementException();
+		}
+	}
+
+	@Override
+	public XMLEvent peek() {
+		if (cursor < events.size()) {
+			return events.get(cursor);
+		}
+		else {
+			return null;
+		}
+	}
+
+	@Override
+	public void close() {
+		super.close();
+		this.events.clear();
+	}
+}

spring-core/src/main/java/org/springframework/util/xml/StaxUtils.java

@@ -16,12 +16,14 @@
 
 package org.springframework.util.xml;
 
+import java.util.List;
 import javax.xml.stream.XMLEventFactory;
 import javax.xml.stream.XMLEventReader;
 import javax.xml.stream.XMLEventWriter;
 import javax.xml.stream.XMLStreamException;
 import javax.xml.stream.XMLStreamReader;
 import javax.xml.stream.XMLStreamWriter;
+import javax.xml.stream.events.XMLEvent;
 import javax.xml.transform.Result;
 import javax.xml.transform.Source;
 import javax.xml.transform.stax.StAXResult;
@@ -211,6 +213,16 @@ public abstract class StaxUtils {
 		}
 	}
 
+	/**
+	 * Create a {@link XMLEventReader} from the given list of {@link XMLEvent}.
+	 * @param events the list of {@link XMLEvent}s.
+	 * @return an {@code XMLEventReader} that reads from the given events
+	 * @since 5.0
+	 */
+	public static XMLEventReader createXMLEventReader(List<XMLEvent> events) {
+		return new ListBasedXMLEventReader(events);
+	}
+
 	/**
 	 * Create a SAX {@link ContentHandler} that writes to the given StAX {@link XMLStreamWriter}.
 	 * @param streamWriter the StAX stream writer