Merge pull request #97 from poutsma/zero_copy_response

@ResponseBody Resource support

spring-web-reactive/src/main/java/org/springframework/http/ZeroCopyHttpOutputMessage.java

@@ -0,0 +1,42 @@
+/*
+ * 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.http;
+
+import java.io.File;
+
+import reactor.core.publisher.Mono;
+
+/**
+ * Sub-interface of {@code ReactiveOutputMessage} that has support for "zero-copy"
+ * file transfers.
+ *
+ * @author Arjen Poutsma
+ * @see <a href="https://en.wikipedia.org/wiki/Zero-copy">Zero-copy</a>
+ */
+public interface ZeroCopyHttpOutputMessage extends ReactiveHttpOutputMessage {
+
+	/**
+	 * Set the body of the message to the given {@link File} which will be
+	 * used to write to the underlying HTTP layer.
+	 * @param file the file to transfer
+	 * @param position the position within the file from which the transfer is to begin
+	 * @param count the number of bytes to be transferred
+	 * @return a publisher that indicates completion or error.
+	 */
+	Mono<Void> setBody(File file, long position, long count);
+
+}

spring-web-reactive/src/main/java/org/springframework/http/converter/reactive/CodecHttpMessageConverter.java

@@ -0,0 +1,102 @@
+/*
+ * 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.http.converter.reactive;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.core.ResolvableType;
+import org.springframework.core.codec.Decoder;
+import org.springframework.core.codec.Encoder;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferAllocator;
+import org.springframework.http.MediaType;
+import org.springframework.http.ReactiveHttpInputMessage;
+import org.springframework.http.ReactiveHttpOutputMessage;
+import org.springframework.http.support.MediaTypeUtils;
+
+/**
+ * @author Arjen Poutsma
+ */
+public class CodecHttpMessageConverter<T> implements HttpMessageConverter<T> {
+
+	private final Encoder<T> encoder;
+
+	private final Decoder<T> decoder;
+
+	public CodecHttpMessageConverter(Encoder<T> encoder, Decoder<T> decoder) {
+		this.encoder = encoder;
+		this.decoder = decoder;
+	}
+
+	@Override
+	public boolean canRead(ResolvableType type, MediaType mediaType) {
+		return this.decoder != null && this.decoder.canDecode(type, mediaType);
+	}
+
+	@Override
+	public boolean canWrite(ResolvableType type, MediaType mediaType) {
+		return this.encoder != null && this.encoder.canEncode(type, mediaType);
+	}
+
+	@Override
+	public List<MediaType> getReadableMediaTypes() {
+		return this.decoder != null ? this.decoder.getSupportedMimeTypes().stream().
+				map(MediaTypeUtils::toMediaType).
+				collect(Collectors.toList()) : Collections.emptyList();
+	}
+
+	@Override
+	public List<MediaType> getWritableMediaTypes() {
+		return this.encoder != null ? this.encoder.getSupportedMimeTypes().stream().
+				map(MediaTypeUtils::toMediaType).
+				collect(Collectors.toList()) : Collections.emptyList();
+	}
+
+	@Override
+	public Flux<T> read(ResolvableType type, ReactiveHttpInputMessage inputMessage) {
+		if (this.decoder == null) {
+			return Flux.error(new IllegalStateException("No decoder set"));
+		}
+		MediaType contentType = inputMessage.getHeaders().getContentType();
+		if (contentType == null) {
+			contentType = MediaType.APPLICATION_OCTET_STREAM;
+		}
+
+		Flux<DataBuffer> body = inputMessage.getBody();
+
+		return this.decoder.decode(body, type, contentType);
+	}
+
+	@Override
+	public Mono<Void> write(Publisher<? extends T> inputStream, ResolvableType type,
+			MediaType contentType,
+			ReactiveHttpOutputMessage outputMessage) {
+		if (this.encoder == null) {
+			return Mono.error(new IllegalStateException("No decoder set"));
+		}
+		outputMessage.getHeaders().setContentType(contentType);
+		DataBufferAllocator allocator = outputMessage.allocator();
+		Flux<DataBuffer> body = encoder.encode(inputStream, allocator, type, contentType);
+		return outputMessage.setBody(body);
+	}
+}

spring-web-reactive/src/main/java/org/springframework/http/converter/reactive/HttpMessageConverter.java

@@ -0,0 +1,91 @@
+/*
+ * 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.http.converter.reactive;
+
+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.http.MediaType;
+import org.springframework.http.ReactiveHttpInputMessage;
+import org.springframework.http.ReactiveHttpOutputMessage;
+
+/**
+ * Strategy interface that specifies a converter that can convert from and to HTTP
+ * requests and responses.
+ * @author Arjen Poutsma
+ */
+public interface HttpMessageConverter<T> {
+
+	/**
+	 * Indicates whether the given class can be read by this converter.
+	 * @param type the type to test for readability
+	 * @param mediaType the media type to read, can be {@code null} if not specified.
+	 * Typically the value of a {@code Content-Type} header.
+	 * @return {@code true} if readable; {@code false} otherwise
+	 */
+	boolean canRead(ResolvableType type, MediaType mediaType);
+
+	/**
+	 * Return the list of {@link MediaType} objects that can be read by this converter.
+	 * @return the list of supported readable media types
+	 */
+	List<MediaType> getReadableMediaTypes();
+
+	/**
+	 * Read an object of the given type form the given input message, and returns it.
+	 * @param type the type of object to return. This type must have previously been
+	 * passed to the
+	 * {@link #canRead canRead} method of this interface, which must have returned {@code
+	 * true}.
+	 * @param inputMessage the HTTP input message to read from
+	 * @return the converted object
+	 */
+	Flux<T> read(ResolvableType type, ReactiveHttpInputMessage inputMessage);
+
+	/**
+	 * Indicates whether the given class can be written by this converter.
+	 * @param type the class to test for writability
+	 * @param mediaType the media type to write, can be {@code null} if not specified.
+	 * Typically the value of an {@code Accept} header.
+	 * @return {@code true} if writable; {@code false} otherwise
+	 */
+	boolean canWrite(ResolvableType type, MediaType mediaType);
+
+	/**
+	 * Return the list of {@link MediaType} objects that can be written by this
+	 * converter.
+	 * @return the list of supported readable media types
+	 */
+	List<MediaType> getWritableMediaTypes();
+
+	/**
+	 * Write an given object to the given output message.
+	 * @param inputStream the input stream to write
+	 * @param type the stream element type to process.
+	 * @param contentType the content type to use when writing. May be {@code null} to
+	 * indicate that the default content type of the converter must be used.
+	 * @param outputMessage the message to write to
+	 * @return
+	 */
+	Mono<Void> write(Publisher<? extends T> inputStream,
+			ResolvableType type, MediaType contentType,
+			ReactiveHttpOutputMessage outputMessage);
+}

spring-web-reactive/src/main/java/org/springframework/http/converter/reactive/ResourceHttpMessageConverter.java

