Add Codec support

This commit adds support for Publisher based codecs that allows to convert
byte stream to object stream and vice & versa.

Jackson, JAXB2 and String codec implementations are provided.

spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/JacksonJsonDecoderTests.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright 2002-2015 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.reactive.codec.decoder;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import org.junit.Test;
+import reactor.io.buffer.Buffer;
+import reactor.rx.Stream;
+import reactor.rx.Streams;
+
+import org.springframework.core.ResolvableType;
+import org.springframework.http.MediaType;
+import org.springframework.reactive.converter.Pojo;
+
+/**
+ * @author Sebastien Deleuze
+ */
+public class JacksonJsonDecoderTests {
+
+	private final JacksonJsonDecoder decoder = new JacksonJsonDecoder();
+
+	@Test
+	public void canDecode() {
+		assertTrue(decoder.canDecode(null, MediaType.APPLICATION_JSON));
+		assertFalse(decoder.canDecode(null, MediaType.APPLICATION_XML));
+	}
+
+	@Test
+	public void decode() throws InterruptedException {
+		Stream<ByteBuffer> source = Streams.just(Buffer.wrap("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}").byteBuffer());
+		List<Object> results = Streams.wrap(decoder.decode(source, ResolvableType.forClass(Pojo.class), null))
+				.toList().await();
+		assertEquals(1, results.size());
+		assertEquals("foofoo", ((Pojo) results.get(0)).getFoo());
+	}
+
+}

spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/Jaxb2DecoderTests.java

@@ -0,0 +1,57 @@
+/*
+ * 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 java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import org.junit.Test;
+import reactor.io.buffer.Buffer;
+import reactor.rx.Stream;
+import reactor.rx.Streams;
+
+import org.springframework.core.ResolvableType;
+import org.springframework.http.MediaType;
+import org.springframework.reactive.converter.Pojo;
+
+/**
+ * @author Sebastien Deleuze
+ */
+public class Jaxb2DecoderTests {
+
+	private final Jaxb2Decoder decoder = new Jaxb2Decoder();
+
+	@Test
+	public void canDecode() {
+		assertTrue(decoder.canDecode(null, MediaType.APPLICATION_XML));
+		assertTrue(decoder.canDecode(null, MediaType.TEXT_XML));
+		assertFalse(decoder.canDecode(null, MediaType.APPLICATION_JSON));
+	}
+
+	@Test
+	public void decode() throws InterruptedException {
+		Stream<ByteBuffer> source = Streams.just(Buffer.wrap("<?xml version=\"1.0\" encoding=\"UTF-8\"?><pojo><bar>barbar</bar><foo>foofoo</foo></pojo>").byteBuffer());
+		List<Object> results = Streams.wrap(decoder.decode(source, ResolvableType.forClass(Pojo.class), null))
+				.toList().await();
+		assertEquals(1, results.size());
+		assertEquals("foofoo", ((Pojo) results.get(0)).getFoo());
+	}
+
+}

spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/JsonObjectDecoderTests.java

@@ -0,0 +1,62 @@
+/*
+ * 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 java.nio.charset.StandardCharsets;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import org.junit.Test;
+import reactor.io.buffer.Buffer;
+import reactor.rx.Stream;
+import reactor.rx.Streams;
+
+/**
+ * @author Sebastien Deleuze
+ */
+public class JsonObjectDecoderTests {
+
+	@Test
+	public void decodeSingleChunkToArray() throws InterruptedException {
+		JsonObjectDecoder decoder = new JsonObjectDecoder(true);
+		Stream<ByteBuffer> source = Streams.just(Buffer.wrap("[{\"foo\": \"foofoo\", \"bar\": \"barbar\"},{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}]").byteBuffer());
+		List<String> results = Streams.wrap(decoder.decode(source, null, null)).map(chunk -> {
+					byte[] b = new byte[chunk.remaining()];
+					chunk.get(b);
+					return new String(b, StandardCharsets.UTF_8);
+				}).toList().await();
+		assertEquals(2, results.size());
+		assertEquals("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}", results.get(0));
+		assertEquals("{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}", results.get(1));
+	}
+
+	@Test
+	public void decodeMultipleChunksToArray() throws InterruptedException {
+		JsonObjectDecoder decoder = new JsonObjectDecoder(true);
+		Stream<ByteBuffer> source = Streams.just(Buffer.wrap("[{\"foo\": \"foofoo\", \"bar\"").byteBuffer(), Buffer.wrap(": \"barbar\"},{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}]").byteBuffer());
+		List<String> results = Streams.wrap(decoder.decode(source, null, null)).map(chunk -> {
+					byte[] b = new byte[chunk.remaining()];
+					chunk.get(b);
+					return new String(b, StandardCharsets.UTF_8);
+				}).toList().await();
+		assertEquals(2, results.size());
+		assertEquals("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}", results.get(0));
+		assertEquals("{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}", results.get(1));
+	}
+
+}

spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/StringDecoderTests.java

@@ -0,0 +1,57 @@
+/*
+ * 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 java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import org.junit.Test;
+import reactor.io.buffer.Buffer;
+import reactor.rx.Stream;
+import reactor.rx.Streams;
+
+import org.springframework.core.ResolvableType;
+import org.springframework.http.MediaType;
+import org.springframework.reactive.converter.Pojo;
+
+/**
+ * @author Sebastien Deleuze
+ */
+public class StringDecoderTests {
+
+	private final StringDecoder decoder = new StringDecoder();
+
+	@Test
+	public void canDecode() {
+		assertTrue(decoder.canDecode(null, MediaType.TEXT_PLAIN));
+		assertFalse(decoder.canDecode(null, MediaType.APPLICATION_JSON));
+	}
+
+	@Test
+	public void decode() throws InterruptedException {
+		Stream<ByteBuffer> source = Streams.just(Buffer.wrap("foo").byteBuffer(), Buffer.wrap("bar").byteBuffer());
+		List<String> results = Streams.wrap(decoder.decode(source, ResolvableType.forClass(Pojo.class), null))
+				.toList().await();
+		assertEquals(2, results.size());
+		assertEquals("foo", results.get(0));
+		assertEquals("bar", results.get(1));
+	}
+
+}

spring-web-reactive/src/test/java/org/springframework/reactive/codec/encoder/JacksonJsonEncoderTests.java

@@ -0,0 +1,58 @@
+/*
+ * 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.charset.StandardCharsets;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import org.junit.Test;
+import reactor.rx.Stream;
+import reactor.rx.Streams;
+
+import org.springframework.http.MediaType;
+import org.springframework.reactive.converter.Pojo;
+
+/**
+ * @author Sebastien Deleuze
+ */
+public class JacksonJsonEncoderTests {
+
+	private final JacksonJsonEncoder encoder = new JacksonJsonEncoder();
+
+	@Test
+	public void canWrite() {
+		assertTrue(encoder.canEncode(null, MediaType.APPLICATION_JSON));
+		assertFalse(encoder.canEncode(null, MediaType.APPLICATION_XML));
+	}
+
+	@Test
+	public void write() throws InterruptedException {
+		Stream<Pojo> source = Streams.just(new Pojo("foofoo", "barbar"), new Pojo("foofoofoo", "barbarbar"));
+		List<String> results = Streams.wrap(encoder.encode(source, null, null)).map(chunk -> {
+			byte[] b = new byte[chunk.remaining()];
+			chunk.get(b);
+			return new String(b, StandardCharsets.UTF_8);
+		}).toList().await();
+		assertEquals(2, results.size());
+		assertEquals("{\"foo\":\"foofoo\",\"bar\":\"barbar\"}", results.get(0));
+		assertEquals("{\"foo\":\"foofoofoo\",\"bar\":\"barbarbar\"}", results.get(1));
+	}
+
+}

spring-web-reactive/src/test/java/org/springframework/reactive/codec/encoder/Jaxb2EncoderTests.java

@@ -0,0 +1,60 @@
+/*
+ * 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.charset.StandardCharsets;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import org.junit.Test;
+import reactor.rx.Stream;
+import reactor.rx.Streams;
+
+import org.springframework.http.MediaType;
+import org.springframework.reactive.codec.decoder.Jaxb2Decoder;
+import org.springframework.reactive.converter.Pojo;
+
+/**
+ * @author Sebastien Deleuze
+ */
+public class Jaxb2EncoderTests {
+
+	private final Jaxb2Encoder encoder = new Jaxb2Encoder();
+
+	@Test
+	public void canEncode() {
+		assertTrue(encoder.canEncode(null, MediaType.APPLICATION_XML));
+		assertTrue(encoder.canEncode(null, MediaType.TEXT_XML));
+		assertFalse(encoder.canEncode(null, MediaType.APPLICATION_JSON));
+	}
+
+	@Test
+	public void encode() throws InterruptedException {
+		Stream<Pojo> source = Streams.just(new Pojo("foofoo", "barbar"), new Pojo("foofoofoo", "barbarbar"));
+		List<String> results = Streams.wrap(encoder.encode(source, null, null)).map(chunk -> {
+					byte[] b = new byte[chunk.remaining()];
+					chunk.get(b);
+					return new String(b, StandardCharsets.UTF_8);
+				}).toList().await();
+		assertEquals(2, results.size());
+		assertEquals("<?xml version=\"1.0\" encoding=\"UTF-8\" standalone=\"yes\"?><pojo><bar>barbar</bar><foo>foofoo</foo></pojo>", results.get(0));
+		assertEquals("<?xml version=\"1.0\" encoding=\"UTF-8\" standalone=\"yes\"?><pojo><bar>barbarbar</bar><foo>foofoofoo</foo></pojo>", results.get(1));
+	}
+
+}

