String encoding for any MIME type

CharSequenceEncoder now supports all MIME types, however since encoding
Flux<String> can overlap with other encoders (e.g. SSE) there are now
two ways to create a CharSequenceEncoder -- with support for text/plain
only or with support for any MIME type.

In WebFlux configuration we insert one CharSequenceEncoder for
text/plain (as we have so far) and a second instance with support for
any MIME type at the very end.

Issue: SPR-15374

spring-core/src/main/java/org/springframework/core/codec/CharSequenceEncoder.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -29,12 +29,14 @@
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferFactory;
 import org.springframework.util.MimeType;
+import org.springframework.util.MimeTypeUtils;
 
 /**
  * Encode from a {@code CharSequence} stream to a bytes stream.
  *
  * @author Sebastien Deleuze
  * @author Arjen Poutsma
+ * @author Rossen Stoyanchev
  * @since 5.0
  * @see StringDecoder
  */
@@ -43,8 +45,8 @@ public class CharSequenceEncoder extends AbstractEncoder<CharSequence> {
 	public static final Charset DEFAULT_CHARSET = StandardCharsets.UTF_8;
 
 
-	public CharSequenceEncoder() {
-		super(new MimeType("text", "plain", DEFAULT_CHARSET));
+	private CharSequenceEncoder(MimeType... mimeTypes) {
+		super(mimeTypes);
 	}
 
 
@@ -73,4 +75,19 @@ public Flux<DataBuffer> encode(Publisher<? extends CharSequence> inputStream,
 		});
 	}
 
+
+	/**
+	 * Create a {@code CharSequenceEncoder} that supports only "text/plain".
+	 */
+	public static CharSequenceEncoder textPlainOnly() {
+		return new CharSequenceEncoder(new MimeType("text", "plain", DEFAULT_CHARSET));
+	}
+
+	/**
+	 * Create a {@code CharSequenceEncoder} that supports all MIME types.
+	 */
+	public static CharSequenceEncoder allMimeTypes() {
+		return new CharSequenceEncoder(new MimeType("text", "plain", DEFAULT_CHARSET), MimeTypeUtils.ALL);
+	}
+
 }

spring-core/src/test/java/org/springframework/core/codec/CharSequenceEncoderTests.java

@@ -43,7 +43,7 @@ public class CharSequenceEncoderTests extends AbstractDataBufferAllocatingTestCa
 
 	@Before
 	public void createEncoder() {
-		this.encoder = new CharSequenceEncoder();
+		this.encoder = CharSequenceEncoder.textPlainOnly();
 	}
 
 	@Test

spring-web/src/main/java/org/springframework/http/codec/ServerSentEventHttpMessageReader.java

@@ -37,7 +37,6 @@
 import org.springframework.core.io.buffer.DefaultDataBufferFactory;
 import org.springframework.http.MediaType;
 import org.springframework.http.ReactiveHttpInputMessage;
-import org.springframework.util.Assert;
 
 import static java.util.stream.Collectors.joining;
 
@@ -63,10 +62,18 @@ public class ServerSentEventHttpMessageReader implements HttpMessageReader<Objec
 
 
 	/**
-	 * Constructor with JSON {@code Encoder} for encoding objects.
+	 * Constructor without a {@code Decoder}. In this mode only {@code String}
+	 * is supported as the data of an event.
+	 */
+	public ServerSentEventHttpMessageReader() {
+		this(null);
+	}
+
+	/**
+	 * Constructor with JSON {@code Decoder} for decoding to Objects. Support
+	 * for decoding to {@code String} event data is built-in.
 	 */
 	public ServerSentEventHttpMessageReader(Decoder<?> decoder) {
-		Assert.notNull(decoder, "Decoder must not be null");
 		this.decoder = decoder;
 	}
 