@@ -0,0 +1,231 @@
+/*
+ * 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.http.converter.reactive;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.URI;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+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.DescriptiveResource;
+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.http.HttpHeaders;
+import org.springframework.http.HttpRangeResource;
+import org.springframework.http.MediaType;
+import org.springframework.http.ReactiveHttpInputMessage;
+import org.springframework.http.ReactiveHttpOutputMessage;
+import org.springframework.http.ZeroCopyHttpOutputMessage;
+import org.springframework.http.support.MediaTypeUtils;
+import org.springframework.util.MimeTypeUtils2;
+import org.springframework.util.ResourceUtils;
+import org.springframework.util.StreamUtils;
+
+/**
+ * @author Arjen Poutsma
+ */
+public class ResourceHttpMessageConverter implements HttpMessageConverter<Resource> {
+
+	private static final int BUFFER_SIZE = StreamUtils.BUFFER_SIZE;
+
+	private static final List<MediaType> SUPPORTED_MEDIA_TYPES =
+			Collections.singletonList(MediaType.ALL);
+
+	@Override
+	public boolean canRead(ResolvableType type, MediaType mediaType) {
+		return Resource.class.isAssignableFrom(type.getRawClass());
+	}
+
+	@Override
+	public boolean canWrite(ResolvableType type, MediaType mediaType) {
+		return Resource.class.isAssignableFrom(type.getRawClass());
+	}
+
+	@Override
+	public List<MediaType> getReadableMediaTypes() {
+		return SUPPORTED_MEDIA_TYPES;
+	}
+
+	@Override
+	public List<MediaType> getWritableMediaTypes() {
+		return SUPPORTED_MEDIA_TYPES;
+	}
+
+	@Override
+	public Flux<Resource> read(ResolvableType type,
+			ReactiveHttpInputMessage inputMessage) {
+		Class<?> clazz = type.getRawClass();
+
+		Flux<DataBuffer> body = inputMessage.getBody();
+
+		if (InputStreamResource.class.equals(clazz)) {
+			InputStream is = DataBufferUtils.toInputStream(body);
+			return Flux.just(new InputStreamResource(is));
+		}
+		else if (clazz.isAssignableFrom(ByteArrayResource.class)) {
+			Mono<DataBuffer> singleBuffer = body.reduce(DataBuffer::write);
+			return Flux.from(singleBuffer.map(buffer -> {
+				byte[] bytes = new byte[buffer.readableByteCount()];
+				buffer.read(bytes);
+				return new ByteArrayResource(bytes);
+			}));
+		}
+		else {
+			return Flux.error(new IllegalStateException(
+					"Unsupported resource class: " + clazz));
+		}
+	}
+
+	@Override
+	public Mono<Void> write(Publisher<? extends Resource> inputStream,
+			ResolvableType type, MediaType contentType,
+			ReactiveHttpOutputMessage outputMessage) {
+
+		if (inputStream instanceof Mono) {
+			// single resource
+			return Mono.from(Flux.from(inputStream).
+					flatMap(resource -> {
+						HttpHeaders headers = outputMessage.getHeaders();
+						addHeaders(headers, resource, contentType);
+
+						if (resource instanceof HttpRangeResource) {
+							return writePartialContent((HttpRangeResource) resource,
+									outputMessage);
+						}
+						else {
+							return writeContent(resource, outputMessage, 0, -1);
+						}
+
+
+					}));
+		}
+		else {
+			// multiple resources, not supported!
+			return Mono.error(new IllegalArgumentException(
+					"Multiple resources not yet supported"));
+		}
+	}
+
+	protected void addHeaders(HttpHeaders headers, Resource resource,
+			MediaType contentType) {
+		if (headers.getContentType() == null) {
+			if (contentType == null ||
+					!contentType.isConcrete() ||
+					MediaType.APPLICATION_OCTET_STREAM.equals(contentType)) {
+				contentType = MimeTypeUtils2.getMimeType(resource.getFilename()).
+						map(MediaTypeUtils::toMediaType).
+						orElse(MediaType.APPLICATION_OCTET_STREAM);
+			}
+			headers.setContentType(contentType);
+		}
+		if (headers.getContentLength() < 0) {
+			contentLength(resource).ifPresent(headers::setContentLength);
+		}
+		headers.add(HttpHeaders.ACCEPT_RANGES, "bytes");
+	}
+
+	private Mono<Void> writeContent(Resource resource,
+			ReactiveHttpOutputMessage outputMessage, long position, long count) {
+		if (outputMessage instanceof ZeroCopyHttpOutputMessage) {
+			Optional<File> file = getFile(resource);
+			if (file.isPresent()) {
+				ZeroCopyHttpOutputMessage zeroCopyResponse =
+						(ZeroCopyHttpOutputMessage) outputMessage;
+
+				if (count < 0) {
+					count = file.get().length();
+				}
+
+				return zeroCopyResponse.setBody(file.get(), position, count);
+			}
+		}
+
+		// non-zero copy fallback
+		try {
+			InputStream is = resource.getInputStream();
+			long skipped = is.skip(position);
+			if (skipped < position) {
+				return Mono.error(new IOException(
+						"Skipped only " + skipped + " bytes out of " + count +
+								" required."));
+			}
+
+			Flux<DataBuffer> responseBody =
+					DataBufferUtils.read(is, outputMessage.allocator(), BUFFER_SIZE);
+			if (count > 0) {
+				responseBody = DataBufferUtils.takeUntilByteCount(responseBody, count);
+			}
+
+			return outputMessage.setBody(responseBody);
+		}
+		catch (IOException ex) {
+			return Mono.error(ex);
+		}
+	}
+
+	protected Mono<Void> writePartialContent(HttpRangeResource resource,
+			ReactiveHttpOutputMessage outputMessage) {
+
+		// TODO: implement
+
+		return Mono.empty();
+	}
+
+	private static Optional<Long> contentLength(Resource resource) {
+		// Don't try to determine contentLength on InputStreamResource - cannot be read afterwards...
+		// Note: custom InputStreamResource subclasses could provide a pre-calculated content length!
+		if (InputStreamResource.class != resource.getClass()) {
+			try {
+				return Optional.of(resource.contentLength());
+			}
+			catch (IOException ignored) {
+			}
+		}
+		return Optional.empty();
+	}
+
+	private static Optional<File> getFile(Resource resource) {
+		// TODO: introduce Resource.hasFile() property to bypass the potential IOException thrown in Resource.getFile()
+		// the following Resource implementations do not support getURI/getFile
+		if (!(resource instanceof ByteArrayResource ||
+				resource instanceof DescriptiveResource ||
+				resource instanceof InputStreamResource)) {
+			try {
+				URI resourceUri = resource.getURI();
+				if (ResourceUtils.URL_PROTOCOL_FILE.equals(resourceUri.getScheme())) {
+					return Optional.of(ResourceUtils.getFile(resourceUri));
+				}
+			}
+			catch (IOException ignored) {
+			}
+		}
+		return Optional.empty();
+	}
+
+
+}

spring-web-reactive/src/main/java/org/springframework/http/server/reactive/AbstractServerHttpResponse.java

@@ -94,7 +94,7 @@ public abstract class AbstractServerHttpResponse implements ServerHttpResponse {
 						applyBeforeCommit().after(() -> setBodyInternal(writePublisher)));
 	}
 
