Add CompletableFuture/Single/Promise support

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

@@ -25,6 +25,7 @@ import io.netty.buffer.ByteBufUtil;
 import io.netty.buffer.Unpooled;
 import org.reactivestreams.Publisher;
 import reactor.fn.Function;
+import reactor.rx.Promise;
 import reactor.rx.Streams;
 import rx.Observable;
 
@@ -90,7 +91,7 @@ public class JsonObjectDecoder implements ByteToMessageDecoder<ByteBuffer> {
 
 	@Override
 	public boolean canDecode(ResolvableType type, MediaType mediaType, Object... hints) {
-		return mediaType.isCompatibleWith(MediaType.APPLICATION_JSON) &&
+		return mediaType.isCompatibleWith(MediaType.APPLICATION_JSON) && !Promise.class.isAssignableFrom(type.getRawClass()) &&
 				(Observable.class.isAssignableFrom(type.getRawClass()) || Publisher.class.isAssignableFrom(type.getRawClass()));
 	}
 

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

@@ -19,6 +19,7 @@ package org.springframework.reactive.codec.encoder;
 import java.nio.ByteBuffer;
 
 import org.reactivestreams.Publisher;
+import reactor.rx.Promise;
 import rx.Observable;
 import rx.RxReactiveStreams;
 
@@ -44,7 +45,7 @@ public class JsonObjectEncoder implements MessageToByteEncoder<ByteBuffer> {
 
 	@Override
 	public boolean canEncode(ResolvableType type, MediaType mediaType, Object... hints) {
-		return mediaType.isCompatibleWith(MediaType.APPLICATION_JSON) &&
+		return mediaType.isCompatibleWith(MediaType.APPLICATION_JSON) && !Promise.class.isAssignableFrom(type.getRawClass()) &&
 				(Observable.class.isAssignableFrom(type.getRawClass()) || Publisher.class.isAssignableFrom(type.getRawClass()));
 	}
 

spring-web-reactive/src/main/java/org/springframework/reactive/util/CompletableFutureUtils.java

@@ -0,0 +1,141 @@
+/*
+ * 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.util;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+import reactor.core.support.Exceptions;
+import reactor.rx.Stream;
+import reactor.rx.action.Action;
+import reactor.rx.subscription.ReactiveSubscription;
+
+import org.springframework.util.Assert;
+
+/**
+ * @author Sebastien Deleuze
+ */
+public class CompletableFutureUtils {
+
+	public static <T> Publisher<T> toPublisher(CompletableFuture<T> future) {
+		return new CompletableFutureStream<T>(future);
+	}
+
+	public static <T> CompletableFuture<List<T>> fromPublisher(Publisher<T> publisher) {
+		final CompletableFuture<List<T>> future = new CompletableFuture<>();
+		publisher.subscribe(new Subscriber<T>() {
+			private final List<T> values = new ArrayList<>();
+
+			@Override
+			public void onSubscribe(Subscription s) {
+				s.request(Long.MAX_VALUE);
+			}
+
+			@Override
+			public void onNext(T t) {
+				values.add(t);
+			}
+
+			@Override
+			public void onError(Throwable t) {
+				future.completeExceptionally(t);
+			}
+
+			@Override
+			public void onComplete() {
+				future.complete(values);
+			}
+		});
+		return future;
+	}
+
+	public static <T> CompletableFuture<T> fromSinglePublisher(Publisher<T> publisher) {
+		final CompletableFuture<T> future = new CompletableFuture<>();
+		publisher.subscribe(new Subscriber<T>() {
+			private T value;
+
+			@Override
+			public void onSubscribe(Subscription s) {
+				s.request(Long.MAX_VALUE);
+			}
+
+			@Override
+			public void onNext(T t) {
+				Assert.state(value == null, "This publisher should not publish multiple values");
+				value = t;
+			}
+
+			@Override
+			public void onError(Throwable t) {
+				future.completeExceptionally(t);
+			}
+
+			@Override
+			public void onComplete() {
+				future.complete(value);
+			}
+		});
+		return future;
+	}
+
+	private static class CompletableFutureStream<T> extends Stream<T> {
+
+		private final CompletableFuture<? extends T> future;
+
+		public CompletableFutureStream(CompletableFuture<? extends T> future) {
+			this.future = future;
+		}
+
+		@Override
+		public void subscribe(final Subscriber<? super T> subscriber) {
+			try {
+				subscriber.onSubscribe(new ReactiveSubscription<T>(this, subscriber) {
+
+					@Override
+					public void request(long elements) {
+						Action.checkRequest(elements);
+						if (isComplete()) return;
+
+						try {
+							future.whenComplete((result, error) -> {
+								if (error != null) {
+									onError(error);
+								}
+								else {
+									subscriber.onNext(result);
+									onComplete();
+								}
+							});
+
+						} catch (Throwable e) {
+							onError(e);
+						}
+					}
+				});
+			} catch (Throwable throwable) {
+				Exceptions.throwIfFatal(throwable);
+				subscriber.onError(throwable);
+			}
+		}
+
+	}
+
+}

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

@@ -21,18 +21,22 @@ import java.nio.charset.Charset;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
 
 import org.reactivestreams.Publisher;
+import reactor.rx.Promise;
 import reactor.rx.Stream;
 import reactor.rx.Streams;
 import rx.Observable;
 import rx.RxReactiveStreams;
+import rx.Single;
 
 import org.springframework.core.MethodParameter;
 import org.springframework.core.ResolvableType;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.MediaType;
 import org.springframework.reactive.codec.decoder.ByteToMessageDecoder;
+import org.springframework.reactive.util.CompletableFutureUtils;
 import org.springframework.reactive.web.dispatch.method.HandlerMethodArgumentResolver;
 import org.springframework.reactive.web.http.ServerHttpRequest;
 import org.springframework.web.bind.annotation.RequestBody;
@@ -69,9 +73,14 @@ public class RequestBodyArgumentResolver implements HandlerMethodArgumentResolve
 		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()) || Publisher.class.isAssignableFrom(type.getRawClass())) {
+		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);
 		}
 