@@ -85,7 +92,7 @@ public List<MediaType> getReadableMediaTypes() {
 
 	@Override
 	public boolean canRead(ResolvableType elementType, MediaType mediaType) {
-		return MediaType.TEXT_EVENT_STREAM.isCompatibleWith(mediaType) ||
+		return MediaType.TEXT_EVENT_STREAM.includes(mediaType) ||
 				ServerSentEvent.class.isAssignableFrom(elementType.getRawClass());
 	}
 
@@ -183,8 +190,6 @@ private Object decodeData(String data, ResolvableType dataType, Map<String, Obje
 	public Mono<Object> readMono(ResolvableType elementType, ReactiveHttpInputMessage message,
 			Map<String, Object> hints) {
 
-		// For single String give StringDecoder a chance which comes after SSE in the order
-
 		if (String.class.equals(elementType.getRawClass())) {
 			Flux<DataBuffer> body = message.getBody();
 			return stringDecoder.decodeToMono(body, elementType, null, null).cast(Object.class);

spring-web/src/main/java/org/springframework/http/codec/ServerSentEventHttpMessageWriter.java

@@ -27,14 +27,14 @@
 import reactor.core.publisher.Mono;
 
 import org.springframework.core.ResolvableType;
+import org.springframework.core.codec.CodecException;
 import org.springframework.core.codec.Encoder;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferFactory;
 import org.springframework.http.MediaType;
 import org.springframework.http.ReactiveHttpOutputMessage;
 import org.springframework.http.server.reactive.ServerHttpRequest;
 import org.springframework.http.server.reactive.ServerHttpResponse;
-import org.springframework.util.Assert;
 
 /**
  * {@code ServerHttpMessageWriter} for {@code "text/event-stream"} responses.
@@ -53,18 +53,25 @@ public class ServerSentEventHttpMessageWriter implements ServerHttpMessageWriter
 	private final Encoder<?> encoder;
 
 
+	/**
+	 * Constructor without an {@code Encoder}. In this mode only {@code String}
+	 * is supported for event data to be encoded.
+	 */
+	public ServerSentEventHttpMessageWriter() {
+		this(null);
+	}
+
 	/**
 	 * Constructor with JSON {@code Encoder} for encoding objects. Support for
 	 * {@code String} event data is built-in.
 	 */
 	public ServerSentEventHttpMessageWriter(Encoder<?> encoder) {
-		Assert.notNull(encoder, "'encoder' must not be null");
 		this.encoder = encoder;
 	}
 
 
 	/**
-	 * Return the configured {@code Encoder}.
+	 * Return the configured {@code Encoder}, possibly {@code null}.
 	 */
 	public Encoder<?> getEncoder() {
 		return this.encoder;
@@ -78,7 +85,7 @@ public List<MediaType> getWritableMediaTypes() {
 
 	@Override
 	public boolean canWrite(ResolvableType elementType, MediaType mediaType) {
-		return mediaType == null || MediaType.TEXT_EVENT_STREAM.isCompatibleWith(mediaType) ||
+		return mediaType == null || MediaType.TEXT_EVENT_STREAM.includes(mediaType) ||
 				ServerSentEvent.class.isAssignableFrom(elementType.getRawClass());
 	}
 
@@ -135,6 +142,10 @@ private <T> Flux<DataBuffer> encodeData(ServerSentEvent<?> event, ResolvableType
 			return Flux.from(encodeText(text.replaceAll("\\n", "\ndata:") + "\n", factory));
 		}
 
+		if (this.encoder == null) {
+			return Flux.error(new CodecException("No SSE encoder configured and the data is not String."));
+		}
+
 		return ((Encoder<T>) this.encoder)
 				.encode(Mono.just((T) data), factory, valueType, MediaType.TEXT_EVENT_STREAM, hints)
 				.concatWith(encodeText("\n", factory));

spring-webflux/src/main/java/org/springframework/web/reactive/config/WebFluxConfigurationSupport.java

@@ -37,6 +37,7 @@
 import org.springframework.core.codec.CharSequenceEncoder;
 import org.springframework.core.codec.DataBufferDecoder;
 import org.springframework.core.codec.DataBufferEncoder;
+import org.springframework.core.codec.Encoder;
 import org.springframework.core.codec.ResourceDecoder;
 import org.springframework.core.codec.StringDecoder;
 import org.springframework.core.convert.converter.Converter;
@@ -469,6 +470,7 @@ protected final List<ServerHttpMessageWriter<?>> getMessageWriters() {
 	 */
 	protected void configureMessageWriters(List<ServerHttpMessageWriter<?>> messageWriters) {
 	}
+
 	/**
 	 * Adds default converters that sub-classes can call from
 	 * {@link #configureMessageWriters(List)}.
@@ -477,15 +479,24 @@ protected final void addDefaultHttpMessageWriters(List<ServerHttpMessageWriter<?
 		writers.add(new EncoderHttpMessageWriter<>(new ByteArrayEncoder()));
 		writers.add(new EncoderHttpMessageWriter<>(new ByteBufferEncoder()));
 		writers.add(new EncoderHttpMessageWriter<>(new DataBufferEncoder()));
-		writers.add(new EncoderHttpMessageWriter<>(new CharSequenceEncoder()));
+		writers.add(new EncoderHttpMessageWriter<>(CharSequenceEncoder.textPlainOnly()));
 		writers.add(new ResourceHttpMessageWriter());
 		if (jaxb2Present) {
 			writers.add(new EncoderHttpMessageWriter<>(new Jaxb2XmlEncoder()));
 		}
 		if (jackson2Present) {
-			Jackson2JsonEncoder jacksonEncoder = new Jackson2JsonEncoder();
-			writers.add(new EncoderHttpMessageWriter<>(jacksonEncoder));
-			writers.add(new ServerSentEventHttpMessageWriter(jacksonEncoder));
+			writers.add(new EncoderHttpMessageWriter<>(new Jackson2JsonEncoder()));
+		}
+		writers.add(new ServerSentEventHttpMessageWriter(getSseEncoder()));
+		writers.add(new EncoderHttpMessageWriter<>(CharSequenceEncoder.allMimeTypes()));
+	}
+
+	private Encoder<?> getSseEncoder() {
+		if (jackson2Present) {
+			return new Jackson2JsonEncoder();
+		}
+		else {
+			return null;
 		}
 	}
 

spring-webflux/src/main/java/org/springframework/web/reactive/function/client/DefaultExchangeStrategiesBuilder.java

@@ -75,12 +75,10 @@ public void defaultConfiguration() {
 	}
 
 	private void defaultReaders() {
+		// SSE first (constrained to "text/event-stream")
+		messageReader(new ServerSentEventHttpMessageReader(getSseDecoder()));
 		messageReader(new DecoderHttpMessageReader<>(new ByteArrayDecoder()));
 		messageReader(new DecoderHttpMessageReader<>(new ByteBufferDecoder()));
-		if (jackson2Present) {
-			// SSE ahead of String e.g. "test/event-stream" + Flux<String>
-			messageReader(new ServerSentEventHttpMessageReader(new Jackson2JsonDecoder()));
-		}
 		messageReader(new DecoderHttpMessageReader<>(new StringDecoder(false)));
 		if (jaxb2Present) {
 			messageReader(new DecoderHttpMessageReader<>(new Jaxb2XmlDecoder()));
@@ -90,10 +88,19 @@ private void defaultReaders() {
 		}
 	}
 
+	private Decoder<?> getSseDecoder() {
+		if (jackson2Present) {
+			return new Jackson2JsonDecoder();
+		}
+		else {
+			return null;
+		}
+	}
+
 	private void defaultWriters() {
 		messageWriter(new EncoderHttpMessageWriter<>(new ByteArrayEncoder()));
 		messageWriter(new EncoderHttpMessageWriter<>(new ByteBufferEncoder()));
-		messageWriter(new EncoderHttpMessageWriter<>(new CharSequenceEncoder()));
+		messageWriter(new EncoderHttpMessageWriter<>(CharSequenceEncoder.allMimeTypes()));
 		messageWriter(new ResourceHttpMessageWriter());
 		messageWriter(new FormHttpMessageWriter());
 		if (jaxb2Present) {

spring-webflux/src/main/java/org/springframework/web/reactive/function/server/DefaultHandlerStrategiesBuilder.java

@@ -31,6 +31,7 @@
 import org.springframework.core.codec.ByteBufferDecoder;
 import org.springframework.core.codec.ByteBufferEncoder;
 import org.springframework.core.codec.CharSequenceEncoder;
+import org.springframework.core.codec.Encoder;
 import org.springframework.core.codec.StringDecoder;
 import org.springframework.http.codec.DecoderHttpMessageReader;
 import org.springframework.http.codec.EncoderHttpMessageWriter;
@@ -88,7 +89,7 @@ public void defaultConfiguration() {
 
 		messageWriter(new EncoderHttpMessageWriter<>(new ByteArrayEncoder()));
 		messageWriter(new EncoderHttpMessageWriter<>(new ByteBufferEncoder()));
-		messageWriter(new EncoderHttpMessageWriter<>(new CharSequenceEncoder()));
+		messageWriter(new EncoderHttpMessageWriter<>(CharSequenceEncoder.textPlainOnly()));
 		messageWriter(new ResourceHttpMessageWriter());
 
 		if (jaxb2Present) {
@@ -97,13 +98,24 @@ public void defaultConfiguration() {
 		}
 		if (jackson2Present) {
 			messageReader(new DecoderHttpMessageReader<>(new Jackson2JsonDecoder()));
-			Jackson2JsonEncoder jsonEncoder = new Jackson2JsonEncoder();
-			messageWriter(new EncoderHttpMessageWriter<>(jsonEncoder));
-			messageWriter(new ServerSentEventHttpMessageWriter(jsonEncoder));
+			messageWriter(new EncoderHttpMessageWriter<>(new Jackson2JsonEncoder()));
 		}
+
+		messageWriter(new ServerSentEventHttpMessageWriter(getSseEncoder()));
+		messageWriter(new EncoderHttpMessageWriter<>(CharSequenceEncoder.allMimeTypes()));
+
 		localeResolver(DEFAULT_LOCALE_RESOLVER);
 	}
 
+	private Encoder<?> getSseEncoder() {
+		if (jackson2Present) {
+			return new Jackson2JsonEncoder();
+		}
+		else {
+			return null;
+		}
+	}
+
 	public void applicationContext(ApplicationContext applicationContext) {
 		applicationContext.getBeansOfType(HttpMessageReader.class).values().forEach(this::messageReader);
 		applicationContext.getBeansOfType(HttpMessageWriter.class).values().forEach(this::messageWriter);

spring-webflux/src/test/java/org/springframework/web/reactive/DispatcherHandlerErrorTests.java

@@ -186,8 +186,8 @@ public RequestMappingHandlerAdapter handlerAdapter() {
 
 		@Bean
 		public ResponseBodyResultHandler resultHandler() {
-			return new ResponseBodyResultHandler(
-					Collections.singletonList(new EncoderHttpMessageWriter<>(new CharSequenceEncoder())),
+			return new ResponseBodyResultHandler(Collections.singletonList(
+					new EncoderHttpMessageWriter<>(CharSequenceEncoder.textPlainOnly())),
 					new HeaderContentTypeResolver());
 		}
 

spring-webflux/src/test/java/org/springframework/web/reactive/config/WebFluxConfigurationSupportTests.java

@@ -178,7 +178,7 @@ public void responseEntityResultHandler() throws Exception {
 		assertEquals(0, handler.getOrder());
 
 		List<ServerHttpMessageWriter<?>> writers = handler.getMessageWriters();
-		assertEquals(8, writers.size());
+		assertEquals(9, writers.size());
 
 		assertHasMessageWriter(writers, byte[].class, APPLICATION_OCTET_STREAM);
 		assertHasMessageWriter(writers, ByteBuffer.class, APPLICATION_OCTET_STREAM);
@@ -204,7 +204,7 @@ public void responseBodyResultHandler() throws Exception {
 		assertEquals(100, handler.getOrder());
 
 		List<ServerHttpMessageWriter<?>> writers = handler.getMessageWriters();
-		assertEquals(8, writers.size());
+		assertEquals(9, writers.size());
 
 		assertHasMessageWriter(writers, byte[].class, APPLICATION_OCTET_STREAM);
 		assertHasMessageWriter(writers, ByteBuffer.class, APPLICATION_OCTET_STREAM);
@@ -303,7 +303,7 @@ protected void configureMessageReaders(List<ServerHttpMessageReader<?>> messageR
 
 		@Override
 		protected void configureMessageWriters(List<ServerHttpMessageWriter<?>> messageWriters) {
-			messageWriters.add(new EncoderHttpMessageWriter<>(new CharSequenceEncoder()));
+			messageWriters.add(new EncoderHttpMessageWriter<>(CharSequenceEncoder.textPlainOnly()));
 		}
 
 		@Override

spring-webflux/src/test/java/org/springframework/web/reactive/function/BodyInsertersTests.java

@@ -77,13 +77,14 @@ public class BodyInsertersTests {
 	public void createContext() {
 		final List<HttpMessageWriter<?>> messageWriters = new ArrayList<>();
 		messageWriters.add(new EncoderHttpMessageWriter<>(new ByteBufferEncoder()));
-		messageWriters.add(new EncoderHttpMessageWriter<>(new CharSequenceEncoder()));
+		messageWriters.add(new EncoderHttpMessageWriter<>(CharSequenceEncoder.textPlainOnly()));
 		messageWriters.add(new ResourceHttpMessageWriter());
 		messageWriters.add(new EncoderHttpMessageWriter<>(new Jaxb2XmlEncoder()));
 		Jackson2JsonEncoder jsonEncoder = new Jackson2JsonEncoder();
 		messageWriters.add(new EncoderHttpMessageWriter<>(jsonEncoder));
 		messageWriters.add(new ServerSentEventHttpMessageWriter(jsonEncoder));
 		messageWriters.add(new FormHttpMessageWriter());
+		messageWriters.add(new EncoderHttpMessageWriter<>(CharSequenceEncoder.allMimeTypes()));
 
 		this.context = new BodyInserter.Context() {
 			@Override

spring-webflux/src/test/java/org/springframework/web/reactive/function/client/DefaultClientRequestBuilderTests.java

@@ -104,7 +104,7 @@ public void bodyInserter() throws Exception {
 				.body(inserter).build();
 
 		List<HttpMessageWriter<?>> messageWriters = new ArrayList<>();
-		messageWriters.add(new EncoderHttpMessageWriter<>(new CharSequenceEncoder()));
+		messageWriters.add(new EncoderHttpMessageWriter<>(CharSequenceEncoder.allMimeTypes()));
 
 		ExchangeStrategies strategies = mock(ExchangeStrategies.class);
 		when(strategies.messageWriters()).thenReturn(messageWriters::stream);

spring-webflux/src/test/java/org/springframework/web/reactive/function/server/DefaultEntityResponseBuilderTests.java

@@ -191,7 +191,9 @@ public void bodyInserter() throws Exception {
 
 		MockServerWebExchange exchange = MockServerHttpRequest.get("http://localhost").toExchange();
 
-		HandlerStrategies strategies = HandlerStrategies.empty().messageWriter(new EncoderHttpMessageWriter<>(new CharSequenceEncoder())).build();
+		HandlerStrategies strategies = HandlerStrategies.empty()
+				.messageWriter(new EncoderHttpMessageWriter<>(CharSequenceEncoder.allMimeTypes()))
+				.build();
 
 		StepVerifier.create(result)
 				.consumeNextWith(response -> {

spring-webflux/src/test/java/org/springframework/web/reactive/result/method/annotation/MessageWriterResultHandlerTests.java

@@ -76,7 +76,7 @@ private AbstractMessageWriterResultHandler initResultHandler(ServerHttpMessageWr
 		if (ObjectUtils.isEmpty(writers)) {
 			writerList = new ArrayList<>();
 			writerList.add(new EncoderHttpMessageWriter<>(new ByteBufferEncoder()));
-			writerList.add(new EncoderHttpMessageWriter<>(new CharSequenceEncoder()));
+			writerList.add(new EncoderHttpMessageWriter<>(CharSequenceEncoder.allMimeTypes()));
 			writerList.add(new ResourceHttpMessageWriter());
 			writerList.add(new EncoderHttpMessageWriter<>(new Jaxb2XmlEncoder()));
 			writerList.add(new EncoderHttpMessageWriter<>(new Jackson2JsonEncoder()));

spring-webflux/src/test/java/org/springframework/web/reactive/result/method/annotation/ResponseBodyResultHandlerTests.java

@@ -66,7 +66,7 @@ public class ResponseBodyResultHandlerTests {
 	public void setup() throws Exception {
 		List<ServerHttpMessageWriter<?>> writerList = new ArrayList<>(5);
 		writerList.add(new EncoderHttpMessageWriter<>(new ByteBufferEncoder()));
-		writerList.add(new EncoderHttpMessageWriter<>(new CharSequenceEncoder()));
+		writerList.add(new EncoderHttpMessageWriter<>(CharSequenceEncoder.allMimeTypes()));
 		writerList.add(new ResourceHttpMessageWriter());
 		writerList.add(new EncoderHttpMessageWriter<>(new Jaxb2XmlEncoder()));
 		writerList.add(new EncoderHttpMessageWriter<>(new Jackson2JsonEncoder()));