-	private Mono<Void> applyBeforeCommit() {
+	protected Mono<Void> applyBeforeCommit() {
 		Mono<Void> mono = Mono.empty();
 		if (this.state.compareAndSet(STATE_NEW, STATE_COMMITTING)) {
 			for (Supplier<? extends Mono<Void>> action : this.beforeCommitActions) {

spring-web-reactive/src/main/java/org/springframework/http/server/reactive/ReactorServerHttpResponse.java

@@ -16,6 +16,8 @@
 
 package org.springframework.http.server.reactive;
 
+import java.io.File;
+
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.handler.codec.http.HttpResponseStatus;
@@ -31,6 +33,7 @@ import org.springframework.core.io.buffer.DataBufferAllocator;
 import org.springframework.core.io.buffer.NettyDataBuffer;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseCookie;
+import org.springframework.http.ZeroCopyHttpOutputMessage;
 import org.springframework.util.Assert;
 
 /**
@@ -39,7 +42,8 @@ import org.springframework.util.Assert;
  * @author Stephane Maldini
  * @author Rossen Stoyanchev
  */
-public class ReactorServerHttpResponse extends AbstractServerHttpResponse {
+public class ReactorServerHttpResponse extends AbstractServerHttpResponse
+		implements ZeroCopyHttpOutputMessage {
 
 	private final HttpChannel channel;
 
@@ -99,4 +103,11 @@ public class ReactorServerHttpResponse extends AbstractServerHttpResponse {
 			return Unpooled.wrappedBuffer(buffer.asByteBuffer());
 		}
 	}
+
+	@Override
+	public Mono<Void> setBody(File file, long position, long count) {
+		return applyBeforeCommit().after(() -> {
+			return this.channel.sendFile(file, position, count);
+		});
+	}
 }

spring-web-reactive/src/main/java/org/springframework/http/server/reactive/RxNettyServerHttpResponse.java

@@ -104,4 +104,35 @@ public class RxNettyServerHttpResponse extends AbstractServerHttpResponse {
 		}
 	}
 
+/*
+	While the underlying implementation of {@link ZeroCopyHttpOutputMessage} seems to
+	work; it does bypass {@link #applyBeforeCommit} and more importantly it doesn't change
+	its {@linkplain #state()). Therefore it's commented out, for now.
+
+	We should revisit this code once
+	https://github.com/ReactiveX/RxNetty/issues/194 has been fixed.
+	
+
+	@Override
+	public Mono<Void> setBody(File file, long position, long count) {
+		Channel channel = this.response.unsafeNettyChannel();
+
+		HttpResponse httpResponse =
+				new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
+		io.netty.handler.codec.http.HttpHeaders headers = httpResponse.headers();
+
+		for (Map.Entry<String, List<String>> header : getHeaders().entrySet()) {
+			String headerName = header.getKey();
+			for (String headerValue : header.getValue()) {
+				headers.add(headerName, headerValue);
+			}
+		}
+		Mono<Void> responseWrite = MonoChannelFuture.from(channel.write(httpResponse));
+
+		FileRegion fileRegion = new DefaultFileRegion(file, position, count);
+		Mono<Void> fileWrite = MonoChannelFuture.from(channel.writeAndFlush(fileRegion));
+
+		return Flux.concat(applyBeforeCommit(), responseWrite, fileWrite).after();
+	}
+*/
 }

spring-web-reactive/src/main/java/org/springframework/http/server/reactive/UndertowHttpHandlerAdapter.java

@@ -67,9 +67,12 @@ public class UndertowHttpHandlerAdapter implements io.undertow.server.HttpHandle
 		requestBody.registerListener();
 		ServerHttpRequest request = new UndertowServerHttpRequest(exchange, requestBody);
 
-		ResponseBodySubscriber responseBody = new ResponseBodySubscriber(exchange);
+		StreamSinkChannel responseChannel = exchange.getResponseChannel();
+		ResponseBodySubscriber responseBody =
+				new ResponseBodySubscriber(exchange, responseChannel);
 		responseBody.registerListener();
-		ServerHttpResponse response = new UndertowServerHttpResponse(exchange,
+		ServerHttpResponse response =
+				new UndertowServerHttpResponse(exchange, responseChannel,
 				publisher -> Mono.from(subscriber -> publisher.subscribe(responseBody)),
 				allocator);
 
@@ -202,9 +205,10 @@ public class UndertowHttpHandlerAdapter implements io.undertow.server.HttpHandle
 
 		private Subscription subscription;
 
-		public ResponseBodySubscriber(HttpServerExchange exchange) {
+		public ResponseBodySubscriber(HttpServerExchange exchange,
+				StreamSinkChannel responseChannel) {
 			this.exchange = exchange;
-			this.responseChannel = exchange.getResponseChannel();
+			this.responseChannel = responseChannel;
 		}
 
 		public void registerListener() {

spring-web-reactive/src/main/java/org/springframework/http/server/reactive/UndertowServerHttpResponse.java

@@ -16,6 +16,10 @@
 
 package org.springframework.http.server.reactive;
 
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.nio.channels.FileChannel;
 import java.util.List;
 import java.util.Map;
 import java.util.function.Function;
@@ -25,12 +29,14 @@ import io.undertow.server.handlers.Cookie;
 import io.undertow.server.handlers.CookieImpl;
 import io.undertow.util.HttpString;
 import org.reactivestreams.Publisher;
+import org.xnio.channels.StreamSinkChannel;
 import reactor.core.publisher.Mono;
 
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferAllocator;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseCookie;
+import org.springframework.http.ZeroCopyHttpOutputMessage;
 import org.springframework.util.Assert;
 
 /**
@@ -39,19 +45,25 @@ import org.springframework.util.Assert;
  * @author Marek Hawrylczak
  * @author Rossen Stoyanchev
  */
-public class UndertowServerHttpResponse extends AbstractServerHttpResponse {
+public class UndertowServerHttpResponse extends AbstractServerHttpResponse
+		implements ZeroCopyHttpOutputMessage {
 
 	private final HttpServerExchange exchange;
 
+	private final StreamSinkChannel responseChannel;
+
 	private final Function<Publisher<DataBuffer>, Mono<Void>> responseBodyWriter;
 
 	public UndertowServerHttpResponse(HttpServerExchange exchange,
+			StreamSinkChannel responseChannel,
 			Function<Publisher<DataBuffer>, Mono<Void>> responseBodyWriter,
 			DataBufferAllocator allocator) {
 		super(allocator);
 		Assert.notNull(exchange, "'exchange' is required.");
+		Assert.notNull(responseChannel, "'responseChannel' must not be null");
 		Assert.notNull(responseBodyWriter, "'responseBodyWriter' must not be null");
 		this.exchange = exchange;
+		this.responseChannel = responseChannel;
 		this.responseBodyWriter = responseBodyWriter;
 	}
 
@@ -71,6 +83,26 @@ public class UndertowServerHttpResponse extends AbstractServerHttpResponse {
 		return this.responseBodyWriter.apply(publisher);
 	}
 
+	@Override
+	public Mono<Void> setBody(File file, long position, long count) {
+		writeHeaders();
+		writeCookies();
+		try {
+			FileChannel in = new FileInputStream(file).getChannel();
+			long result = this.responseChannel.transferFrom(in, position, count);
+			if (result < count) {
+				return Mono.error(new IOException("Could only write " + result +
+						" out of " + count + " bytes"));
+			}
+			else {
+				return Mono.empty();
+			}
+		}
+		catch (IOException ex) {
+			return Mono.error(ex);
+		}
+	}
+
 	@Override
 	protected void writeHeaders() {
 		for (Map.Entry<String, List<String>> entry : getHeaders().entrySet()) {

spring-web-reactive/src/main/java/org/springframework/web/reactive/accept/PathExtensionContentTypeResolver.java

@@ -13,23 +13,19 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package org.springframework.web.reactive.accept;
 
-import java.io.IOException;
-import java.io.InputStream;
 import java.util.Locale;
 import java.util.Map;
-import javax.activation.FileTypeMap;
-import javax.activation.MimetypesFileTypeMap;
+import java.util.Optional;
 
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-
-import org.springframework.core.io.ClassPathResource;
 import org.springframework.core.io.Resource;
 import org.springframework.http.MediaType;
+import org.springframework.http.support.MediaTypeUtils;
 import org.springframework.util.Assert;
-import org.springframework.util.ClassUtils;
+import org.springframework.util.MimeType;
+import org.springframework.util.MimeTypeUtils2;
 import org.springframework.util.StringUtils;
 import org.springframework.web.server.NotAcceptableStatusException;
 import org.springframework.web.server.ServerWebExchange;
@@ -48,12 +44,6 @@ import org.springframework.web.util.WebUtils;
  */
 public class PathExtensionContentTypeResolver extends AbstractMappingContentTypeResolver {
 
-	private static final Log logger = LogFactory.getLog(PathExtensionContentTypeResolver.class);
-
-	private static final boolean JAF_PRESENT = ClassUtils.isPresent("javax.activation.FileTypeMap",
-			PathExtensionContentTypeResolver.class.getClassLoader());
-
-
 	private boolean useJaf = true;
 
 	private boolean ignoreUnknownExtensions = true;
@@ -103,8 +93,9 @@ public class PathExtensionContentTypeResolver extends AbstractMappingContentType
 
 	@Override
 	protected MediaType handleNoMatch(String key) throws NotAcceptableStatusException {
-		if (this.useJaf && JAF_PRESENT) {
-			MediaType mediaType = JafMediaTypeFactory.getMediaType("file." + key);
+		if (this.useJaf) {
+			Optional<MimeType> mimeType = MimeTypeUtils2.getMimeType("file." + key);
+			MediaType mediaType = mimeType.map(MediaTypeUtils::toMediaType).orElse(null);
 			if (mediaType != null && !MediaType.APPLICATION_OCTET_STREAM.equals(mediaType)) {
 				return mediaType;
 			}
@@ -130,8 +121,10 @@ public class PathExtensionContentTypeResolver extends AbstractMappingContentType
 		if (extension != null) {
 			mediaType = getMediaType(extension);
 		}
-		if (mediaType == null && JAF_PRESENT) {
-			mediaType = JafMediaTypeFactory.getMediaType(filename);
+		if (mediaType == null) {
+			mediaType =
+					MimeTypeUtils2.getMimeType(filename).map(MediaTypeUtils::toMediaType)
+							.orElse(null);
 		}
 		if (MediaType.APPLICATION_OCTET_STREAM.equals(mediaType)) {
 			mediaType = null;
@@ -139,56 +132,4 @@ public class PathExtensionContentTypeResolver extends AbstractMappingContentType
 		return mediaType;
 	}
 
-
-	/**
-	 * Inner class to avoid hard-coded dependency on JAF.
-	 */
-	private static class JafMediaTypeFactory {
-
-		private static final FileTypeMap fileTypeMap;
-
-		static {
-			fileTypeMap = initFileTypeMap();
-		}
-
-		/**
-		 * Find extended mime.types from the spring-context-support module.
-		 */
-		private static FileTypeMap initFileTypeMap() {
-			Resource resource = new ClassPathResource("org/springframework/mail/javamail/mime.types");
-			if (resource.exists()) {
-				if (logger.isTraceEnabled()) {
-					logger.trace("Loading JAF FileTypeMap from " + resource);
-				}
-				InputStream inputStream = null;
-				try {
-					inputStream = resource.getInputStream();
-					return new MimetypesFileTypeMap(inputStream);
-				}
-				catch (IOException ex) {
-					// ignore
-				}
-				finally {
-					if (inputStream != null) {
-						try {
-							inputStream.close();
-						}
-						catch (IOException ex) {
-							// ignore
-						}
-					}
-				}
-			}
-			if (logger.isTraceEnabled()) {
-				logger.trace("Loading default Java Activation Framework FileTypeMap");
-			}
-			return FileTypeMap.getDefaultFileTypeMap();
-		}
-
-		public static MediaType getMediaType(String filename) {
-			String mediaType = fileTypeMap.getContentType(filename);
-			return (StringUtils.hasText(mediaType) ? MediaType.parseMediaType(mediaType) : null);
-		}
-	}
-
 }

spring-web-reactive/src/main/java/org/springframework/web/reactive/result/method/annotation/RequestBodyArgumentResolver.java

@@ -24,10 +24,10 @@ import reactor.core.publisher.Mono;
 
 import org.springframework.core.MethodParameter;
 import org.springframework.core.ResolvableType;
-import org.springframework.core.codec.Decoder;
 import org.springframework.core.convert.ConversionService;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.http.MediaType;
+import org.springframework.http.converter.reactive.HttpMessageConverter;
 import org.springframework.ui.ModelMap;
 import org.springframework.util.Assert;
 import org.springframework.web.bind.annotation.RequestBody;
@@ -40,15 +40,15 @@ import org.springframework.web.server.ServerWebExchange;
  */
 public class RequestBodyArgumentResolver implements HandlerMethodArgumentResolver {
 
-	private final List<Decoder<?>> decoders;
+	private final List<HttpMessageConverter<?>> messageConverters;
 
 	private final ConversionService conversionService;
 
-
-	public RequestBodyArgumentResolver(List<Decoder<?>> decoders, ConversionService service) {
-		Assert.notEmpty(decoders, "At least one decoder is required.");
+	public RequestBodyArgumentResolver(List<HttpMessageConverter<?>> messageConverters,
+			ConversionService service) {
+		Assert.notEmpty(messageConverters, "At least one message converter is required.");
 		Assert.notNull(service, "'conversionService' is required.");
-		this.decoders = decoders;
+		this.messageConverters = messageConverters;
 		this.conversionService = service;
 	}
 
@@ -62,22 +62,29 @@ public class RequestBodyArgumentResolver implements HandlerMethodArgumentResolve
 	public Mono<Object> resolveArgument(MethodParameter parameter, ModelMap model,
 			ServerWebExchange exchange) {
 
+		ResolvableType type = ResolvableType.forMethodParameter(parameter);
+		ResolvableType elementType = type.hasGenerics() ? type.getGeneric(0) : type;
+
 		MediaType mediaType = exchange.getRequest().getHeaders().getContentType();
 		if (mediaType == null) {
 			mediaType = MediaType.APPLICATION_OCTET_STREAM;
 		}
-		ResolvableType type = ResolvableType.forMethodParameter(parameter);
-		Flux<DataBuffer> body = exchange.getRequest().getBody();
-		Flux<?> elementFlux = body;
-		ResolvableType elementType = type.hasGenerics() ? type.getGeneric(0) : type;
 
-		Decoder<?> decoder = resolveDecoder(elementType, mediaType);
-		if (decoder != null) {
-			elementFlux = decoder.decode(body, elementType, mediaType);
+		Flux<DataBuffer> body = exchange.getRequest().getBody();
+		Flux<?> elementFlux;
+
+		HttpMessageConverter<?> messageConverter =
+				resolveMessageConverter(elementType, mediaType);
+		if (messageConverter != null) {
+			elementFlux = messageConverter.read(elementType, exchange.getRequest());
+		}
+		else {
+			elementFlux = body;
 		}
 
 		if (this.conversionService.canConvert(Publisher.class, type.getRawClass())) {
-			return Mono.just(this.conversionService.convert(elementFlux, type.getRawClass()));
+			return Mono.just(this.conversionService
+					.convert(elementFlux, type.getRawClass()));
 		}
 		else if (type.getRawClass() == Flux.class) {
 			return Mono.just(elementFlux);
@@ -90,10 +97,11 @@ public class RequestBodyArgumentResolver implements HandlerMethodArgumentResolve
 		return elementFlux.next().map(o -> o);
 	}
 
-	private Decoder<?> resolveDecoder(ResolvableType type, MediaType mediaType, Object... hints) {
-		for (Decoder<?> decoder : this.decoders) {
-			if (decoder.canDecode(type, mediaType, hints)) {
-				return decoder;
+	private HttpMessageConverter<?> resolveMessageConverter(ResolvableType type,
+			MediaType mediaType) {
+		for (HttpMessageConverter<?> messageConverter : this.messageConverters) {
+			if (messageConverter.canRead(type, mediaType)) {
+				return messageConverter;
 			}
 		}
 		return null;

spring-web-reactive/src/main/java/org/springframework/web/reactive/result/method/annotation/RequestMappingHandlerAdapter.java

@@ -17,6 +17,7 @@
 package org.springframework.web.reactive.result.method.annotation;
 
 import java.lang.reflect.Method;
+import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
@@ -28,16 +29,19 @@ import org.apache.commons.logging.LogFactory;
 import reactor.core.publisher.Mono;
 
 import org.springframework.beans.factory.InitializingBean;
-import org.springframework.core.codec.Decoder;
 import org.springframework.core.codec.support.ByteBufferDecoder;
+import org.springframework.core.codec.support.ByteBufferEncoder;
 import org.springframework.core.codec.support.JacksonJsonDecoder;
+import org.springframework.core.codec.support.JacksonJsonEncoder;
 import org.springframework.core.codec.support.Jaxb2Decoder;
+import org.springframework.core.codec.support.Jaxb2Encoder;
 import org.springframework.core.codec.support.JsonObjectDecoder;
 import org.springframework.core.codec.support.StringDecoder;
+import org.springframework.core.codec.support.StringEncoder;
 import org.springframework.core.convert.ConversionService;
 import org.springframework.core.convert.support.DefaultConversionService;
-import org.springframework.core.io.buffer.DataBufferAllocator;
-import org.springframework.core.io.buffer.DefaultDataBufferAllocator;
+import org.springframework.http.converter.reactive.CodecHttpMessageConverter;
+import org.springframework.http.converter.reactive.HttpMessageConverter;
 import org.springframework.ui.ExtendedModelMap;
 import org.springframework.ui.ModelMap;
 import org.springframework.util.ObjectUtils;
@@ -62,8 +66,6 @@ public class RequestMappingHandlerAdapter implements HandlerAdapter, Initializin
 
 	private ConversionService conversionService = new DefaultConversionService();
 
-	private DataBufferAllocator allocator = new DefaultDataBufferAllocator();
-
 	private final Map<Class<?>, ExceptionHandlerMethodResolver> exceptionHandlerCache =
 			new ConcurrentHashMap<>(64);
 
@@ -92,20 +94,23 @@ public class RequestMappingHandlerAdapter implements HandlerAdapter, Initializin
 		return this.conversionService;
 	}
 
-	public void setAllocator(DataBufferAllocator allocator) {
-		this.allocator = allocator;
-	}
-
 	@Override
 	public void afterPropertiesSet() throws Exception {
 		if (ObjectUtils.isEmpty(this.argumentResolvers)) {
+			List<HttpMessageConverter<?>> messageConverters = Arrays.asList(
+					new CodecHttpMessageConverter<ByteBuffer>(new ByteBufferEncoder(),
+							new ByteBufferDecoder()),
+					new CodecHttpMessageConverter<String>(new StringEncoder(),
+							new StringDecoder()),
+					new CodecHttpMessageConverter<Object>(new Jaxb2Encoder(),
+							new Jaxb2Decoder()),
+					new CodecHttpMessageConverter<Object>(new JacksonJsonEncoder(),
+							new JacksonJsonDecoder(new JsonObjectDecoder())));
 
-			List<Decoder<?>> decoders = Arrays.asList(new ByteBufferDecoder(),
-					new StringDecoder(), new Jaxb2Decoder(),
-					new JacksonJsonDecoder(new JsonObjectDecoder()));
 
 			this.argumentResolvers.add(new RequestParamArgumentResolver());
-			this.argumentResolvers.add(new RequestBodyArgumentResolver(decoders, this.conversionService));
+			this.argumentResolvers.add(new RequestBodyArgumentResolver(messageConverters,
+					this.conversionService));
 			this.argumentResolvers.add(new ModelArgumentResolver());
 		}
 	}

spring-web-reactive/src/main/java/org/springframework/web/reactive/result/method/annotation/ResponseBodyResultHandler.java

@@ -34,10 +34,9 @@ import org.springframework.core.MethodParameter;
 import org.springframework.core.Ordered;
 import org.springframework.core.ResolvableType;
 import org.springframework.core.annotation.AnnotationUtils;
-import org.springframework.core.codec.Encoder;
 import org.springframework.core.convert.ConversionService;
-import org.springframework.core.io.buffer.DataBufferAllocator;
 import org.springframework.http.MediaType;
+import org.springframework.http.converter.reactive.HttpMessageConverter;
 import org.springframework.http.server.reactive.ServerHttpRequest;
 import org.springframework.http.server.reactive.ServerHttpResponse;
 import org.springframework.util.Assert;
@@ -54,60 +53,51 @@ import org.springframework.web.server.ServerWebExchange;
  * @author Rossen Stoyanchev
  * @author Stephane Maldini
  * @author Sebastien Deleuze
+ * @author Arjen Poutsma
  */
 public class ResponseBodyResultHandler implements HandlerResultHandler, Ordered {
 
 	private static final MediaType MEDIA_TYPE_APPLICATION = new MediaType("application");
 
-	private final List<Encoder<?>> encoders;
+	private final List<HttpMessageConverter<?>> messageConverters;
 
 	private final ConversionService conversionService;
 
 	private final List<MediaType> allMediaTypes;
 
-	private final Map<Encoder<?>, List<MediaType>> mediaTypesByEncoder;
+	private final Map<HttpMessageConverter<?>, List<MediaType>> mediaTypesByEncoder;
 
 	private int order = 0; // TODO: should be MAX_VALUE
 
-
-	public ResponseBodyResultHandler(List<Encoder<?>> encoders, ConversionService service) {
-		Assert.notEmpty(encoders, "At least one encoders is required.");
+	public ResponseBodyResultHandler(List<HttpMessageConverter<?>> messageConverters,
+			ConversionService service) {
+		Assert.notEmpty(messageConverters, "At least one message converter is required.");
 		Assert.notNull(service, "'conversionService' is required.");
-		this.encoders = encoders;
+		this.messageConverters = messageConverters;
 		this.conversionService = service;
-		this.allMediaTypes = getAllMediaTypes(encoders);
-		this.mediaTypesByEncoder = getMediaTypesByEncoder(encoders);
+		this.allMediaTypes = getAllMediaTypes(messageConverters);
+		this.mediaTypesByEncoder = getMediaTypesByConverter(messageConverters);
 	}
 
-	private static List<MediaType> getAllMediaTypes(List<Encoder<?>> encoders) {
+	private static List<MediaType> getAllMediaTypes(
+			List<HttpMessageConverter<?>> messageConverters) {
 		Set<MediaType> set = new LinkedHashSet<>();
-		encoders.forEach(encoder -> set.addAll(toMediaTypes(encoder.getSupportedMimeTypes())));
+		messageConverters.forEach(
+				converter -> set.addAll(converter.getWritableMediaTypes()));
 		List<MediaType> result = new ArrayList<>(set);
 		MediaType.sortBySpecificity(result);
 		return Collections.unmodifiableList(result);
 	}
 
-	private static Map<Encoder<?>, List<MediaType>> getMediaTypesByEncoder(List<Encoder<?>> encoders) {
-		Map<Encoder<?>, List<MediaType>> result = new HashMap<>(encoders.size());
-		encoders.forEach(encoder -> result.put(encoder, toMediaTypes(encoder.getSupportedMimeTypes())));
+	private static Map<HttpMessageConverter<?>, List<MediaType>> getMediaTypesByConverter(
+			List<HttpMessageConverter<?>> converters) {
+		Map<HttpMessageConverter<?>, List<MediaType>> result =
+				new HashMap<>(converters.size());
+		converters.forEach(converter -> result
+				.put(converter, converter.getWritableMediaTypes()));
 		return Collections.unmodifiableMap(result);
 	}
 
-	/**
-	 * TODO: MediaType static method
-	 */
-	private static List<MediaType> toMediaTypes(List<MimeType> mimeTypes) {
-		return mimeTypes.stream().map(ResponseBodyResultHandler::toMediaType).collect(Collectors.toList());
-	}
-
-	/**
-	 * TODO: MediaType constructor
-	 */
-	private static MediaType toMediaType(MimeType mimeType) {
-		return new MediaType(mimeType.getType(), mimeType.getSubtype(), mimeType.getParameters());
-	}
-
-
 	public void setOrder(int order) {
 		this.order = order;
 	}
@@ -154,65 +144,77 @@ public class ResponseBodyResultHandler implements HandlerResultHandler, Ordered
 			elementType = returnType;
 		}
 
-		List<MediaType> requestedMediaTypes = getAcceptableMediaTypes(exchange.getRequest());
-		List<MediaType> producibleMediaTypes = getProducibleMediaTypes(elementType);
-
-		if (producibleMediaTypes.isEmpty()) {
-			producibleMediaTypes.add(MediaType.ALL);
-		}
-
-		Set<MediaType> compatibleMediaTypes = new LinkedHashSet<>();
-		for (MediaType requestedType : requestedMediaTypes) {
-			for (MediaType producibleType : producibleMediaTypes) {
-				if (requestedType.isCompatibleWith(producibleType)) {
-					compatibleMediaTypes.add(getMostSpecificMediaType(requestedType, producibleType));
-				}
-			}
-		}
+		List<MediaType> compatibleMediaTypes =
+				getCompatibleMediaTypes(exchange.getRequest(), elementType);
 		if (compatibleMediaTypes.isEmpty()) {
-			return Mono.error(new NotAcceptableStatusException(producibleMediaTypes));
+			return Mono.error(new NotAcceptableStatusException(
+					getProducibleMediaTypes(elementType)));
 		}
 
-		List<MediaType> mediaTypes = new ArrayList<>(compatibleMediaTypes);
-		MediaType.sortBySpecificityAndQuality(mediaTypes);
+		Optional<MediaType> selectedMediaType = selectBestMediaType(compatibleMediaTypes);
 
-		MediaType selectedMediaType = null;
-		for (MediaType mediaType : mediaTypes) {
-			if (mediaType.isConcrete()) {
-				selectedMediaType = mediaType;
-				break;
-			}
-			else if (mediaType.equals(MediaType.ALL) || mediaType.equals(MEDIA_TYPE_APPLICATION)) {
-				selectedMediaType = MediaType.APPLICATION_OCTET_STREAM;
-				break;
-			}
-		}
-
-		if (selectedMediaType != null) {
-			Encoder<?> encoder = resolveEncoder(elementType, selectedMediaType);
-			if (encoder != null) {
+		if (selectedMediaType.isPresent()) {
+			HttpMessageConverter<?> converter =
+					resolveEncoder(elementType, selectedMediaType.get());
+			if (converter != null) {
 				ServerHttpResponse response = exchange.getResponse();
-				response.getHeaders().setContentType(selectedMediaType);
-				DataBufferAllocator allocator = response.allocator();
-				return response.setBody(
-						encoder.encode((Publisher) publisher, allocator, elementType,
-								selectedMediaType));
+				return converter.write((Publisher) publisher, elementType,
+						selectedMediaType.get(),
+								response);
 			}
 		}
 
 		return Mono.error(new NotAcceptableStatusException(this.allMediaTypes));
 	}
 
+	private List<MediaType> getCompatibleMediaTypes(ServerHttpRequest request,
+			ResolvableType elementType) {
+
+		List<MediaType> acceptableMediaTypes = getAcceptableMediaTypes(request);
+		List<MediaType> producibleMediaTypes = getProducibleMediaTypes(elementType);
+
+		Set<MediaType> compatibleMediaTypes = new LinkedHashSet<>();
+		for (MediaType acceptableMediaType : acceptableMediaTypes) {
+			compatibleMediaTypes.addAll(producibleMediaTypes.stream().
+					filter(acceptableMediaType::isCompatibleWith).
+					map(producibleType -> getMostSpecificMediaType(acceptableMediaType,
+							producibleType)).collect(Collectors.toList()));
+		}
+
+		List<MediaType> result = new ArrayList<>(compatibleMediaTypes);
+		MediaType.sortBySpecificityAndQuality(result);
+		return result;
+	}
+
 	private List<MediaType> getAcceptableMediaTypes(ServerHttpRequest request) {
 		List<MediaType> mediaTypes = request.getHeaders().getAccept();
 		return (mediaTypes.isEmpty() ? Collections.singletonList(MediaType.ALL) : mediaTypes);
 	}
 
+	private Optional<MediaType> selectBestMediaType(
+			List<MediaType> compatibleMediaTypes) {
+		for (MediaType mediaType : compatibleMediaTypes) {
+			if (mediaType.isConcrete()) {
+				return Optional.of(mediaType);
+			}
+			else if (mediaType.equals(MediaType.ALL) ||
+					mediaType.equals(MEDIA_TYPE_APPLICATION)) {
+				return Optional.of(MediaType.APPLICATION_OCTET_STREAM);
+			}
+		}
+		return Optional.empty();
+	}
+
 	private List<MediaType> getProducibleMediaTypes(ResolvableType type) {
-		return this.encoders.stream()
-				.filter(encoder -> encoder.canEncode(type, null))
+		List<MediaType> result = this.messageConverters.stream()
+				.filter(converter -> converter.canWrite(type, null))
 				.flatMap(encoder -> this.mediaTypesByEncoder.get(encoder).stream())
 				.collect(Collectors.toList());
+		if (result.isEmpty()) {
+			result.add(MediaType.ALL);
+		}
+
+		return result;
 	}
 
 	/**
@@ -225,10 +227,11 @@ public class ResponseBodyResultHandler implements HandlerResultHandler, Ordered
 		return (comparator.compare(acceptType, produceType) <= 0 ? acceptType : produceType);
 	}
 
-	private Encoder<?> resolveEncoder(ResolvableType type, MediaType mediaType, Object... hints) {
-		for (Encoder<?> encoder : this.encoders) {
-			if (encoder.canEncode(type, mediaType, hints)) {
-				return encoder;
+	private HttpMessageConverter<?> resolveEncoder(ResolvableType type,
+			MediaType mediaType) {
+		for (HttpMessageConverter<?> converter : this.messageConverters) {
+			if (converter.canWrite(type, mediaType)) {
+				return converter;
 			}
 		}
 		return null;

spring-web-reactive/src/test/java/org/springframework/http/server/reactive/ZeroCopyIntegrationTests.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.http.server.reactive;
+
+import java.io.File;
+import java.net.URI;
+
+import org.junit.Test;
+import reactor.core.publisher.Mono;
+
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.core.io.Resource;
+import org.springframework.http.MediaType;
+import org.springframework.http.RequestEntity;
+import org.springframework.http.ResponseEntity;
+import org.springframework.http.ZeroCopyHttpOutputMessage;
+import org.springframework.http.server.reactive.boot.ReactorHttpServer;
+import org.springframework.http.server.reactive.boot.UndertowHttpServer;
+import org.springframework.web.client.RestTemplate;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assume.assumeTrue;
+
+/**
+ * @author Arjen Poutsma
+ */
+public class ZeroCopyIntegrationTests extends AbstractHttpHandlerIntegrationTests {
+
+	private final ZeroCopyHandler handler = new ZeroCopyHandler();
+
+	@Override
+	protected HttpHandler createHttpHandler() {
+		return handler;
+	}
+
+	@Test
+	public void zeroCopy() throws Exception {
+		// Zero-copy only does not support servlet
+		assumeTrue(server instanceof ReactorHttpServer ||
+				server instanceof UndertowHttpServer);
+
+		RestTemplate restTemplate = new RestTemplate();
+
+		RequestEntity request =
+				RequestEntity.get(new URI("http://localhost:" + port)).build();
+
+		ResponseEntity<byte[]> response = restTemplate.exchange(request, byte[].class);
+
+		Resource logo =
+				new ClassPathResource("spring.png", ZeroCopyIntegrationTests.class);
+
+		assertTrue(response.hasBody());
+		assertEquals(logo.contentLength(), response.getHeaders().getContentLength());
+		assertEquals(logo.contentLength(), response.getBody().length);
+		assertEquals(MediaType.IMAGE_PNG, response.getHeaders().getContentType());
+
+	}
+
+	private static class ZeroCopyHandler implements HttpHandler {
+
+		@Override
+		public Mono<Void> handle(ServerHttpRequest request, ServerHttpResponse response) {
+			try {
+				ZeroCopyHttpOutputMessage zeroCopyResponse =
+						(ZeroCopyHttpOutputMessage) response;
+
+				Resource logo = new ClassPathResource("spring.png",
+						ZeroCopyIntegrationTests.class);
+				File logoFile = logo.getFile();
+				zeroCopyResponse.getHeaders().setContentType(MediaType.IMAGE_PNG);
+				zeroCopyResponse.getHeaders().setContentLength(logoFile.length());
+				return zeroCopyResponse.setBody(logoFile, 0, logoFile.length());
+
+			}
+			catch (Throwable ex) {
+				return Mono.error(ex);
+			}
+
+
+		}
+	}
+
+}

spring-web-reactive/src/test/java/org/springframework/web/reactive/DispatcherHandlerErrorTests.java

@@ -30,7 +30,7 @@ import reactor.core.util.SignalKind;
 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
-import org.springframework.core.codec.Encoder;
+import org.springframework.core.codec.support.StringDecoder;
 import org.springframework.core.codec.support.StringEncoder;
 import org.springframework.core.convert.support.DefaultConversionService;
 import org.springframework.core.io.buffer.DataBuffer;
@@ -38,6 +38,8 @@ import org.springframework.core.io.buffer.DefaultDataBufferAllocator;
 import org.springframework.http.HttpMethod;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.MediaType;
+import org.springframework.http.converter.reactive.CodecHttpMessageConverter;
+import org.springframework.http.converter.reactive.HttpMessageConverter;
 import org.springframework.http.server.reactive.MockServerHttpRequest;
 import org.springframework.http.server.reactive.MockServerHttpResponse;
 import org.springframework.stereotype.Controller;
@@ -230,8 +232,11 @@ public class DispatcherHandlerErrorTests {
 
 		@Bean
 		public ResponseBodyResultHandler resultHandler() {
-			List<Encoder<?>> encoders = Collections.singletonList(new StringEncoder());
-			return new ResponseBodyResultHandler(encoders, new DefaultConversionService());
+			List<HttpMessageConverter<?>> converters = Collections.singletonList(
+					new CodecHttpMessageConverter<>(new StringEncoder(),
+							new StringDecoder()));
+			return new ResponseBodyResultHandler(converters,
+					new DefaultConversionService());
 		}
 
 		@Bean

spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/RequestMappingIntegrationTests.java

@@ -40,14 +40,18 @@ import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.core.ParameterizedTypeReference;
 import org.springframework.core.ResolvableType;
-import org.springframework.core.codec.Encoder;
+import org.springframework.core.codec.support.ByteBufferDecoder;
 import org.springframework.core.codec.support.ByteBufferEncoder;
+import org.springframework.core.codec.support.JacksonJsonDecoder;
 import org.springframework.core.codec.support.JacksonJsonEncoder;
+import org.springframework.core.codec.support.StringDecoder;
 import org.springframework.core.codec.support.StringEncoder;
 import org.springframework.core.convert.ConversionService;
 import org.springframework.core.convert.support.GenericConversionService;
 import org.springframework.core.convert.support.ReactiveStreamsToCompletableFutureConverter;
 import org.springframework.core.convert.support.ReactiveStreamsToRxJava1Converter;
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.core.io.Resource;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferAllocator;
 import org.springframework.core.io.buffer.DefaultDataBufferAllocator;
@@ -55,14 +59,19 @@ import org.springframework.http.HttpStatus;
 import org.springframework.http.MediaType;
 import org.springframework.http.RequestEntity;
 import org.springframework.http.ResponseEntity;
+import org.springframework.http.converter.reactive.CodecHttpMessageConverter;
+import org.springframework.http.converter.reactive.HttpMessageConverter;
+import org.springframework.http.converter.reactive.ResourceHttpMessageConverter;
 import org.springframework.http.server.reactive.AbstractHttpHandlerIntegrationTests;
 import org.springframework.http.server.reactive.HttpHandler;
+import org.springframework.http.server.reactive.ZeroCopyIntegrationTests;
 import org.springframework.stereotype.Controller;
 import org.springframework.ui.Model;
 import org.springframework.web.bind.annotation.ExceptionHandler;
 import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.ResponseBody;
 import org.springframework.web.bind.annotation.RestController;
 import org.springframework.web.client.RestTemplate;
 import org.springframework.web.reactive.DispatcherHandler;
@@ -73,8 +82,7 @@ import org.springframework.web.reactive.view.freemarker.FreeMarkerConfigurer;
 import org.springframework.web.reactive.view.freemarker.FreeMarkerViewResolver;
 import org.springframework.web.server.adapter.WebHttpHandlerBuilder;
 
-import static org.junit.Assert.assertArrayEquals;
-import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.*;
 
 /**
  * @author Rossen Stoyanchev
@@ -85,6 +93,8 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati
 
 	private AnnotationConfigApplicationContext wac;
 
+	private RestTemplate restTemplate = new RestTemplate();
+
 
 	@Override
 	protected HttpHandler createHttpHandler() {
@@ -100,9 +110,6 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati
 
 	@Test
 	public void helloWithQueryParam() throws Exception {
-
-		RestTemplate restTemplate = new RestTemplate();
-
 		URI url = new URI("http://localhost:" + port + "/param?name=George");
 		RequestEntity<Void> request = RequestEntity.get(url).build();
 		ResponseEntity<String> response = restTemplate.exchange(request, String.class);
@@ -112,9 +119,6 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati
 
 	@Test
 	public void rawPojoResponse() throws Exception {
-
-		RestTemplate restTemplate = new RestTemplate();
-
 		URI url = new URI("http://localhost:" + port + "/raw");
 		RequestEntity<Void> request =
 				RequestEntity.get(url).accept(MediaType.APPLICATION_JSON).build();
@@ -125,9 +129,6 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati
 
 	@Test
 	public void rawFluxResponse() throws Exception {
-
-		RestTemplate restTemplate = new RestTemplate();
-
 		URI url = new URI("http://localhost:" + port + "/raw-flux");
 		RequestEntity<Void> request = RequestEntity.get(url).build();
 		ResponseEntity<String> response = restTemplate.exchange(request, String.class);
@@ -137,9 +138,6 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati
 
 	@Test
 	public void rawObservableResponse() throws Exception {
-
-		RestTemplate restTemplate = new RestTemplate();
-
 		URI url = new URI("http://localhost:" + port + "/raw-observable");
 		RequestEntity<Void> request = RequestEntity.get(url).build();
 		ResponseEntity<String> response = restTemplate.exchange(request, String.class);
@@ -149,9 +147,6 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati
 
 	@Test
 	public void handleWithThrownException() throws Exception {
-
-		RestTemplate restTemplate = new RestTemplate();
-
 		URI url = new URI("http://localhost:" + port + "/thrown-exception");
 		RequestEntity<Void> request = RequestEntity.get(url).build();
 		ResponseEntity<String> response = restTemplate.exchange(request, String.class);
@@ -161,9 +156,6 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati
 
 	@Test
 	public void handleWithErrorSignal() throws Exception {
-
-		RestTemplate restTemplate = new RestTemplate();
-
 		URI url = new URI("http://localhost:" + port + "/error-signal");
 		RequestEntity<Void> request = RequestEntity.get(url).build();
 		ResponseEntity<String> response = restTemplate.exchange(request, String.class);
@@ -174,8 +166,6 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati
 	@Test
 	@Ignore
 	public void streamResult() throws Exception {
-		RestTemplate restTemplate = new RestTemplate();
-
 		URI url = new URI("http://localhost:" + port + "/stream-result");
 		RequestEntity<Void> request = RequestEntity.get(url).build();
 		ResponseEntity<String[]> response = restTemplate.exchange(request, String[].class);
@@ -295,9 +285,6 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati
 
 	@Test
 	public void html() throws Exception {
-
-		RestTemplate restTemplate = new RestTemplate();
-
 		URI url = new URI("http://localhost:" + port + "/html?name=Jason");
 		RequestEntity<Void> request = RequestEntity.get(url).accept(MediaType.TEXT_HTML).build();
 		ResponseEntity<String> response = restTemplate.exchange(request, String.class);
@@ -305,9 +292,20 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati
 		assertEquals("<html><body>Hello: Jason!</body></html>", response.getBody());
 	}
 
+	@Test
+	public void resource() throws Exception {
+		URI url = new URI("http://localhost:" + port + "/resource");
+		RequestEntity<Void> request = RequestEntity.get(url).build();
+		ResponseEntity<byte[]> response = restTemplate.exchange(request, byte[].class);
+
+		assertTrue(response.hasBody());
+		assertEquals(951, response.getHeaders().getContentLength());
+		assertEquals(951, response.getBody().length);
+		assertEquals(new MediaType("image", "x-png"),
+				response.getHeaders().getContentType());
+	}
 
 	private void serializeAsPojo(String requestUrl) throws Exception {
-		RestTemplate restTemplate = new RestTemplate();
 		RequestEntity<Void> request = RequestEntity.get(new URI(requestUrl))
 				.accept(MediaType.APPLICATION_JSON)
 				.build();
@@ -317,7 +315,6 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati
 	}
 
 	private void serializeAsCollection(String requestUrl) throws Exception {
-		RestTemplate restTemplate = new RestTemplate();
 		RequestEntity<Void> request = RequestEntity.get(new URI(requestUrl))
 				.accept(MediaType.APPLICATION_JSON)
 				.build();
@@ -331,7 +328,6 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati
 
 
 	private void capitalizePojo(String requestUrl) throws Exception {
-		RestTemplate restTemplate = new RestTemplate();
 		RequestEntity<Person> request = RequestEntity.post(new URI(requestUrl))
 				.contentType(MediaType.APPLICATION_JSON)
 				.accept(MediaType.APPLICATION_JSON)
@@ -342,7 +338,6 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati
 	}
 
 	private void capitalizeCollection(String requestUrl) throws Exception {
-		RestTemplate restTemplate = new RestTemplate();
 		RequestEntity<List<Person>> request = RequestEntity.post(new URI(requestUrl))
 				.contentType(MediaType.APPLICATION_JSON)
 				.accept(MediaType.APPLICATION_JSON)
@@ -356,7 +351,6 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati
 	}
 
 	private void createJson(String requestUrl) throws Exception {
-		RestTemplate restTemplate = new RestTemplate();
 		URI url = new URI(requestUrl);
 		RequestEntity<List<Person>> request = RequestEntity.post(url)
 				.contentType(MediaType.APPLICATION_JSON)
@@ -368,7 +362,6 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati
 	}
 
 	private void createXml(String requestUrl) throws Exception {
-		RestTemplate restTemplate = new RestTemplate();
 		URI url = new URI(requestUrl);
 		People people = new People();
 		people.getPerson().add(new Person("Robert"));
@@ -413,9 +406,16 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati
 
 		@Bean
 		public ResponseBodyResultHandler responseBodyResultHandler() {
-			List<Encoder<?>> encoders = Arrays.asList(new ByteBufferEncoder(),
-					new StringEncoder(), new JacksonJsonEncoder());
-			ResponseBodyResultHandler resultHandler = new ResponseBodyResultHandler(encoders, conversionService());
+			List<HttpMessageConverter<?>> converters =
+					Arrays.asList(new ResourceHttpMessageConverter(),
+							new CodecHttpMessageConverter<ByteBuffer>(
+									new ByteBufferEncoder(), new ByteBufferDecoder()),
+							new CodecHttpMessageConverter<String>(new StringEncoder(),
+									new StringDecoder()),
+							new CodecHttpMessageConverter<Object>(
+									new JacksonJsonEncoder(), new JacksonJsonDecoder()));
+			ResponseBodyResultHandler resultHandler =
+					new ResponseBodyResultHandler(converters, conversionService());
 			resultHandler.setOrder(1);
 			return resultHandler;
 		}
@@ -626,6 +626,12 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati
 			return Mono.just("Recovered from error: " + ex.getMessage());
 		}
 
+		@RequestMapping("/resource")
+		@ResponseBody
+		public Resource resource() {
+			return new ClassPathResource("spring.png", ZeroCopyIntegrationTests.class);
+		}
+
 		//TODO add mixed and T request mappings tests
 
 	}

spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/ResponseBodyResultHandlerTests.java

@@ -24,6 +24,7 @@ import org.reactivestreams.Publisher;
 import org.springframework.core.ResolvableType;
 import org.springframework.core.codec.support.StringEncoder;
 import org.springframework.core.convert.support.DefaultConversionService;
+import org.springframework.http.converter.reactive.CodecHttpMessageConverter;
 import org.springframework.ui.ExtendedModelMap;
 import org.springframework.web.bind.annotation.ResponseBody;
 import org.springframework.web.method.HandlerMethod;
@@ -41,7 +42,7 @@ public class ResponseBodyResultHandlerTests {
 	@Test
 	public void supports() throws NoSuchMethodException {
 		ResponseBodyResultHandler handler = new ResponseBodyResultHandler(Collections.singletonList(
-				new StringEncoder()),
+				new CodecHttpMessageConverter<String>(new StringEncoder(), null)),
 				new DefaultConversionService());
 		TestController controller = new TestController();