Add handler method parameter and result converters

This commit introduces the following changes:
 - Publisher -> Observable/Stream/etc. conversion is now managed
    in a dedicated ConversionService instead of directly in
    RequestBodyArgumentResolver and ResponseBodyResultHandler
 - More isolated logic that decides if the stream should be
    serialized as a JSON array or not
 - Publisher<ByteBuffer> are now handled by regular
   ByteBufferEncoder and ByteBufferDecoder
 - Handle Publisher<Void> return value properly
 - Ensure that the headers are properly written even for response
   without body
 - Improve JsonObjectEncoder to autodetect JSON arrays

spring-web-reactive/src/main/java/org/springframework/core/convert/support/ReactiveStreamsToCompletableFutureConverter.java

@@ -0,0 +1,53 @@
+/*
+ * Copyright 2002-2015 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.core.convert.support;
+
+import java.util.LinkedHashSet;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+
+import org.reactivestreams.Publisher;
+
+import org.springframework.core.convert.TypeDescriptor;
+import org.springframework.core.convert.converter.GenericConverter;
+
+/**
+ * @author Sebastien Deleuze
+ */
+public class ReactiveStreamsToCompletableFutureConverter implements GenericConverter {
+
+	@Override
+	public Set<ConvertiblePair> getConvertibleTypes() {
+		Set<GenericConverter.ConvertiblePair> convertibleTypes = new LinkedHashSet<>();
+		convertibleTypes.add(new GenericConverter.ConvertiblePair(Publisher.class, CompletableFuture.class));
+		convertibleTypes.add(new GenericConverter.ConvertiblePair(CompletableFuture.class, Publisher.class));
+		return convertibleTypes;
+	}
+
+	@Override
+	public Object convert(Object source, TypeDescriptor sourceType, TypeDescriptor targetType) {
+		if (source != null) {
+			if (CompletableFuture.class.isAssignableFrom(source.getClass())) {
+				return reactor.core.publisher.convert.CompletableFutureConverter.from((CompletableFuture)source);
+			} else if (CompletableFuture.class.isAssignableFrom(targetType.getResolvableType().getRawClass())) {
+				return reactor.core.publisher.convert.CompletableFutureConverter.fromSingle((Publisher)source);
+			}
+		}
+		return null;
+	}
+
+}

spring-web-reactive/src/main/java/org/springframework/core/convert/support/ReactiveStreamsToReactorConverter.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright (c) 2011-2015 Pivotal Software Inc, All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.core.convert.support;
+
+import java.util.LinkedHashSet;
+import java.util.Set;
+
+import org.reactivestreams.Publisher;
+import reactor.rx.Promise;
+import reactor.rx.Stream;
+import reactor.rx.Streams;
+
+import org.springframework.core.convert.TypeDescriptor;
+import org.springframework.core.convert.converter.GenericConverter;
+
+/**
+ * @author Stephane Maldini
+ * @author Sebastien Deleuze
+ */
+public final class ReactiveStreamsToReactorConverter implements GenericConverter {
+
+	@Override
+	public Set<GenericConverter.ConvertiblePair> getConvertibleTypes() {
+		Set<GenericConverter.ConvertiblePair> convertibleTypes = new LinkedHashSet<>();
+		convertibleTypes.add(new GenericConverter.ConvertiblePair(Publisher.class, Stream.class));
+		convertibleTypes.add(new GenericConverter.ConvertiblePair(Stream.class, Publisher.class));
+		convertibleTypes.add(new GenericConverter.ConvertiblePair(Publisher.class, Promise.class));
+		convertibleTypes.add(new GenericConverter.ConvertiblePair(Promise.class, Publisher.class));
+		return convertibleTypes;
+	}
+
+	@Override
+	public Object convert(Object source, TypeDescriptor sourceType, TypeDescriptor targetType) {
+		if (source != null) {
+			if (Stream.class.isAssignableFrom(source.getClass())) {
+				return source;
+			} else if (Stream.class.isAssignableFrom(targetType.getResolvableType().getRawClass())) {
+				return Streams.wrap((Publisher)source);
+			} else if (Promise.class.isAssignableFrom(source.getClass())) {
+				return ((Promise<?>)source);
+			}  else if (Promise.class.isAssignableFrom(targetType.getResolvableType().getRawClass())) {
+				return Streams.wrap((Publisher)source).next();
+			}
+		}
+		return null;
+	}
+
+}

spring-web-reactive/src/main/java/org/springframework/core/convert/support/ReactiveStreamsToRxJava1Converter.java