spring-web-reactive/src/test/java/org/springframework/reactive/codec/encoder/JsonObjectEncoderTests.java

@@ -0,0 +1,47 @@
+/*
+ * 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 java.nio.charset.StandardCharsets;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import org.junit.Test;
+import reactor.io.buffer.Buffer;
+import reactor.rx.Stream;
+import reactor.rx.Streams;
+
+/**
+ * @author Sebastien Deleuze
+ */
+public class JsonObjectEncoderTests {
+
+	@Test
+	public void encodeToArray() throws InterruptedException {
+		JsonObjectEncoder encoder = new JsonObjectEncoder();
+		Stream<ByteBuffer> source = Streams.just(Buffer.wrap("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}").byteBuffer(), Buffer.wrap("{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}").byteBuffer());
+		List<String> results = Streams.wrap(encoder.encode(source, null, null)).map(chunk -> {
+			byte[] b = new byte[chunk.remaining()];
+			chunk.get(b);
+			return new String(b, StandardCharsets.UTF_8);
+		}).toList().await();
+		String result = String.join("", results);
+		assertEquals("[{\"foo\": \"foofoo\", \"bar\": \"barbar\"},{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}]", result);
+	}
+
+}

spring-web-reactive/src/test/java/org/springframework/reactive/codec/encoder/StringEncoderTests.java

@@ -0,0 +1,55 @@
+/*
+ * 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.charset.StandardCharsets;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import org.junit.Test;
+import reactor.rx.Streams;
+
+import org.springframework.http.MediaType;
+
+/**
+ * @author Sebastien Deleuze
+ */
+public class StringEncoderTests {
+
+	private final StringEncoder encoder = new StringEncoder();
+
+	@Test
+	public void canWrite() {
+		assertTrue(encoder.canEncode(null, MediaType.TEXT_PLAIN));
+		assertFalse(encoder.canEncode(null, MediaType.APPLICATION_JSON));
+	}
+
+	@Test
+	public void write() throws InterruptedException {
+		List<String> results = Streams.wrap(encoder.encode(Streams.just("foo"), null, null))
+				.map(chunk -> {
+					byte[] b = new byte[chunk.remaining()];
+					chunk.get(b);
+					return new String(b, StandardCharsets.UTF_8);
+				}).toList().await();
+		assertEquals(1, results.size());
+		assertEquals("foo", results.get(0));
+	}
+
+}

spring-web-reactive/src/test/java/org/springframework/reactive/converter/Pojo.java

@@ -0,0 +1,55 @@
+/*
+ * 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.converter;
+
+import javax.xml.bind.annotation.XmlRootElement;
+
+/**
+ * @author Sebastien Deleuze
+ */
+@XmlRootElement
+public class Pojo {
+
+	private String foo;
+
+	private String bar;
+
+	public Pojo() {
+	}
+
+	public Pojo(String foo, String bar) {
+		this.foo = foo;
+		this.bar = bar;
+	}
+
+	public String getFoo() {
+		return this.foo;
+	}
+
+	public void setFoo(String foo) {
+		this.foo = foo;
+	}
+
+	public String getBar() {
+		return this.bar;
+	}
+
+	public void setBar(String bar) {
+		this.bar = bar;
+	}
+
+}

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

@@ -17,41 +17,48 @@ package org.springframework.reactive.web.dispatch.method.annotation;
 
 
 import java.net.URI;
-import java.nio.charset.Charset;
+import java.util.Arrays;
+import java.util.List;
 
+import static org.junit.Assert.assertEquals;
 import org.junit.Test;
 import org.reactivestreams.Publisher;
+import reactor.rx.Stream;
 import reactor.rx.Streams;
+import rx.Observable;
 
+import org.springframework.core.ParameterizedTypeReference;
+import org.springframework.http.MediaType;
 import org.springframework.http.RequestEntity;
 import org.springframework.http.ResponseEntity;
+import org.springframework.reactive.codec.encoder.JacksonJsonEncoder;
+import org.springframework.reactive.codec.encoder.JsonObjectEncoder;
+import org.springframework.reactive.codec.encoder.StringEncoder;
 import org.springframework.reactive.web.dispatch.DispatcherHandler;
 import org.springframework.reactive.web.http.AbstractHttpHandlerIntegrationTests;
 import org.springframework.reactive.web.http.HttpHandler;
 import org.springframework.stereotype.Controller;
