Refactor AbstractEncoderTestCase

Refactor AbstractEncoderTestCase to resemble AbstractDecoderTestCase Issue: SPR-17449
2018-11-16 14:33:03 +01:00
parent 39ce989d1a
commit 539cfc24c6
10 changed files with 488 additions and 389 deletions
--- a/spring-web/src/test/java/org/springframework/http/codec/json/Jackson2JsonEncoderTests.java
+++ b/spring-web/src/test/java/org/springframework/http/codec/json/Jackson2JsonEncoderTests.java
@@ -20,8 +20,6 @@ import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.Map;
-import java.util.function.Consumer;
-import java.util.stream.Stream;

 import com.fasterxml.jackson.annotation.JsonTypeInfo;
 import com.fasterxml.jackson.annotation.JsonTypeName;
@@ -29,18 +27,15 @@ import com.fasterxml.jackson.databind.ObjectMapper;
 import org.junit.Test;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
-import reactor.test.StepVerifier;

 import org.springframework.core.ResolvableType;
 import org.springframework.core.codec.AbstractEncoderTestCase;
-import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferUtils;
 import org.springframework.http.MediaType;
 import org.springframework.http.codec.Pojo;
 import org.springframework.http.codec.ServerSentEvent;
 import org.springframework.util.MimeType;

-import static java.util.Collections.emptyMap;
 import static java.util.Collections.singletonMap;
 import static org.junit.Assert.*;
 import static org.springframework.http.MediaType.APPLICATION_JSON;
@@ -55,30 +50,14 @@ import static org.springframework.http.codec.json.JacksonViewBean.MyJacksonView3
 /**
 * @author Sebastien Deleuze
 */