@@ -89,9 +98,18 @@ public class RequestBodyArgumentResolver implements HandlerMethodArgumentResolve
 			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 RxReactiveStreams.toObservable(elementStream);
 			}
+			else if (Single.class.isAssignableFrom(type.getRawClass())) {
+				return RxReactiveStreams.toObservable(elementStream).toSingle();
+			}
+			else if (CompletableFuture.class.isAssignableFrom(type.getRawClass())) {
+				return CompletableFutureUtils.fromSinglePublisher(elementStream);
+			}
 			else if (Publisher.class.isAssignableFrom(type.getRawClass())) {
 				return elementStream;
 			}
@@ -99,11 +117,11 @@ public class RequestBodyArgumentResolver implements HandlerMethodArgumentResolve
 				try {
 					return Streams.wrap(elementStream).next().await();
 				} catch(InterruptedException ex) {
-					throw new IllegalStateException("Timeout before getter the value");
+					return Streams.fail(new IllegalStateException("Timeout before getter the value"));
 				}
 			}
 		}
-		throw new IllegalStateException("Argument type not supported: " + type);
+		return Streams.fail(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/ResponseBodyResultHandler.java

@@ -21,11 +21,15 @@ import java.nio.charset.Charset;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
 
 import org.reactivestreams.Publisher;
+import reactor.rx.Promise;
+import reactor.rx.Stream;
 import reactor.rx.Streams;
 import rx.Observable;
 import rx.RxReactiveStreams;
+import rx.Single;
 
 import org.springframework.core.MethodParameter;
 import org.springframework.core.Ordered;
@@ -34,6 +38,7 @@ import org.springframework.core.annotation.AnnotatedElementUtils;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.MediaType;
 import org.springframework.reactive.codec.encoder.MessageToByteEncoder;
+import org.springframework.reactive.util.CompletableFutureUtils;
 import org.springframework.reactive.web.dispatch.HandlerResult;
 import org.springframework.reactive.web.dispatch.HandlerResultHandler;
 import org.springframework.reactive.web.http.ServerHttpRequest;
@@ -108,9 +113,18 @@ public class ResponseBodyResultHandler implements HandlerResultHandler, Ordered
 			Publisher<Object> elementStream;
 
 			// TODO: Refactor type conversion
-			if (Observable.class.isAssignableFrom(type.getRawClass())) {
+			if (Promise.class.isAssignableFrom(type.getRawClass())) {
+				elementStream = ((Promise)value).stream();
+			}
+			else if (Observable.class.isAssignableFrom(type.getRawClass())) {
 				elementStream = RxReactiveStreams.toPublisher((Observable) value);
 			}
+			else if (Single.class.isAssignableFrom(type.getRawClass())) {
+				elementStream = RxReactiveStreams.toPublisher(((Single)value).toObservable());
+			}
+			else if (CompletableFuture.class.isAssignableFrom(type.getRawClass())) {
+				elementStream = CompletableFutureUtils.toPublisher((CompletableFuture) value);
+			}
 			else if (Publisher.class.isAssignableFrom(type.getRawClass())) {
 				elementStream = (Publisher)value;
 			}

spring-web-reactive/src/test/java/org/springframework/reactive/util/CompletableFutureUtilsTests.java

@@ -0,0 +1,101 @@
+/*
+ * 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.util;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import org.junit.Test;
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+import reactor.rx.Streams;
+
+/**
+ * @author Sebastien Deleuze
+ */
+public class CompletableFutureUtilsTests {
+
+	private CountDownLatch lock = new CountDownLatch(1);
+	private final List<Boolean> results = new ArrayList<>();
+	private final List<Throwable> errors = new ArrayList<>();
+
+	@Test
+	public void fromPublisher() throws InterruptedException {
+		Publisher<Boolean> publisher = Streams.just(true, false);
+		CompletableFuture<List<Boolean>> future = CompletableFutureUtils.fromPublisher(publisher);
+		future.whenComplete((result, error) -> {
+			if (error != null) {
+				errors.add(error);
+			}
+			else {
+				results.addAll(result);
+			}
+			lock.countDown();
+		});
+		lock.await(2000, TimeUnit.MILLISECONDS);
+		assertEquals("onError not expected: " + errors.toString(), 0, errors.size());
+		assertEquals(2, results.size());
+		assertTrue(results.get(0));
+		assertFalse(results.get(1));
+	}
+
+	@Test
+	public void fromSinglePublisher() throws InterruptedException {
+		Publisher<Boolean> publisher = Streams.just(true);
+		CompletableFuture<Boolean> future = CompletableFutureUtils.fromSinglePublisher(publisher);
+		future.whenComplete((result, error) -> {
+			if (error != null) {
+				errors.add(error);
+			}
+			else {
+				results.add(result);
+			}
+			lock.countDown();
+		});
+		lock.await(2000, TimeUnit.MILLISECONDS);
+		assertEquals("onError not expected: " + errors.toString(), 0, errors.size());
+		assertEquals(1, results.size());
+		assertTrue(results.get(0));
+	}
+
+	@Test
+	public void fromSinglePublisherWithMultipleValues() throws InterruptedException {
+		Publisher<Boolean> publisher = Streams.just(true, false);
+		CompletableFuture<Boolean> future = CompletableFutureUtils.fromSinglePublisher(publisher);
+		future.whenComplete((result, error) -> {
+			if (error != null) {
+				errors.add(error);
+			}
+			else {
+				results.add(result);
+			}
+			lock.countDown();
+		});
+		lock.await(2000, TimeUnit.MILLISECONDS);
+		assertEquals(1, errors.size());
+		assertEquals(IllegalStateException.class, errors.get(0).getClass());
+		assertEquals(0, results.size());
+	}
+
+}

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

@@ -19,13 +19,17 @@ package org.springframework.reactive.web.dispatch.method.annotation;
 import java.net.URI;
 import java.util.Arrays;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
 
 import static org.junit.Assert.assertEquals;
 import org.junit.Test;
 import org.reactivestreams.Publisher;
+import reactor.rx.Promise;
+import reactor.rx.Promises;
 import reactor.rx.Stream;
 import reactor.rx.Streams;
 import rx.Observable;
+import rx.Single;
 
 import org.springframework.core.ParameterizedTypeReference;
 import org.springframework.http.MediaType;
@@ -81,22 +85,89 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati
 
 	@Test
 	public void serializeAsPojo() throws Exception {
+		serializeAsPojo("http://localhost:" + port + "/person");
+	}
 
+	@Test
+	public void serializeAsCompletableFuture() throws Exception {
+		serializeAsPojo("http://localhost:" + port + "/completable-future");
+	}
+
+	@Test
+	public void serializeAsSingle() throws Exception {
+		serializeAsPojo("http://localhost:" + port + "/single");
+	}
+
+	@Test
+	public void serializeAsPromise() throws Exception {
+		serializeAsPojo("http://localhost:" + port + "/promise");
+	}
+
+	@Test
+	public void serializeAsList() throws Exception {
+		serializeAsCollection("http://localhost:" + port + "/list");
+	}
+
+	@Test
+	public void serializeAsPublisher() throws Exception {
+		serializeAsCollection("http://localhost:" + port + "/publisher");
+	}
+
+	@Test
+	public void serializeAsObservable() throws Exception {
+		serializeAsCollection("http://localhost:" + port + "/observable");
+	}
+
+	@Test
+	public void serializeAsReactorStream() throws Exception {
+		serializeAsCollection("http://localhost:" + port + "/stream");
+	}
+
+	@Test
+	public void publisherCapitalize() throws Exception {
+		capitalizeCollection("http://localhost:" + port + "/publisher-capitalize");
+	}
+
+	@Test
+	public void observableCapitalize() throws Exception {
+		capitalizeCollection("http://localhost:" + port + "/observable-capitalize");
+	}
+
+	@Test
+	public void streamCapitalize() throws Exception {
+		capitalizeCollection("http://localhost:" + port + "/stream-capitalize");
+	}
+
+	@Test
+	public void completableFutureCapitalize() throws Exception {
+		capitalizePojo("http://localhost:" + port + "/completable-future-capitalize");
+	}
+
+	@Test
+	public void singleCapitalize() throws Exception {
+		capitalizePojo("http://localhost:" + port + "/single-capitalize");
+	}
+
+	@Test
+	public void promiseCapitalize() throws Exception {
+		capitalizePojo("http://localhost:" + port + "/promise-capitalize");
+	}
+
+
+	public void serializeAsPojo(String requestUrl) throws Exception {
 		RestTemplate restTemplate = new RestTemplate();
 
-		URI url = new URI("http://localhost:" + port + "/person");
+		URI url = new URI(requestUrl);
 		RequestEntity<Void> request = RequestEntity.get(url).accept(MediaType.APPLICATION_JSON).build();
 		ResponseEntity<Person> response = restTemplate.exchange(request, Person.class);
 
 		assertEquals(new Person("Robert"), response.getBody());
 	}
 
-	@Test
-	public void serializeAsList() throws Exception {
-
+	public void serializeAsCollection(String requestUrl) throws Exception {
 		RestTemplate restTemplate = new RestTemplate();
 
-		URI url = new URI("http://localhost:" + port + "/list");
+		URI url = new URI(requestUrl);
 		RequestEntity<Void> request = RequestEntity.get(url).accept(MediaType.APPLICATION_JSON).build();
 		List<Person> results = restTemplate.exchange(request, new ParameterizedTypeReference<List<Person>>(){}).getBody();
 
@@ -105,67 +176,23 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati
 		assertEquals(new Person("Marie"), results.get(1));
 	}
 
-	@Test
-	public void serializeAsPublisher() throws Exception {
 
+	public void capitalizePojo(String requestUrl) throws Exception {
 		RestTemplate restTemplate = new RestTemplate();
 
-		URI url = new URI("http://localhost:" + port + "/publisher");
-		RequestEntity<Void> request = RequestEntity.get(url).accept(MediaType.APPLICATION_JSON).build();
-		List<Person> results = restTemplate.exchange(request, new ParameterizedTypeReference<List<Person>>(){}).getBody();
+		URI url = new URI(requestUrl);
+		RequestEntity<Person> request = RequestEntity
+				.post(url)
+				.contentType(MediaType.APPLICATION_JSON)
+				.accept(MediaType.APPLICATION_JSON)
+				.body(new Person("Robert"));
+		ResponseEntity<Person> response = restTemplate.exchange(request, Person.class);
 
-		assertEquals(2, results.size());
-		assertEquals(new Person("Robert"), results.get(0));
-		assertEquals(new Person("Marie"), results.get(1));
-	}
-
-	@Test
-	public void serializeAsObservable() throws Exception {
-
-		RestTemplate restTemplate = new RestTemplate();
-
-		URI url = new URI("http://localhost:" + port + "/observable");
-		RequestEntity<Void> request = RequestEntity.get(url).accept(MediaType.APPLICATION_JSON).build();
-		List<Person> results = restTemplate.exchange(request, new ParameterizedTypeReference<List<Person>>() {
-		}).getBody();
-
-		assertEquals(2, results.size());
-		assertEquals(new Person("Robert"), results.get(0));
-		assertEquals(new Person("Marie"), results.get(1));
-	}
-
-	@Test
-	public void serializeAsReactorStream() throws Exception {
-
-		RestTemplate restTemplate = new RestTemplate();
-
-		URI url = new URI("http://localhost:" + port + "/stream");
-		RequestEntity<Void> request = RequestEntity.get(url).accept(MediaType.APPLICATION_JSON).build();
-		List<Person> results = restTemplate.exchange(request, new ParameterizedTypeReference<List<Person>>() {
-		}).getBody();
-
-		assertEquals(2, results.size());
-		assertEquals(new Person("Robert"), results.get(0));
-		assertEquals(new Person("Marie"), results.get(1));
-	}
-
-	@Test
-	public void publisherCapitalize() throws Exception {
-		capitalize("http://localhost:" + port + "/publisher-capitalize");
-	}
-
-	@Test
-	public void observableCapitalize() throws Exception {
-		capitalize("http://localhost:" + port + "/observable-capitalize");
-	}
-
-	@Test
-	public void streamCapitalize() throws Exception {
-		capitalize("http://localhost:" + port + "/stream-capitalize");
+		assertEquals(new Person("ROBERT"), response.getBody());
 	}
 
 
-	public void capitalize(String requestUrl) throws Exception {
+	public void capitalizeCollection(String requestUrl) throws Exception {
 		RestTemplate restTemplate = new RestTemplate();
 
 		URI url = new URI(requestUrl);
@@ -199,6 +226,24 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati
 			return new Person("Robert");
 		}
 
+		@RequestMapping("/completable-future")
+		@ResponseBody
+		public CompletableFuture<Person> completableFutureResponseBody() {
+			return CompletableFuture.completedFuture(new Person("Robert"));
+		}
+
+		@RequestMapping("/single")
+		@ResponseBody
+		public Single<Person> singleResponseBody() {
+			return Single.just(new Person("Robert"));
+		}
+
+		@RequestMapping("/promise")
+		@ResponseBody
+		public Promise<Person> promiseResponseBody() {
+			return Promises.success(new Person("Robert"));
+		}
+
 		@RequestMapping("/list")
 		@ResponseBody
 		public List<Person> listResponseBody() {
@@ -250,6 +295,33 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati
 			});
 		}
 
+		@RequestMapping("/completable-future-capitalize")
+		@ResponseBody
+		public CompletableFuture<Person> completableFutureCapitalize(@RequestBody CompletableFuture<Person> personFuture) {
+			return personFuture.thenApply(person -> {
+				person.setName(person.getName().toUpperCase());
+				return person;
+			});
+		}
+
+		@RequestMapping("/single-capitalize")
+		@ResponseBody
+		public Single<Person> singleCapitalize(@RequestBody Single<Person> personFuture) {
+			return personFuture.map(person -> {
+				person.setName(person.getName().toUpperCase());
+				return person;
+			});
+		}
+
+		@RequestMapping("/promise-capitalize")
+		@ResponseBody
+		public Promise<Person> promiseCapitalize(@RequestBody Promise<Person> personFuture) {
+			return personFuture.map(person -> {
+				person.setName(person.getName().toUpperCase());
+				return person;
+			});
+		}
+
 	}
 
 	private static class Person {