+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.client.RestTemplate;
 import org.springframework.web.context.support.StaticWebApplicationContext;
 
-import static org.junit.Assert.assertArrayEquals;
-
 /**
  * @author Rossen Stoyanchev
+ * @author Sebastien Deleuze
  */
 public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrationTests {
 
-	private static final Charset UTF_8 = Charset.forName("UTF-8");
-
-
 	@Override
 	protected HttpHandler createHttpHandler() {
 
 		StaticWebApplicationContext wac = new StaticWebApplicationContext();
 		wac.registerSingleton("handlerMapping", RequestMappingHandlerMapping.class);
 		wac.registerSingleton("handlerAdapter", RequestMappingHandlerAdapter.class);
-		wac.registerSingleton("responseBodyResultHandler", ResponseBodyResultHandler.class);
+		wac.getDefaultListableBeanFactory().registerSingleton("responseBodyResultHandler",
+				new ResponseBodyResultHandler(Arrays.asList(new StringEncoder(), new JacksonJsonEncoder()), Arrays.asList(new JsonObjectEncoder())));
 		wac.registerSingleton("controller", TestController.class);
 		wac.refresh();
 
@@ -67,9 +74,97 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati
 
 		URI url = new URI("http://localhost:" + port + "/param?name=George");
 		RequestEntity<Void> request = RequestEntity.get(url).build();
-		ResponseEntity<byte[]> response = restTemplate.exchange(request, byte[].class);
+		ResponseEntity<String> response = restTemplate.exchange(request, String.class);
 
-		assertArrayEquals("Hello George!".getBytes(UTF_8), response.getBody());
+		assertEquals("Hello George!", response.getBody());
+	}
+
+	@Test
+	public void serializeAsPojo() throws Exception {
+
+		RestTemplate restTemplate = new RestTemplate();
+
+		URI url = new URI("http://localhost:" + port + "/person");
+		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 {
+
+		RestTemplate restTemplate = new RestTemplate();
+
+		URI url = new URI("http://localhost:" + port + "/list");
+		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 serializeAsPublisher() 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();
+
+		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 echo() throws Exception {
+		RestTemplate restTemplate = new RestTemplate();
+
+		URI url = new URI("http://localhost:" + port + "/echo");
+		List<Person> persons = Arrays.asList(new Person("Robert"), new Person("Marie"));
+		RequestEntity<List<Person>> request = RequestEntity
+				.post(url)
+				.contentType(MediaType.APPLICATION_JSON)
+				.accept(MediaType.APPLICATION_JSON)
+				.body(persons);
+		List<Person> results = restTemplate.exchange(request, new ParameterizedTypeReference<List<Person>>(){}).getBody();
+
+		assertEquals(2, results.size());
+		assertEquals("Robert", results.get(0).getName());
+		assertEquals("Marie", results.get(1).getName());
 	}
 
 
@@ -82,6 +177,73 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati
 		public Publisher<String> handleWithParam(@RequestParam String name) {
 			return Streams.just("Hello ", name, "!");
 		}
+
+		@RequestMapping("/person")
+		@ResponseBody
+		public Person personResponseBody() {
+			return new Person("Robert");
+		}
+
+		@RequestMapping("/list")
+		@ResponseBody
+		public List<Person> listResponseBody() {
+			return Arrays.asList(new Person("Robert"), new Person("Marie"));
+		}
+
+		@RequestMapping("/publisher")
+		@ResponseBody
+		public Publisher<Person> publisherResponseBody() {
+			return Streams.just(new Person("Robert"), new Person("Marie"));
+		}
+
+		@RequestMapping("/observable")
+		@ResponseBody
+		public Observable<Person> observableResponseBody() {
+			return Observable.just(new Person("Robert"), new Person("Marie"));
+		}
+
+		@RequestMapping("/stream")
+		@ResponseBody
+		public Stream<Person> reactorStreamResponseBody() {
+			return Streams.just(new Person("Robert"), new Person("Marie"));
+		}
+
+		@RequestMapping("/echo")
+		@ResponseBody
+		public Publisher<Person> echo(@RequestBody Publisher<Person> persons) {
+			return persons;
+		}
+
+	}
+
+	private static class Person {
+
+		private String name;
+
+		public Person() {
+		}
+
+		public Person(String name) {
+			this.name = name;
+		}
+
+		public String getName() {
+			return name;
+		}
+
+		public void setName(String name) {
+			this.name = name;
+		}
+
+		@Override
+		public boolean equals(Object o) {
+			return name.equals(((Person)o).name);
+		}
+
+		@Override
+		public int hashCode() {
+			return name.hashCode();
+		}
 	}
 
 }