@@ -0,0 +1,65 @@
+/*
+ * Copyright (c) 2011-2015 Pivotal Software Inc, All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.core.convert.support;
+
+import java.util.LinkedHashSet;
+import java.util.Set;
+
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.convert.RxJava1Converter;
+import rx.Observable;
+import rx.Single;
+
+import org.springframework.core.convert.TypeDescriptor;
+import org.springframework.core.convert.converter.GenericConverter;
+
+/**
+ * TODO Avoid classpath exception for older RxJava1 version without Single type
+ * @author Stephane Maldini
+ * @author Sebastien Deleuze
+ */
+public final class ReactiveStreamsToRxJava1Converter implements GenericConverter {
+
+	@Override
+	public Set<GenericConverter.ConvertiblePair> getConvertibleTypes() {
+		Set<GenericConverter.ConvertiblePair> convertibleTypes = new LinkedHashSet<>();
+		convertibleTypes.add(new GenericConverter.ConvertiblePair(Publisher.class, Observable.class));
+		convertibleTypes.add(new GenericConverter.ConvertiblePair(Observable.class, Publisher.class));
+		convertibleTypes.add(new GenericConverter.ConvertiblePair(Publisher.class, Single.class));
+		convertibleTypes.add(new GenericConverter.ConvertiblePair(Single.class, Publisher.class));
+		return convertibleTypes;
+	}
+
+	@Override
+	public Object convert(Object source, TypeDescriptor sourceType, TypeDescriptor targetType) {
+		if (source != null) {
+			if (Observable.class.isAssignableFrom(source.getClass())) {
+				return RxJava1Converter.from((Observable) source);
+			}
+			else if (Observable.class.isAssignableFrom(targetType.getResolvableType().getRawClass())) {
+				return RxJava1Converter.from((Publisher)source);
+			}
+			else if (Single.class.isAssignableFrom(source.getClass())) {
+				return reactor.core.publisher.convert.RxJava1SingleConverter.from((Single) source);
+			} else if (Single.class.isAssignableFrom(targetType.getResolvableType().getRawClass())) {
+				return reactor.core.publisher.convert.RxJava1SingleConverter.from((Publisher)source);
+			}
+		}
+		return null;
+	}
+
+}

spring-web-reactive/src/main/java/org/springframework/reactive/codec/decoder/ByteBufferDecoder.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright 2002-2015 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.reactive.codec.decoder;
+
+import java.nio.ByteBuffer;
+
+import org.reactivestreams.Publisher;
+
+import org.springframework.core.ResolvableType;
+import org.springframework.http.MediaType;
+
+/**
+ * @author Sebastien Deleuze
+ */
+public class ByteBufferDecoder implements ByteToMessageDecoder<ByteBuffer> {
+
+	@Override
+	public boolean canDecode(ResolvableType type, MediaType mediaType, Object... hints) {
+		return ByteBuffer.class.isAssignableFrom(type.getRawClass());
+	}
+
+	@Override
+	public Publisher<ByteBuffer> decode(Publisher<ByteBuffer> inputStream, ResolvableType type, MediaType mediaType, Object... hints) {
+		return inputStream;
+	}
+}

spring-web-reactive/src/main/java/org/springframework/reactive/codec/decoder/ByteToMessageDecoder.java

