Wrapping up zero-copy support

This commit wraps up the previous commits:

 - It uses HttpMessageConverter in the web.reactive.server package instead of Encoder/Decoder.
 - It introduces tests for the Resource @ResponseBodies.

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 org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
+import java.util.Optional;
 
-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;
+		Flux<?> elementFlux;
 
-		Decoder<?> decoder = resolveDecoder(elementType, mediaType);
-		if (decoder != null) {
-			elementFlux = decoder.decode(body, elementType, mediaType);
+		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);
-
-		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;
-			}
-		}
+		Optional<MediaType> selectedMediaType = selectBestMediaType(compatibleMediaTypes);
 
-		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/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();