-public class Jackson2JsonEncoderTests extends AbstractEncoderTestCase<Object, Jackson2JsonEncoder> {
+public class Jackson2JsonEncoderTests extends AbstractEncoderTestCase<Jackson2JsonEncoder> {


 	public Jackson2JsonEncoderTests() {
-		super(new Jackson2JsonEncoder(), ResolvableType.forClass(Pojo.class),
-				APPLICATION_STREAM_JSON, null);
+		super(new Jackson2JsonEncoder());
 	}

 	@Override
-	protected Flux<Object> input() {
-		return Flux.just(new Pojo("foo", "bar"),
-				new Pojo("foofoo", "barbar"),
-				new Pojo("foofoofoo", "barbarbar"));
-	}
-
-	@Override
-	protected Stream<Consumer<DataBuffer>> outputConsumers() {
-		return Stream.<Consumer<DataBuffer>>builder()
-				.add(resultConsumer("{\"foo\":\"foo\",\"bar\":\"bar\"}\n"))
-				.add(resultConsumer("{\"foo\":\"foofoo\",\"bar\":\"barbar\"}\n"))
-				.add(resultConsumer("{\"foo\":\"foofoofoo\",\"bar\":\"barbarbar\"}\n"))
-				.build();
-	}
-
 	@Test
 	public void canEncode() {
 		ResolvableType pojoType = ResolvableType.forClass(Pojo.class);
@@ -94,6 +73,22 @@ public class Jackson2JsonEncoderTests extends AbstractEncoderTestCase<Object, Ja
 		assertFalse(this.encoder.canEncode(ResolvableType.forClass(Object.class), APPLICATION_OCTET_STREAM));
 	}

+	@Override
+	public void encode() throws Exception {
+		Flux<Object> input = Flux.just(new Pojo("foo", "bar"),
+				new Pojo("foofoo", "barbar"),
+				new Pojo("foofoofoo", "barbarbar"));
+
+		testEncodeAll(input, ResolvableType.forClass(Pojo.class), step -> step
+				.consumeNextWith(expectString("{\"foo\":\"foo\",\"bar\":\"bar\"}\n"))
+				.consumeNextWith(expectString("{\"foo\":\"foofoo\",\"bar\":\"barbar\"}\n"))
+				.consumeNextWith(expectString("{\"foo\":\"foofoofoo\",\"bar\":\"barbarbar\"}\n"))
+				.verifyComplete(),
+				APPLICATION_STREAM_JSON, null);
+
+
+	}
+
 	@Test // SPR-15866
 	public void canEncodeWithCustomMimeType() {
 		MimeType textJavascript = new MimeType("text", "javascript", StandardCharsets.UTF_8);
@@ -121,33 +116,29 @@ public class Jackson2JsonEncoderTests extends AbstractEncoderTestCase<Object, Ja

 	@Test
 	public void encodeNonStream() {
-		Flux<Pojo> source = Flux.just(
+		Flux<Pojo> input = Flux.just(
 				new Pojo("foo", "bar"),
 				new Pojo("foofoo", "barbar"),
 				new Pojo("foofoofoo", "barbarbar")
 		);
-		ResolvableType type = ResolvableType.forClass(Pojo.class);
-		Flux<DataBuffer> output = this.encoder.encode(source, this.bufferFactory, type, null, emptyMap());

-		StepVerifier.create(output)
-				.consumeNextWith(resultConsumer("[" +
+		testEncode(input, Pojo.class, step -> step
+				.consumeNextWith(expectString("[" +
 						"{\"foo\":\"foo\",\"bar\":\"bar\"}," +
 						"{\"foo\":\"foofoo\",\"bar\":\"barbar\"}," +
 						"{\"foo\":\"foofoofoo\",\"bar\":\"barbarbar\"}]")
 						.andThen(DataBufferUtils::release))
-				.verifyComplete();
+				.verifyComplete());
 	}

 	@Test
 	public void encodeWithType() {
-		Flux<ParentClass> source = Flux.just(new Foo(), new Bar());
-		ResolvableType type = ResolvableType.forClass(ParentClass.class);
-		Flux<DataBuffer> output = this.encoder.encode(source, this.bufferFactory, type, null, emptyMap());
+		Flux<ParentClass> input = Flux.just(new Foo(), new Bar());

-		StepVerifier.create(output)
-				.consumeNextWith(resultConsumer("[{\"type\":\"foo\"},{\"type\":\"bar\"}]")
+		testEncode(input, ParentClass.class, step -> step
+				.consumeNextWith(expectString("[{\"type\":\"foo\"},{\"type\":\"bar\"}]")
 						.andThen(DataBufferUtils::release))
-				.verifyComplete();
+				.verifyComplete());
 	}


@@ -156,22 +147,21 @@ public class Jackson2JsonEncoderTests extends AbstractEncoderTestCase<Object, Ja
 		MediaType fooMediaType = new MediaType("application", "foo");
 		MediaType barMediaType = new MediaType("application", "bar");
 		this.encoder.setStreamingMediaTypes(Arrays.asList(fooMediaType, barMediaType));
-		Flux<Pojo> source = Flux.just(
+		Flux<Pojo> input = Flux.just(
 				new Pojo("foo", "bar"),
 				new Pojo("foofoo", "barbar"),
 				new Pojo("foofoofoo", "barbarbar")
 		);
-		ResolvableType type = ResolvableType.forClass(Pojo.class);
-		Flux<DataBuffer> output = this.encoder.encode(source, this.bufferFactory, type, barMediaType, emptyMap());

-		StepVerifier.create(output)
-				.consumeNextWith(resultConsumer("{\"foo\":\"foo\",\"bar\":\"bar\"}\n")
+		testEncode(input, ResolvableType.forClass(Pojo.class), step -> step
+				.consumeNextWith(expectString("{\"foo\":\"foo\",\"bar\":\"bar\"}\n")
 						.andThen(DataBufferUtils::release))
-				.consumeNextWith(resultConsumer("{\"foo\":\"foofoo\",\"bar\":\"barbar\"}\n")
+				.consumeNextWith(expectString("{\"foo\":\"foofoo\",\"bar\":\"barbar\"}\n")
 						.andThen(DataBufferUtils::release))
-				.consumeNextWith(resultConsumer("{\"foo\":\"foofoofoo\",\"bar\":\"barbarbar\"}\n")
+				.consumeNextWith(expectString("{\"foo\":\"foofoofoo\",\"bar\":\"barbarbar\"}\n")
 						.andThen(DataBufferUtils::release))
-				.verifyComplete();
+				.verifyComplete(),
+				barMediaType, null);
 	}

 	@Test
@@ -180,15 +170,16 @@ public class Jackson2JsonEncoderTests extends AbstractEncoderTestCase<Object, Ja
 		bean.setWithView1("with");
 		bean.setWithView2("with");
 		bean.setWithoutView("without");
+		Mono<JacksonViewBean> input = Mono.just(bean);

 		ResolvableType type = ResolvableType.forClass(JacksonViewBean.class);
 		Map<String, Object> hints = singletonMap(JSON_VIEW_HINT, MyJacksonView1.class);
-		Flux<DataBuffer> output = this.encoder.encode(Mono.just(bean), this.bufferFactory, type, null, hints);

-		StepVerifier.create(output)
-				.consumeNextWith(resultConsumer("{\"withView1\":\"with\"}")
-						.andThen(DataBufferUtils::release))
-				.verifyComplete();
+		testEncode(input, type, step -> step
+						.consumeNextWith(expectString("{\"withView1\":\"with\"}")
+								.andThen(DataBufferUtils::release))
+						.verifyComplete(),
+				null, hints);
 	}

 	@Test
@@ -197,17 +188,19 @@ public class Jackson2JsonEncoderTests extends AbstractEncoderTestCase<Object, Ja
 		bean.setWithView1("with");
 		bean.setWithView2("with");
 		bean.setWithoutView("without");
+		Mono<JacksonViewBean> input = Mono.just(bean);

 		ResolvableType type = ResolvableType.forClass(JacksonViewBean.class);
 		Map<String, Object> hints = singletonMap(JSON_VIEW_HINT, MyJacksonView3.class);
-		Flux<DataBuffer> output = this.encoder.encode(Mono.just(bean), this.bufferFactory, type, null, hints);

-		StepVerifier.create(output)
-				.consumeNextWith(resultConsumer("{\"withoutView\":\"without\"}")
-						.andThen(DataBufferUtils::release))
-				.verifyComplete();
+		testEncode(input, type, step -> step
+						.consumeNextWith(expectString("{\"withoutView\":\"without\"}")
+								.andThen(DataBufferUtils::release))
+						.verifyComplete(),
+				null, hints);
 	}

+
 	@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type")
 	private static class ParentClass {
 	}
--- a/spring-web/src/test/java/org/springframework/http/codec/json/Jackson2SmileEncoderTests.java
+++ b/spring-web/src/test/java/org/springframework/http/codec/json/Jackson2SmileEncoderTests.java
@@ -18,20 +18,18 @@ package org.springframework.http.codec.json;

 import java.io.IOException;
 import java.io.UncheckedIOException;
+import java.util.Arrays;
 import java.util.List;
 import java.util.function.Consumer;
-import java.util.stream.Stream;

 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.databind.type.CollectionType;
 import org.junit.Test;
 import reactor.core.publisher.Flux;
-import reactor.test.StepVerifier;
+import reactor.core.publisher.Mono;

 import org.springframework.core.ResolvableType;
 import org.springframework.core.codec.AbstractEncoderTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
-import org.springframework.core.io.buffer.DataBufferUtils;
 import org.springframework.core.io.buffer.support.DataBufferTestUtils;
 import org.springframework.http.codec.Pojo;
 import org.springframework.http.codec.ServerSentEvent;
@@ -39,6 +37,7 @@ import org.springframework.http.converter.json.Jackson2ObjectMapperBuilder;
 import org.springframework.util.MimeType;

 import static org.junit.Assert.*;
+import static org.springframework.core.io.buffer.DataBufferUtils.release;
 import static org.springframework.http.MediaType.APPLICATION_XML;

 /**
@@ -46,7 +45,7 @@ import static org.springframework.http.MediaType.APPLICATION_XML;
 *
 * @author Sebastien Deleuze
 */
-public class Jackson2SmileEncoderTests extends AbstractEncoderTestCase<Object, Jackson2SmileEncoder> {
+public class Jackson2SmileEncoderTests extends AbstractEncoderTestCase<Jackson2SmileEncoder> {

 	private final static MimeType SMILE_MIME_TYPE = new MimeType("application", "x-jackson-smile");
 	private final static MimeType STREAM_SMILE_MIME_TYPE = new MimeType("application", "stream+x-jackson-smile");
@@ -55,39 +54,18 @@ public class Jackson2SmileEncoderTests extends AbstractEncoderTestCase<Object, J

 	private final ObjectMapper mapper = Jackson2ObjectMapperBuilder.smile().build();

-	private Pojo pojo1 = new Pojo("foo", "bar");
-
-	private Pojo pojo2 = new Pojo("foofoo", "barbar");
-
-	private Pojo pojo3 = new Pojo("foofoofoo", "barbarbar");
-
 	public Jackson2SmileEncoderTests() {
-		super(new Jackson2SmileEncoder(), ResolvableType.forClass(Pojo.class),
-				STREAM_SMILE_MIME_TYPE, null);
+		super(new Jackson2SmileEncoder());

 	}

-	@Override
-	protected Flux<Object> input() {
-		return Flux.just(this.pojo1, this.pojo2, this.pojo3);
-	}
-
-	@Override
-	protected Stream<Consumer<DataBuffer>> outputConsumers() {
-		return Stream.<Consumer<DataBuffer>>builder()
-				.add(pojoConsumer(this.pojo1))
-				.add(pojoConsumer(this.pojo2))
-				.add(pojoConsumer(this.pojo3))
-				.build();
-	}
-
 	public Consumer<DataBuffer> pojoConsumer(Pojo expected) {
 		return dataBuffer -> {
 			try {
 				Pojo actual = this.mapper.reader().forType(Pojo.class)
 						.readValue(DataBufferTestUtils.dumpBytes(dataBuffer));
 				assertEquals(expected, actual);
-				DataBufferUtils.release(dataBuffer);
+				release(dataBuffer);
 			}
 			catch (IOException ex) {
 				throw new UncheckedIOException(ex);
@@ -96,6 +74,7 @@ public class Jackson2SmileEncoderTests extends AbstractEncoderTestCase<Object, J
 	}


+	@Override
 	@Test
 	public void canEncode() {
 		ResolvableType pojoType = ResolvableType.forClass(Pojo.class);
@@ -116,32 +95,62 @@ public class Jackson2SmileEncoderTests extends AbstractEncoderTestCase<Object, J
 		assertFalse(this.encoder.canEncode(sseType, SMILE_MIME_TYPE));
 	}

+	@Override
 	@Test
-	public void encodeNonStream() {
-		Flux<DataBuffer> output = this.encoder.encode(input(), this.bufferFactory, elementType,
-				null, null);
+	public void encode() {
+		List<Pojo> list = Arrays.asList(
+				new Pojo("foo", "bar"),
+				new Pojo("foofoo", "barbar"),
+				new Pojo("foofoofoo", "barbarbar"));
+
+		Flux<Pojo> input = Flux.fromIterable(list);
+
+		testEncode(input, Pojo.class, step -> step
+				.consumeNextWith(expect(list, List.class)));
+	}
+
+	@Test
+	public void encodeError() throws Exception {
+		Mono<Pojo> input = Mono.error(new InputException());
+
+		testEncode(input, Pojo.class, step -> step
+				.expectError(InputException.class)
+				.verify());
+
+	}
+
+	@Test
+	public void encodeAsStream() throws Exception {
+		Pojo pojo1 = new Pojo("foo", "bar");
+		Pojo pojo2 = new Pojo("foofoo", "barbar");
+		Pojo pojo3 = new Pojo("foofoofoo", "barbarbar");
+		Flux<Pojo> input = Flux.just(pojo1, pojo2, pojo3);
+		ResolvableType type = ResolvableType.forClass(Pojo.class);
+
+		testEncodeAll(input, type, step -> step
+				.consumeNextWith(expect(pojo1, Pojo.class))
+				.consumeNextWith(expect(pojo2, Pojo.class))
+				.consumeNextWith(expect(pojo3, Pojo.class))
+				.verifyComplete(),
+		STREAM_SMILE_MIME_TYPE, null);
+	}
+
+
+	private <T> Consumer<DataBuffer> expect(T expected, Class<T> expectedType) {
+		return dataBuffer -> {
+			try {
+				Object actual = this.mapper.reader().forType(expectedType)
+						.readValue(dataBuffer.asInputStream());
+				assertEquals(expected, actual);
+			}
+			catch (IOException e) {
+				throw new UncheckedIOException(e);
+			}
+			finally {
+				release(dataBuffer);
+			}
+		};

-		ObjectMapper mapper = Jackson2ObjectMapperBuilder.smile().build();
-		StepVerifier.create(output)
-				.consumeNextWith(dataBuffer -> {
-					try {
-						CollectionType type = mapper.getTypeFactory()
-								.constructCollectionType(List.class, Pojo.class);
-						List<Pojo> value = mapper.reader().forType(type)
-								.readValue(dataBuffer.asInputStream());
-						assertEquals(3, value.size());
-						assertEquals(pojo1, value.get(0));
-						assertEquals(pojo2, value.get(1));
-						assertEquals(pojo3, value.get(2));
-					}
-					catch (IOException ex) {
-						throw new UncheckedIOException(ex);
-					}
-					finally {
-						DataBufferUtils.release(dataBuffer);
-					}
-				})
-				.verifyComplete();
 	}


--- a/spring-web/src/test/java/org/springframework/http/codec/protobuf/ProtobufEncoderTests.java
+++ b/spring-web/src/test/java/org/springframework/http/codec/protobuf/ProtobufEncoderTests.java
@@ -19,14 +19,15 @@ package org.springframework.http.codec.protobuf;
 import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.util.function.Consumer;
-import java.util.stream.Stream;

 import com.google.protobuf.Message;
 import org.junit.Test;
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;

 import org.springframework.core.codec.AbstractEncoderTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferUtils;
 import org.springframework.http.MediaType;
 import org.springframework.protobuf.Msg;
 import org.springframework.protobuf.SecondMsg;
@@ -40,7 +41,7 @@ import static org.springframework.core.ResolvableType.forClass;
 *
 * @author Sebastien Deleuze
 */
-public class ProtobufEncoderTests extends AbstractEncoderTestCase<Message, ProtobufEncoder> {
+public class ProtobufEncoderTests extends AbstractEncoderTestCase<ProtobufEncoder> {

 	private final static MimeType PROTOBUF_MIME_TYPE = new MimeType("application", "x-protobuf");

@@ -50,35 +51,12 @@ public class ProtobufEncoderTests extends AbstractEncoderTestCase<Message, Proto
 	private Msg msg2 =
 			Msg.newBuilder().setFoo("Bar").setBlah(SecondMsg.newBuilder().setBlah(456).build()).build();

+
 	public ProtobufEncoderTests() {
-		super(new ProtobufEncoder(), Msg.class);
+		super(new ProtobufEncoder());
 	}

 	@Override
-	protected Flux<Message> input() {
-		return Flux.just(this.msg1, this.msg2);
-	}
-
-	@Override
-	protected Stream<Consumer<DataBuffer>> outputConsumers() {
-		return Stream.<Consumer<DataBuffer>>builder()
-				.add(resultConsumer(this.msg1))
-				.add(resultConsumer(this.msg2))
-				.build();
-
-	}
-
-	protected final Consumer<DataBuffer> resultConsumer(Msg msg) {
-		return dataBuffer -> {
-			try {
-				assertEquals(msg, Msg.parseDelimitedFrom(dataBuffer.asInputStream()));
-			}
-			catch (IOException ex) {
-				throw new UncheckedIOException(ex);
-			}
-		};
-	}
-
 	@Test
 	public void canEncode() {
 		assertTrue(this.encoder.canEncode(forClass(Msg.class), null));
@@ -88,4 +66,49 @@ public class ProtobufEncoderTests extends AbstractEncoderTestCase<Message, Proto
 		assertFalse(this.encoder.canEncode(forClass(Object.class), PROTOBUF_MIME_TYPE));
 	}

+	@Override
+	@Test
+	public void encode() {
+		Mono<Message> input = Mono.just(this.msg1);
+
+		testEncodeAll(input, Msg.class, step -> step
+				.consumeNextWith(dataBuffer -> {
+					try {
+						assertEquals(this.msg1, Msg.parseFrom(dataBuffer.asInputStream()));
+
+					}
+					catch (IOException ex) {
+						throw new UncheckedIOException(ex);
+					}
+					finally {
+						DataBufferUtils.release(dataBuffer);
+					}
+				})
+				.verifyComplete());
+	}
+
+	@Test
+	public void encodeStream() {
+		Flux<Message> input = Flux.just(this.msg1, this.msg2);
+
+		testEncodeAll(input, Msg.class, step -> step
+				.consumeNextWith(expect(this.msg1))
+				.consumeNextWith(expect(this.msg2))
+				.verifyComplete());
+	}
+
+	protected final Consumer<DataBuffer> expect(Msg msg) {
+		return dataBuffer -> {
+			try {
+				assertEquals(msg, Msg.parseDelimitedFrom(dataBuffer.asInputStream()));
+
+			}
+			catch (IOException ex) {
+				throw new UncheckedIOException(ex);
+			}
+			finally {
+				DataBufferUtils.release(dataBuffer);
+			}
+		};
+	}
 }
--- a/spring-web/src/test/java/org/springframework/http/codec/xml/Jaxb2XmlEncoderTests.java
+++ b/spring-web/src/test/java/org/springframework/http/codec/xml/Jaxb2XmlEncoderTests.java
@@ -16,38 +16,39 @@

 package org.springframework.http.codec.xml;

-import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
 import java.util.List;
 import java.util.function.Consumer;
-import java.util.stream.Stream;
 import javax.xml.bind.annotation.XmlElement;
 import javax.xml.bind.annotation.XmlElements;
 import javax.xml.bind.annotation.XmlRootElement;

 import org.junit.Test;
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;

 import org.springframework.core.ResolvableType;
 import org.springframework.core.codec.AbstractEncoderTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
-import org.springframework.core.io.buffer.support.DataBufferTestUtils;
 import org.springframework.http.MediaType;
 import org.springframework.http.codec.Pojo;

+import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.junit.Assert.*;
+import static org.springframework.core.io.buffer.DataBufferUtils.release;
 import static org.xmlunit.matchers.CompareMatcher.isSimilarTo;

 /**
 * @author Sebastien Deleuze
 * @author Arjen Poutsma
 */
-public class Jaxb2XmlEncoderTests extends AbstractEncoderTestCase<Object, Jaxb2XmlEncoder> {
+public class Jaxb2XmlEncoderTests extends AbstractEncoderTestCase<Jaxb2XmlEncoder> {

 	public Jaxb2XmlEncoderTests() {
-		super(new Jaxb2XmlEncoder(), Pojo.class);
+		super(new Jaxb2XmlEncoder());
 	}

+	@Override
 	@Test
 	public void canEncode() {
 		assertTrue(this.encoder.canEncode(ResolvableType.forClass(Pojo.class),
@@ -69,23 +70,46 @@ public class Jaxb2XmlEncoderTests extends AbstractEncoderTestCase<Object, Jaxb2X
 	}

 	@Override
-	protected Flux<Object> input() {
-		return Flux.just(new Container());
+	@Test
+	public void encode() {
+		Mono<Pojo> input = Mono.just(new Pojo("foofoo", "barbar"));
+
+		testEncode(input, Pojo.class, step -> step
+				.consumeNextWith(
+						expectXml("<?xml version='1.0' encoding='UTF-8' standalone='yes'?>" +
+								"<pojo><bar>barbar</bar><foo>foofoo</foo></pojo>"))
+				.verifyComplete());
 	}

-	@Override
-	protected Stream<Consumer<DataBuffer>> outputConsumers() {
-		return Stream.<Consumer<DataBuffer>>builder()
-				.add(dataBuffer -> {
-					String s = DataBufferTestUtils
-							.dumpString(dataBuffer, StandardCharsets.UTF_8);
-					assertThat(s,
-							isSimilarTo("<?xml version='1.0' encoding='UTF-8' standalone='yes'?>" +
-									"<container><foo><name>name1</name></foo><bar><title>title1</title></bar></container>"));
-				})
-				.build();
+	@Test
+	public void encodeError() {
+		Flux<Pojo> input = Flux.error(RuntimeException::new);
+
+		testEncode(input, Pojo.class, step -> step
+				.expectError(RuntimeException.class)
+				.verify());
 	}

+	@Test
+	public void encodeElementsWithCommonType() {
+		Mono<Container> input = Mono.just(new Container());
+
+		testEncode(input, Pojo.class, step -> step
+				.consumeNextWith(
+						expectXml("<?xml version='1.0' encoding='UTF-8' standalone='yes'?>" +
+								"<container><foo><name>name1</name></foo><bar><title>title1</title></bar></container>"))
+				.verifyComplete());
+	}
+
+	protected Consumer<DataBuffer> expectXml(String expected) {
+		return dataBuffer -> {
+			byte[] resultBytes = new byte[dataBuffer.readableByteCount()];
+			dataBuffer.read(resultBytes);
+			release(dataBuffer);
+			String actual = new String(resultBytes, UTF_8);
+			assertThat(actual, isSimilarTo(expected));
+		};
+	}

 	public static class Model {}