@@ -34,8 +34,7 @@ public interface ByteToMessageDecoder<T> {
 
 	/**
 	 * Indicate whether the given type and media type can be processed by this decoder.
-	 * @param type the (potentially generic) type to ultimately decode to.
-	 * Could be different from {@code T} type.
+	 * @param type the stream element type to ultimately decode to.
 	 * @param mediaType the media type to decode from.
 	 * Typically the value of a {@code Content-Type} header for HTTP request.
 	 * @param hints Additional information about how to do decode, optional.
@@ -46,8 +45,7 @@ public interface ByteToMessageDecoder<T> {
 	/**
 	 * Decode a bytes stream to a message stream.
 	 * @param inputStream the input stream that represent the whole object to decode.
-	 * @param type the (potentially generic) type to ultimately decode to.
-	 * Could be different from {@code T} type.
+	 * @param type the stream element type to ultimately decode to.
 	 * @param hints Additional information about how to do decode, optional.
 	 * @return the decoded message stream
 	 */

spring-web-reactive/src/main/java/org/springframework/reactive/codec/decoder/JsonObjectDecoder.java

@@ -23,22 +23,18 @@ import org.reactivestreams.Publisher;
 import org.springframework.core.ResolvableType;
 import org.springframework.http.MediaType;
 import org.springframework.reactive.codec.encoder.JsonObjectEncoder;
+
 import reactor.Publishers;
 import reactor.fn.Function;
-import reactor.rx.Promise;
-import rx.Observable;
 
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.List;
 
 /**
- * Decode an arbitrary split byte stream representing JSON objects to a bye stream
+ * Decode an arbitrary split byte stream representing JSON objects to a byte stream
  * where each chunk is a well-formed JSON object.
  *
- * If {@code Hints.STREAM_ARRAY_ELEMENTS} is enabled, each element of top level JSON array
- * will be streamed as an individual JSON object.
- *
  * This class does not do any real parsing or validation. A sequence of bytes is considered a JSON object/array
  * if it contains a matching number of opening and closing braces/brackets.
  *
@@ -90,8 +86,7 @@ public class JsonObjectDecoder implements ByteToMessageDecoder<ByteBuffer> {
 
 	@Override
 	public boolean canDecode(ResolvableType type, MediaType mediaType, Object... hints) {
-		return mediaType.isCompatibleWith(MediaType.APPLICATION_JSON) && !Promise.class.isAssignableFrom(type.getRawClass()) &&
-				(Observable.class.isAssignableFrom(type.getRawClass()) || Publisher.class.isAssignableFrom(type.getRawClass()));
+		return mediaType.isCompatibleWith(MediaType.APPLICATION_JSON);
 	}
 
 	@Override

spring-web-reactive/src/main/java/org/springframework/reactive/codec/decoder/StringDecoder.java

@@ -41,7 +41,8 @@ public class StringDecoder implements ByteToMessageDecoder<String> {
 
 	@Override
 	public boolean canDecode(ResolvableType type, MediaType mediaType, Object... hints) {
-		return mediaType.isCompatibleWith(MediaType.TEXT_PLAIN);
+		return mediaType.isCompatibleWith(MediaType.TEXT_PLAIN)
+				&& String.class.isAssignableFrom(type.getRawClass());
 	}
 
 	@Override

spring-web-reactive/src/main/java/org/springframework/reactive/codec/encoder/ByteBufferEncoder.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2002-2015 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.reactive.codec.encoder;
+
+import java.nio.ByteBuffer;
+
+import org.reactivestreams.Publisher;
+
+import org.springframework.core.ResolvableType;
+import org.springframework.http.MediaType;
+
+/**
+ * @author Sebastien Deleuze
+ */
+public class ByteBufferEncoder implements MessageToByteEncoder<ByteBuffer> {
+
+	@Override
+	public boolean canEncode(ResolvableType type, MediaType mediaType, Object... hints) {
+		return ByteBuffer.class.isAssignableFrom(type.getRawClass());
+	}
+
+	@Override
+	public Publisher<ByteBuffer> encode(Publisher<? extends ByteBuffer> messageStream, ResolvableType type, MediaType mediaType, Object... hints) {
+		return (Publisher<ByteBuffer>)messageStream;
+	}
+
+}

spring-web-reactive/src/main/java/org/springframework/reactive/codec/encoder/JsonObjectEncoder.java

@@ -21,21 +21,17 @@ import org.reactivestreams.Subscriber;
 import org.springframework.core.ResolvableType;
 import org.springframework.http.MediaType;
 import org.springframework.reactive.codec.decoder.JsonObjectDecoder;
-import org.springframework.util.ClassUtils;
 
 import reactor.core.subscriber.SubscriberBarrier;
-import reactor.io.buffer.Buffer;
-import reactor.rx.Promise;
-import rx.Observable;
 
 import java.nio.ByteBuffer;
-import java.util.Arrays;
 
 import static reactor.Publishers.*;
+import reactor.io.buffer.Buffer;
 
 /**
- * Encode a bye stream of individual JSON element to a byte stream representing a single
- * JSON array when {@code Hints.ENCODE_AS_ARRAY} is enabled.
+ * Encode a byte stream of individual JSON element to a byte stream representing a single
+ * JSON array when if it contains more than one element.
  *
  * @author Sebastien Deleuze
  * @author Stephane Maldini
@@ -44,57 +40,31 @@ import static reactor.Publishers.*;
  */
 public class JsonObjectEncoder implements MessageToByteEncoder<ByteBuffer> {
 
-	private static final boolean rxJava1Present =
-			ClassUtils.isPresent("rx.Observable", JsonObjectEncoder.class.getClassLoader());
-
-	private static final boolean reactorPresent =
-			ClassUtils.isPresent("reactor.rx.Promise", JsonObjectEncoder.class.getClassLoader());
-
-	final ByteBuffer START_ARRAY = ByteBuffer.wrap("[".getBytes());
-
-	final ByteBuffer END_ARRAY = ByteBuffer.wrap("]".getBytes());
-
-	final ByteBuffer COMMA = ByteBuffer.wrap(",".getBytes());
-
 
 	@Override
 	public boolean canEncode(ResolvableType type, MediaType mediaType, Object... hints) {
-		return mediaType.isCompatibleWith(MediaType.APPLICATION_JSON) &&
-				!(reactorPresent && Promise.class.isAssignableFrom(type.getRawClass())) &&
-				(rxJava1Present && Observable.class.isAssignableFrom(type.getRawClass())
-				|| Publisher.class.isAssignableFrom(type.getRawClass()));
+		return mediaType.isCompatibleWith(MediaType.APPLICATION_JSON);
 	}
 
 	@Override
-	public Publisher<ByteBuffer> encode(Publisher<? extends ByteBuffer> messageStream, ResolvableType type, MediaType
-	  mediaType, Object... hints) {
-		//TODO Merge some chunks, there is no need to have chunks with only '[', ']' or ',' characters
-		return
-		  concat(
-			  from(
-			    Arrays.<Publisher<ByteBuffer>>asList(
-				  just(START_ARRAY),
-				  lift(
-				    flatMap(messageStream, (ByteBuffer b) -> from(Arrays.asList(b, COMMA))),
-				    sub -> new SkipLastBarrier(sub)
-				  ),
-				  just(END_ARRAY)
-			    )
-			  )
-		  );
+	public Publisher<ByteBuffer> encode(Publisher<? extends ByteBuffer> messageStream,
+			ResolvableType type, MediaType mediaType, Object... hints) {
+		return lift(messageStream, sub -> new JsonEncoderBarrier(sub));
 	}
 
-	private static class SkipLastBarrier extends SubscriberBarrier<ByteBuffer, ByteBuffer> {
+	private static class JsonEncoderBarrier extends SubscriberBarrier<ByteBuffer, ByteBuffer> {
 
-		public SkipLastBarrier(Subscriber<? super ByteBuffer> subscriber) {
+		public JsonEncoderBarrier(Subscriber<? super ByteBuffer> subscriber) {
 			super(subscriber);
 		}
 
 		ByteBuffer prev = null;
+		long count = 0;
 
 		@Override
 		protected void doNext(ByteBuffer next) {
-			if (prev == null) {
+			count++;
+			if (count == 1) {
 				prev = next;
 				doRequest(1);
 				return;
@@ -102,8 +72,27 @@ public class JsonObjectEncoder implements MessageToByteEncoder<ByteBuffer> {
 
 			ByteBuffer tmp = prev;
 			prev = next;
-			subscriber.onNext(tmp);
+			Buffer buffer = new Buffer();
+			if (count == 2) {
+				buffer.append("[");
+			}
+			buffer.append(tmp);
+			buffer.append(",");
+			buffer.flip();
+			subscriber.onNext(buffer.byteBuffer());
 		}
 
+		@Override
+		protected void doComplete() {
+			Buffer buffer = new Buffer();
+			buffer.append(prev);
+			if (count > 1) {
+				buffer.append("]");
+			}
+			buffer.flip();
+			subscriber.onNext(buffer.byteBuffer());
+			subscriber.onComplete();
+		}
 	}
+
 }

spring-web-reactive/src/main/java/org/springframework/reactive/codec/encoder/MessageToByteEncoder.java

@@ -34,8 +34,7 @@ public interface MessageToByteEncoder<T> {
 
 	/**
 	 * Indicate whether the given type and media type can be processed by this encoder.
-	 * @param type the (potentially generic) type to ultimately encode from.
-	 * Could be different from {@code T} type.
+	 * @param type the stream element type to encode.
 	 * @param mediaType the media type to encode.
 	 * Typically the value of an {@code Accept} header for HTTP request.
 	 * @param hints Additional information about how to encode, optional.
@@ -46,8 +45,7 @@ public interface MessageToByteEncoder<T> {
 	/**
 	 * Encode a given message stream to the given output byte stream.
 	 * @param messageStream the message stream to encode.
-	 * @param type the (potentially generic) type to ultimately encode from.
-	 * Could be different from {@code T} type.
+	 * @param type the stream element type to encode.
 	 * @param mediaType the media type to encode.
 	 * Typically the value of an {@code Accept} header for HTTP request.
 	 * @param hints Additional information about how to encode, optional.

spring-web-reactive/src/main/java/org/springframework/reactive/codec/encoder/StringEncoder.java

@@ -40,7 +40,8 @@ public class StringEncoder implements MessageToByteEncoder<String> {
 
 	@Override
 	public boolean canEncode(ResolvableType type, MediaType mediaType, Object... hints) {
-		return mediaType.isCompatibleWith(MediaType.TEXT_PLAIN);
+		return mediaType.isCompatibleWith(MediaType.TEXT_PLAIN)
+				&& String.class.isAssignableFrom(type.getRawClass());
 	}
 
 	@Override

spring-web-reactive/src/main/java/org/springframework/reactive/web/dispatch/DispatcherHandler.java

@@ -101,6 +101,7 @@ public class DispatcherHandler implements HttpHandler, ApplicationContextAware {
 		if (handler == null) {
 			// No exception handling mechanism yet
 			response.setStatusCode(HttpStatus.NOT_FOUND);
+			response.writeHeaders();
 			return Publishers.empty();
 		}
 

spring-web-reactive/src/main/java/org/springframework/reactive/web/dispatch/SimpleHandlerResultHandler.java

@@ -16,6 +16,8 @@
 
 package org.springframework.reactive.web.dispatch;
 
+import java.util.Arrays;
+
 import org.reactivestreams.Publisher;
 import reactor.Publishers;
 
@@ -45,6 +47,8 @@ public class SimpleHandlerResultHandler implements Ordered, HandlerResultHandler
 
 	@Override
 	public Publisher<Void> handleResult(ServerHttpRequest request, ServerHttpResponse response, HandlerResult result) {
-		return Publishers.completable((Publisher<?>)result.getValue());
+		Publisher<Void> handleComplete = Publishers.completable((Publisher<?>)result.getValue());
+		return Publishers.concat(Publishers.from(Arrays.asList(handleComplete, response.writeHeaders())));
 	}
+
 }

spring-web-reactive/src/main/java/org/springframework/reactive/web/dispatch/method/annotation/DefaultConversionService.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright (c) 2011-2015 Pivotal Software Inc, All Rights Reserved.
+ *
+ * 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.reactive.web.dispatch.method.annotation;
+
+import reactor.core.publisher.convert.DependencyUtils;
+
+import org.springframework.core.convert.converter.ConverterRegistry;
+import org.springframework.core.convert.support.GenericConversionService;
+import org.springframework.core.convert.support.ReactiveStreamsToCompletableFutureConverter;
+import org.springframework.core.convert.support.ReactiveStreamsToReactorConverter;
+import org.springframework.core.convert.support.ReactiveStreamsToRxJava1Converter;
+
+/**
+ * TODO temporary class designed to be replaced by org.springframework.core.convert.support.DefaultConversionService when it will contain Reactive Streams converter
+ * @author Sebastien Deleuze
+ */
+class DefaultConversionService extends GenericConversionService {
+
+	public DefaultConversionService() {
+		addDefaultConverters(this);
+	}
+
+	public static void addDefaultConverters(ConverterRegistry converterRegistry) {
+		converterRegistry.addConverter(new ReactiveStreamsToCompletableFutureConverter());
+		if (DependencyUtils.hasReactorStream()) {
+			converterRegistry.addConverter(new ReactiveStreamsToReactorConverter());
+		}
+		if (DependencyUtils.hasRxJava1()) {
+			converterRegistry.addConverter(new ReactiveStreamsToRxJava1Converter());
+		}
+	}
+
+}

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

@@ -16,27 +16,10 @@
 
 package org.springframework.reactive.web.dispatch.method.annotation;
 
-import java.nio.ByteBuffer;
-import java.nio.charset.Charset;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.TimeUnit;
-
 import org.reactivestreams.Publisher;
-import reactor.Publishers;
-import reactor.core.publisher.convert.CompletableFutureConverter;
-import reactor.core.publisher.convert.RxJava1Converter;
-import reactor.core.publisher.convert.RxJava1SingleConverter;
-import reactor.rx.Promise;
-import reactor.rx.Stream;
-import reactor.rx.Streams;
-import rx.Observable;
-import rx.Single;
-
 import org.springframework.core.MethodParameter;
 import org.springframework.core.ResolvableType;
+import org.springframework.core.convert.ConversionService;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.MediaType;
 import org.springframework.reactive.codec.decoder.ByteToMessageDecoder;
@@ -44,8 +27,15 @@ import org.springframework.reactive.web.dispatch.method.HandlerMethodArgumentRes
 import org.springframework.reactive.web.http.ServerHttpRequest;
 import org.springframework.web.bind.annotation.RequestBody;
 
+import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
 /**
  * @author Sebastien Deleuze
+ * @author Stephane Maldini
  */
 public class RequestBodyArgumentResolver implements HandlerMethodArgumentResolver {
 
@@ -53,14 +43,19 @@ public class RequestBodyArgumentResolver implements HandlerMethodArgumentResolve
 
 	private final List<ByteToMessageDecoder<?>> deserializers;
 	private final List<ByteToMessageDecoder<ByteBuffer>> preProcessors;
+	private final ConversionService conversionService;
 
 
-	public RequestBodyArgumentResolver(List<ByteToMessageDecoder<?>> deserializers) {
-		this(deserializers, Collections.EMPTY_LIST);
+	public RequestBodyArgumentResolver(List<ByteToMessageDecoder<?>> deserializers,
+			ConversionService conversionService) {
+		this(deserializers, conversionService, Collections.EMPTY_LIST);
 	}
 
-	public RequestBodyArgumentResolver(List<ByteToMessageDecoder<?>> deserializers, List<ByteToMessageDecoder<ByteBuffer>> preProcessors) {
+	public RequestBodyArgumentResolver(List<ByteToMessageDecoder<?>> deserializers,
+			ConversionService conversionService,
+			List<ByteToMessageDecoder<ByteBuffer>> preProcessors) {
 		this.deserializers = deserializers;
+		this.conversionService = conversionService;
 		this.preProcessors = preProcessors;
 	}
 
@@ -70,61 +65,31 @@ public class RequestBodyArgumentResolver implements HandlerMethodArgumentResolve
 	}
 
 	@Override
+	@SuppressWarnings("unchecked")
 	public Object resolveArgument(MethodParameter parameter, ServerHttpRequest request) {
 
 		MediaType mediaType = resolveMediaType(request);
 		ResolvableType type = ResolvableType.forMethodParameter(parameter);
 		List<Object> hints = new ArrayList<>();
 		hints.add(UTF_8);
-
-		// TODO: Refactor type conversion
-		ResolvableType readType = type;
-		if (Observable.class.isAssignableFrom(type.getRawClass()) ||
-				Single.class.isAssignableFrom(type.getRawClass()) ||
-				Promise.class.isAssignableFrom(type.getRawClass()) ||
-				Publisher.class.isAssignableFrom(type.getRawClass()) ||
-				CompletableFuture.class.isAssignableFrom(type.getRawClass())) {
-			readType = type.getGeneric(0);
-		}
-
-		ByteToMessageDecoder<?> deserializer = resolveDeserializers(request, type, mediaType, hints.toArray());
+		Publisher<ByteBuffer> inputStream = request.getBody();
+		Publisher<?> elementStream = inputStream;
+		ResolvableType elementType = type.hasGenerics() ? type.getGeneric(0) : type;
+		ByteToMessageDecoder<?> deserializer = resolveDeserializers(request, elementType, mediaType, hints.toArray());
 		if (deserializer != null) {
-
-			Publisher<ByteBuffer> inputStream = request.getBody();
-			List<ByteToMessageDecoder<ByteBuffer>> preProcessors = resolvePreProcessors(request, type, mediaType, hints.toArray());
+			List<ByteToMessageDecoder<ByteBuffer>> preProcessors =
+					resolvePreProcessors(request, elementType, mediaType,hints.toArray());
 			for (ByteToMessageDecoder<ByteBuffer> preProcessor : preProcessors) {
-				inputStream = preProcessor.decode(inputStream, type, mediaType, hints.toArray());
-			}
-			Publisher<?> elementStream = deserializer.decode(inputStream, readType, mediaType, UTF_8);
-
-			// TODO: Refactor type conversion
-			if (Stream.class.isAssignableFrom(type.getRawClass())) {
-				return Streams.wrap(elementStream);
-			}
-			else if (Promise.class.isAssignableFrom(type.getRawClass())) {
-				return Streams.wrap(elementStream).take(1).next();
-			}
-			else if (Observable.class.isAssignableFrom(type.getRawClass())) {
-				return RxJava1Converter.from(elementStream);
-			}
-			else if (Single.class.isAssignableFrom(type.getRawClass())) {
-				return RxJava1SingleConverter.from(elementStream);
-			}
-			else if (CompletableFuture.class.isAssignableFrom(type.getRawClass())) {
-				return CompletableFutureConverter.fromSingle(elementStream);
-			}
-			else if (Publisher.class.isAssignableFrom(type.getRawClass())) {
-				return elementStream;
-			}
-			else {
-				try {
-					return Publishers.toReadQueue(elementStream, 1, true).poll(30, TimeUnit.SECONDS);
-				} catch(InterruptedException ex) {
-					return Publishers.error(new IllegalStateException("Timeout before getter the value"));
-				}
+				inputStream = preProcessor.decode(inputStream, elementType, mediaType, hints.toArray());
 			}
+			elementStream = deserializer.decode(inputStream, elementType, mediaType, hints.toArray());
+		}
+		if (conversionService.canConvert(Publisher.class, type.getRawClass())) {
+			return conversionService.convert(elementStream, type.getRawClass());
+		}
+		else {
+			return elementStream;
 		}
-		return Publishers.error(new IllegalStateException("Argument type not supported: " + type));
 	}
 
 	private MediaType resolveMediaType(ServerHttpRequest request) {

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

@@ -15,11 +15,14 @@
  */
 package org.springframework.reactive.web.dispatch.method.annotation;
 
+import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 
 import org.springframework.beans.factory.InitializingBean;
+import org.springframework.reactive.codec.decoder.ByteBufferDecoder;
+import org.springframework.reactive.codec.decoder.ByteToMessageDecoder;
 import org.springframework.reactive.codec.decoder.JacksonJsonDecoder;
 import org.springframework.reactive.codec.decoder.JsonObjectDecoder;
 import org.springframework.reactive.codec.decoder.StringDecoder;
@@ -51,7 +54,11 @@ public class RequestMappingHandlerAdapter implements HandlerAdapter, Initializin
 		if (this.argumentResolvers == null) {
 			this.argumentResolvers = new ArrayList<>();
 			this.argumentResolvers.add(new RequestParamArgumentResolver());
-			this.argumentResolvers.add(new RequestBodyArgumentResolver(Arrays.asList(new StringDecoder(), new JacksonJsonDecoder()), Arrays.asList(new JsonObjectDecoder(true))));
+			List<ByteToMessageDecoder<?>> deserializers = Arrays.asList(new ByteBufferDecoder(),
+					new StringDecoder(), new JacksonJsonDecoder());
+			List<ByteToMessageDecoder<ByteBuffer>> preProcessors = Arrays.asList(new JsonObjectDecoder());
+			this.argumentResolvers.add(new RequestBodyArgumentResolver(deserializers,
+					new DefaultConversionService(), preProcessors));
 		}
 	}
 

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

@@ -18,9 +18,9 @@ package org.springframework.reactive.web.dispatch.method.annotation;
 import org.reactivestreams.Publisher;
 import org.springframework.core.MethodParameter;
 import org.springframework.core.Ordered;
-import org.springframework.core.ParameterizedTypeReference;
 import org.springframework.core.ResolvableType;
 import org.springframework.core.annotation.AnnotatedElementUtils;
+import org.springframework.core.convert.ConversionService;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.MediaType;
 import org.springframework.reactive.codec.encoder.MessageToByteEncoder;
@@ -31,26 +31,20 @@ import org.springframework.reactive.web.http.ServerHttpResponse;
 import org.springframework.web.bind.annotation.ResponseBody;
 import org.springframework.web.method.HandlerMethod;
 import reactor.Publishers;
-import reactor.core.publisher.convert.CompletableFutureConverter;
-import reactor.core.publisher.convert.RxJava1Converter;
-import reactor.core.publisher.convert.RxJava1SingleConverter;
-import reactor.rx.Promise;
-import rx.Observable;
-import rx.Single;
 
-import java.lang.reflect.Type;
 import java.nio.ByteBuffer;
 import java.nio.charset.Charset;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
-import java.util.concurrent.CompletableFuture;
 
 
 /**
  * First version using {@link MessageToByteEncoder}s
  *
  * @author Rossen Stoyanchev
+ * @author Stephane Maldini
+ * @author Sebastien Deleuze
  */
 public class ResponseBodyResultHandler implements HandlerResultHandler, Ordered {
 
@@ -59,6 +53,7 @@ public class ResponseBodyResultHandler implements HandlerResultHandler, Ordered
 
 	private final List<MessageToByteEncoder<?>> serializers;
 	private final List<MessageToByteEncoder<ByteBuffer>> postProcessors;
+	private final ConversionService conversionService;
 
 	private int order = 0;
 
@@ -68,8 +63,14 @@ public class ResponseBodyResultHandler implements HandlerResultHandler, Ordered
 	}
 
 	public ResponseBodyResultHandler(List<MessageToByteEncoder<?>> serializers, List<MessageToByteEncoder<ByteBuffer>> postProcessors) {
+		this(serializers, postProcessors, new DefaultConversionService());
+	}
+
+	public ResponseBodyResultHandler(List<MessageToByteEncoder<?>> serializers, List<MessageToByteEncoder<ByteBuffer>>
+	  postProcessors, ConversionService conversionService) {
 		this.serializers = serializers;
 		this.postProcessors = postProcessors;
+		this.conversionService = conversionService;
 	}
 
 	public void setOrder(int order) {
@@ -87,14 +88,13 @@ public class ResponseBodyResultHandler implements HandlerResultHandler, Ordered
 		Object handler = result.getHandler();
 		if (handler instanceof HandlerMethod) {
 			HandlerMethod handlerMethod = (HandlerMethod) handler;
-			Type publisherVoidType = new ParameterizedTypeReference<Publisher<Void>>(){}.getType();
-			return AnnotatedElementUtils.isAnnotated(handlerMethod.getMethod(), ResponseBody.class.getName()) &&
-					!handlerMethod.getReturnType().getGenericParameterType().equals(publisherVoidType);
+			return AnnotatedElementUtils.isAnnotated(handlerMethod.getMethod(), ResponseBody.class.getName());
 		}
 		return false;
 	}
 
 	@Override
+	@SuppressWarnings("unchecked")
 	public Publisher<Void> handleResult(ServerHttpRequest request, ServerHttpResponse response,
 			HandlerResult result) {
 
@@ -106,38 +106,27 @@ public class ResponseBodyResultHandler implements HandlerResultHandler, Ordered
 			return Publishers.empty();
 		}
 
-		MediaType mediaType = resolveMediaType(request);
 		ResolvableType type = ResolvableType.forMethodParameter(returnType);
+		MediaType mediaType = resolveMediaType(request);
 		List<Object> hints = new ArrayList<>();
 		hints.add(UTF_8);
-		MessageToByteEncoder<Object> serializer = (MessageToByteEncoder<Object>)resolveSerializer(request, type, mediaType, hints.toArray());
+		Publisher<Object> elementStream;
+		ResolvableType elementType;
+		if (conversionService.canConvert(type.getRawClass(), Publisher.class)) {
+			elementStream = conversionService.convert(value, Publisher.class);
+			elementType = type.getGeneric(0);
+		}
+		else {
+			elementStream = Publishers.just(value);
+			elementType = type;
+		}
+
+		MessageToByteEncoder<Object> serializer = (MessageToByteEncoder<Object>) resolveSerializer(request, elementType, mediaType, hints.toArray());
 		if (serializer != null) {
-			Publisher<Object> elementStream;
-
-			// TODO: Refactor type conversion
-			if (Promise.class.isAssignableFrom(type.getRawClass())) {
-				elementStream = ((Promise)value).stream();
-			}
-			else if (Observable.class.isAssignableFrom(type.getRawClass())) {
-				elementStream = RxJava1Converter.from((Observable) value);
-			}
-			else if (Single.class.isAssignableFrom(type.getRawClass())) {
-				elementStream = RxJava1SingleConverter.from((Single)value);
-			}
-			else if (CompletableFuture.class.isAssignableFrom(type.getRawClass())) {
-				elementStream = CompletableFutureConverter.from((CompletableFuture) value);
-			}
-			else if (Publisher.class.isAssignableFrom(type.getRawClass())) {
-				elementStream = (Publisher)value;
-			}
-			else {
-				elementStream = Publishers.just(value);
-			}
-
 			Publisher<ByteBuffer> outputStream = serializer.encode(elementStream, type, mediaType, hints.toArray());
-			List<MessageToByteEncoder<ByteBuffer>> postProcessors = resolvePostProcessors(request, type, mediaType, hints.toArray());
+			List<MessageToByteEncoder<ByteBuffer>> postProcessors = resolvePostProcessors(request, elementType, mediaType, hints.toArray());
 			for (MessageToByteEncoder<ByteBuffer> postProcessor : postProcessors) {
-				outputStream = postProcessor.encode(outputStream, type, mediaType, hints.toArray());
+				outputStream = postProcessor.encode(outputStream, elementType, mediaType, hints.toArray());
 			}
 			response.getHeaders().setContentType(mediaType);
 			return response.writeWith(outputStream);

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

@@ -30,9 +30,18 @@ public interface ServerHttpResponse extends HttpMessage {
 
 	void setStatusCode(HttpStatus status);
 
+	/**
+	 * Write the response headers. This method must be invoked to send responses without body.
+	 * @return A {@code Publisher<Void>} used to signal the demand, and receive a notification
+	 * when the handling is complete (success or error) including the flush of the data on the
+	 * network.
+	 */
+	Publisher<Void> writeHeaders();
+
 	/**
 	 * Write the provided reactive stream of bytes to the response body. Most servers
-	 * support multiple {@code writeWith} calls.
+	 * support multiple {@code writeWith} calls. Headers are written automatically
+	 * before the body, so not need to call {@link #writeHeaders()} explicitly.
 	 * @param contentPublisher the stream to write in the response body.
 	 * @return A {@code Publisher<Void>} used to signal the demand, and receive a notification
 	 * when the handling is complete (success or error) including the flush of the data on the

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

@@ -15,13 +15,13 @@
  */
 package org.springframework.reactive.web.http.reactor;
 
-import org.reactivestreams.Publisher;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpMethod;
 import org.springframework.reactive.web.http.ServerHttpRequest;
 import org.springframework.util.Assert;
 import reactor.io.buffer.Buffer;
 import reactor.io.net.http.HttpChannel;
+import reactor.rx.Stream;
 
 import java.net.URI;
 import java.net.URISyntaxException;
@@ -72,7 +72,7 @@ public class ReactorServerHttpRequest implements ServerHttpRequest {
 	}
 
 	@Override
-	public Publisher<ByteBuffer> getBody() {
+	public Stream<ByteBuffer> getBody() {
 		return this.channel.map(Buffer::byteBuffer);
 	}
 

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

@@ -24,6 +24,7 @@ import reactor.Publishers;
 import reactor.io.buffer.Buffer;
 import reactor.io.net.http.HttpChannel;
 import reactor.io.net.http.model.Status;
+import reactor.rx.Stream;
 
 import java.nio.ByteBuffer;
 
@@ -57,18 +58,28 @@ public class ReactorServerHttpResponse implements ServerHttpResponse {
 	}
 
 	@Override
-	public Publisher<Void> writeWith(Publisher<ByteBuffer> contentPublisher) {
-		writeHeaders();
+	public Publisher<Void> writeHeaders() {
+		if (this.headersWritten) {
+			return Publishers.empty();
+		}
+		applyHeaders();
+		return this.channel.writeHeaders();
+	}
+
+	@Override
+	public Stream<Void> writeWith(Publisher<ByteBuffer> contentPublisher) {
+		applyHeaders();
 		return this.channel.writeWith(Publishers.map(contentPublisher, Buffer::new));
 	}
 
-	private void writeHeaders() {
+	private void applyHeaders() {
 		if (!this.headersWritten) {
 			for (String name : this.headers.keySet()) {
 				for (String value : this.headers.get(name)) {
 					this.channel.responseHeaders().add(name, value);
 				}
 			}
+			this.headersWritten = true;
 		}
 	}
 }

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

@@ -23,6 +23,7 @@ import org.springframework.http.HttpStatus;
 import org.springframework.reactive.web.http.ServerHttpResponse;
 import org.springframework.util.Assert;
 
+import reactor.Publishers;
 import reactor.core.publisher.convert.RxJava1Converter;
 import reactor.io.buffer.Buffer;
 import rx.Observable;
@@ -59,24 +60,30 @@ public class RxNettyServerHttpResponse implements ServerHttpResponse {
 		return (this.headersWritten ? HttpHeaders.readOnlyHttpHeaders(this.headers) : this.headers);
 	}
 
-	public Observable<Void> writeWith(Observable<ByteBuffer> contentPublisher) {
-		return this.response.writeBytes(contentPublisher.map(content -> new Buffer(content).asBytes()));
+	@Override
+	public Publisher<Void> writeHeaders() {
+		if (this.headersWritten) {
+			return Publishers.empty();
+		}
+		applyHeaders();
+		return RxJava1Converter.from(this.response.sendHeaders());
 	}
 
 	@Override
 	public Publisher<Void> writeWith(Publisher<ByteBuffer> contentPublisher) {
-		writeHeaders();
+		applyHeaders();
 		Observable<byte[]> contentObservable = RxJava1Converter.from(contentPublisher).map(content -> new Buffer(content).asBytes());
 		return RxJava1Converter.from(this.response.writeBytes(contentObservable));
 	}
 
-	private void writeHeaders() {
+	private void applyHeaders() {
 		if (!this.headersWritten) {
 			for (String name : this.headers.keySet()) {
 				for (String value : this.headers.get(name)) {
 					this.response.addHeader(name, value);
 				}
 			}
+			this.headersWritten = true;
 		}
 	}
 }

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

@@ -22,6 +22,7 @@ import java.util.Map;
 import javax.servlet.http.HttpServletResponse;
 
 import org.reactivestreams.Publisher;
+import reactor.Publishers;
 
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpStatus;
@@ -61,13 +62,19 @@ public class ServletServerHttpResponse implements ServerHttpResponse {
 		return (this.headersWritten ? HttpHeaders.readOnlyHttpHeaders(this.headers) : this.headers);
 	}
 
+	@Override
+	public Publisher<Void> writeHeaders() {
+		applyHeaders();
+		return Publishers.empty();
+	}
+
 	@Override
 	public Publisher<Void> writeWith(final Publisher<ByteBuffer> contentPublisher) {
-		writeHeaders();
+		applyHeaders();
 		return (s -> contentPublisher.subscribe(responseSubscriber));
 	}
 
-	private void writeHeaders() {
+	private void applyHeaders() {
 		if (!this.headersWritten) {
 			for (Map.Entry<String, List<String>> entry : this.headers.entrySet()) {
 				String headerName = entry.getKey();