Revise "streaming" MediaType support

Push the knowledge of what media types represent "streaming" down to
the Encoder level where knowledge is required (e.g. to encode a
JSON array vs a stream of JSON elements).

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

@@ -129,13 +129,13 @@ public class DecoderHttpMessageReader<T> implements ServerHttpMessageReader<T> {
 	/**
 	 * Get additional hints for decoding for example based on the server request
 	 * or annotations from controller method parameters. By default, delegate to
-	 * the decoder if it is an instance of {@link ServerHttpDecoder}.
+	 * the decoder if it is an instance of {@link HttpDecoder}.
 	 */
 	protected Map<String, Object> getReadHints(ResolvableType streamType,
 			ResolvableType elementType, ServerHttpRequest request, ServerHttpResponse response) {
 
-		if (this.decoder instanceof ServerHttpDecoder) {
-			ServerHttpDecoder<?> httpDecoder = (ServerHttpDecoder<?>) this.decoder;
+		if (this.decoder instanceof HttpDecoder) {
+			HttpDecoder<?> httpDecoder = (HttpDecoder<?>) this.decoder;
 			return httpDecoder.getDecodeHints(streamType, elementType, request, response);
 		}
 		return Collections.emptyMap();

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

@@ -16,7 +16,6 @@
 
 package org.springframework.http.codec;
 
-import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
@@ -49,22 +48,12 @@ import org.springframework.util.Assert;
  */
 public class EncoderHttpMessageWriter<T> implements ServerHttpMessageWriter<T> {
 
-	/**
-	 * Default list of media types that signify a "streaming" scenario such that
-	 * there may be a time lag between items written and hence requires flushing.
-	 */
-	public static final List<MediaType> DEFAULT_STREAMING_MEDIA_TYPES =
-			Collections.singletonList(MediaType.APPLICATION_STREAM_JSON);
-
-
 	private final Encoder<T> encoder;
 
 	private final List<MediaType> mediaTypes;
 
 	private final MediaType defaultMediaType;
 
-	private final List<MediaType> streamingMediaTypes = new ArrayList<>(1);
-
 
 	/**
 	 * Create an instance wrapping the given {@link Encoder}.
@@ -74,7 +63,6 @@ public class EncoderHttpMessageWriter<T> implements ServerHttpMessageWriter<T> {
 		this.encoder = encoder;
 		this.mediaTypes = MediaType.asMediaTypes(encoder.getEncodableMimeTypes());
 		this.defaultMediaType = initDefaultMediaType(this.mediaTypes);
-		this.streamingMediaTypes.addAll(DEFAULT_STREAMING_MEDIA_TYPES);
 	}
 
 	private static MediaType initDefaultMediaType(List<MediaType> mediaTypes) {
@@ -94,23 +82,6 @@ public class EncoderHttpMessageWriter<T> implements ServerHttpMessageWriter<T> {
 		return this.mediaTypes;
 	}
 
-	/**
-	 * Configure "streaming" media types for which flushing should be performed
-	 * automatically vs at the end of the input stream.
-	 * <p>By default this is set to {@link #DEFAULT_STREAMING_MEDIA_TYPES}.
-	 * @param mediaTypes one or more media types to add to the list
-	 */
-	public void setStreamingMediaTypes(List<MediaType> mediaTypes) {
-		this.streamingMediaTypes.addAll(mediaTypes);
-	}
-
-	/**
-	 * Return the configured list of "streaming" media types.
-	 */
-	public List<MediaType> getStreamingMediaTypes() {
-		return Collections.unmodifiableList(this.streamingMediaTypes);
-	}
-
 
 	@Override
 	public boolean canWrite(ResolvableType elementType, MediaType mediaType) {
@@ -159,7 +130,9 @@ public class EncoderHttpMessageWriter<T> implements ServerHttpMessageWriter<T> {
 	}
 
 	private boolean isStreamingMediaType(MediaType contentType) {
-		return this.streamingMediaTypes.stream().anyMatch(contentType::isCompatibleWith);
+		return this.encoder instanceof HttpEncoder &&
+				((HttpEncoder<?>) this.encoder).getStreamingMediaTypes().stream()
+						.anyMatch(contentType::isCompatibleWith);
 	}
 
 
@@ -180,13 +153,13 @@ public class EncoderHttpMessageWriter<T> implements ServerHttpMessageWriter<T> {
 	/**
 	 * Get additional hints for encoding for example based on the server request
 	 * or annotations from controller method parameters. By default, delegate to
-	 * the encoder if it is an instance of {@link ServerHttpEncoder}.
+	 * the encoder if it is an instance of {@link HttpEncoder}.
 	 */
 	protected Map<String, Object> getWriteHints(ResolvableType streamType, ResolvableType elementType,
 			MediaType mediaType, ServerHttpRequest request, ServerHttpResponse response) {
 
-		if (this.encoder instanceof ServerHttpEncoder) {
-			ServerHttpEncoder<?> httpEncoder = (ServerHttpEncoder<?>) this.encoder;
+		if (this.encoder instanceof HttpEncoder) {
+			HttpEncoder<?> httpEncoder = (HttpEncoder<?>) this.encoder;
 			return httpEncoder.getEncodeHints(streamType, elementType, mediaType, request, response);
 		}
 		return Collections.emptyMap();

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

@@ -24,12 +24,13 @@ import org.springframework.http.server.reactive.ServerHttpRequest;
 import org.springframework.http.server.reactive.ServerHttpResponse;
 
 /**
- * {@code Decoder} extension for server-side decoding of the HTTP request body.
+ * Extension of {@code Decoder} exposing extra methods relevant in the context
+ * of HTTP applications.
  *
  * @author Rossen Stoyanchev
  * @since 5.0
  */
-public interface ServerHttpDecoder<T> extends Decoder<T> {
+public interface HttpDecoder<T> extends Decoder<T> {
 
 	/**
 	 * Get decoding hints based on the server request or annotations on the

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

@@ -16,6 +16,8 @@
 
 package org.springframework.http.codec;
 
+import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 
 import org.springframework.core.ResolvableType;
@@ -26,12 +28,19 @@ import org.springframework.http.server.reactive.ServerHttpResponse;
 
 
 /**
- * {@code Encoder} extension for server-side encoding of the HTTP response body.
+ * Extension of {@code Encoder} exposing extra methods relevant in the context
+ * of HTTP applications.
  *
  * @author Rossen Stoyanchev
  * @since 5.0
  */
-public interface ServerHttpEncoder<T> extends Encoder<T> {
+public interface HttpEncoder<T> extends Encoder<T> {
+
+	/**
+	 * Return "streaming" media types for which flushing should be performed
+	 * automatically vs at the end of the input stream.
+	 */
+	List<MediaType> getStreamingMediaTypes();
 
 	/**
 	 * Get decoding hints based on the server request or annotations on the
@@ -46,7 +55,10 @@ public interface ServerHttpEncoder<T> extends Encoder<T> {
 	 * @param response the current response
 	 * @return a Map with hints, possibly empty
 	 */
-	Map<String, Object> getEncodeHints(ResolvableType actualType, ResolvableType elementType,
-			MediaType mediaType, ServerHttpRequest request, ServerHttpResponse response);
+	default Map<String, Object> getEncodeHints(ResolvableType actualType, ResolvableType elementType,
+			MediaType mediaType, ServerHttpRequest request, ServerHttpResponse response) {
+
+		return Collections.emptyMap();
+	}
 
 }

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

@@ -71,6 +71,13 @@ public class ServerSentEventHttpMessageReader implements HttpMessageReader<Objec
 	}
 
 
+	/**
+	 * Return the configured {@code Decoder}.
+	 */
+	public Decoder<?> getDecoder() {
+		return this.decoder;
+	}
+
 	@Override
 	public List<MediaType> getReadableMediaTypes() {
 		return Collections.singletonList(MediaType.TEXT_EVENT_STREAM);

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

@@ -63,6 +63,13 @@ public class ServerSentEventHttpMessageWriter implements ServerHttpMessageWriter
 	}
 
 
+	/**
+	 * Return the configured {@code Encoder}.
+	 */
+	public Encoder<?> getEncoder() {
+		return this.encoder;
+	}
+
 	@Override
 	public List<MediaType> getWritableMediaTypes() {
 		return WRITABLE_MEDIA_TYPES;
@@ -154,8 +161,8 @@ public class ServerSentEventHttpMessageWriter implements ServerHttpMessageWriter
 	private Map<String, Object> getEncodeHints(ResolvableType actualType, ResolvableType elementType,
 			MediaType mediaType, ServerHttpRequest request, ServerHttpResponse response) {
 
-		if (this.encoder instanceof ServerHttpEncoder) {
-			ServerHttpEncoder<?> httpEncoder = (ServerHttpEncoder<?>) this.encoder;
+		if (this.encoder instanceof HttpEncoder) {
+			HttpEncoder<?> httpEncoder = (HttpEncoder<?>) this.encoder;
 			return httpEncoder.getEncodeHints(actualType, elementType, mediaType, request, response);
 		}
 		return Collections.emptyMap();

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

@@ -33,7 +33,7 @@ import org.springframework.core.ResolvableType;
 import org.springframework.core.codec.CodecException;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferUtils;
-import org.springframework.http.codec.ServerHttpDecoder;
+import org.springframework.http.codec.HttpDecoder;
 import org.springframework.http.converter.json.Jackson2ObjectMapperBuilder;
 import org.springframework.http.server.reactive.ServerHttpRequest;
 import org.springframework.http.server.reactive.ServerHttpResponse;
@@ -48,7 +48,7 @@ import org.springframework.util.MimeType;
  * @since 5.0
  * @see Jackson2JsonEncoder
  */
-public class Jackson2JsonDecoder extends Jackson2CodecSupport implements ServerHttpDecoder<Object> {
+public class Jackson2JsonDecoder extends Jackson2CodecSupport implements HttpDecoder<Object> {
 
 	private final JsonObjectDecoder fluxDecoder = new JsonObjectDecoder(true);
 
@@ -118,7 +118,7 @@ public class Jackson2JsonDecoder extends Jackson2CodecSupport implements ServerH
 	}
 
 
-	// ServerHttpDecoder...
+	// HttpDecoder...
 
 	@Override
 	public Map<String, Object> getDecodeHints(ResolvableType actualType, ResolvableType elementType,

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

@@ -19,6 +19,8 @@ package org.springframework.http.codec.json;
 import java.io.IOException;
 import java.io.OutputStream;
 import java.lang.annotation.Annotation;
+import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 
@@ -40,14 +42,13 @@ import org.springframework.core.codec.CodecException;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferFactory;
 import org.springframework.http.MediaType;
-import org.springframework.http.codec.ServerHttpEncoder;
+import org.springframework.http.codec.HttpEncoder;
 import org.springframework.http.converter.json.Jackson2ObjectMapperBuilder;
 import org.springframework.http.server.reactive.ServerHttpRequest;
 import org.springframework.http.server.reactive.ServerHttpResponse;
 import org.springframework.util.Assert;
 import org.springframework.util.MimeType;
 
-import static org.springframework.http.MediaType.APPLICATION_STREAM_JSON;
 
 /**
  * Encode from an {@code Object} stream to a byte stream of JSON objects,
@@ -58,27 +59,41 @@ import static org.springframework.http.MediaType.APPLICATION_STREAM_JSON;
  * @since 5.0
  * @see Jackson2JsonDecoder
  */
-public class Jackson2JsonEncoder extends Jackson2CodecSupport implements ServerHttpEncoder<Object> {
+public class Jackson2JsonEncoder extends Jackson2CodecSupport implements HttpEncoder<Object> {
+
+	private final List<MediaType> streamingMediaTypes = new ArrayList<>(1);
 
 	private final PrettyPrinter ssePrettyPrinter;
 
 
+
 	public Jackson2JsonEncoder() {
 		this(Jackson2ObjectMapperBuilder.json().build());
 	}
 
 	public Jackson2JsonEncoder(ObjectMapper mapper) {
 		super(mapper);
-		DefaultPrettyPrinter prettyPrinter = new DefaultPrettyPrinter();
-		prettyPrinter.indentObjectsWith(new DefaultIndenter("  ", "\ndata:"));
-		this.ssePrettyPrinter = prettyPrinter;
+		this.streamingMediaTypes.add(MediaType.APPLICATION_STREAM_JSON);
+		this.ssePrettyPrinter = initSsePrettyPrinter();
 	}
 
+	private static PrettyPrinter initSsePrettyPrinter() {
+		DefaultPrettyPrinter printer = new DefaultPrettyPrinter();
+		printer.indentObjectsWith(new DefaultIndenter("  ", "\ndata:"));
+		return printer;
+	}
 
-	@Override
-	public boolean canEncode(ResolvableType elementType, MimeType mimeType) {
-		return this.mapper.canSerialize(elementType.getRawClass()) &&
-				(mimeType == null || JSON_MIME_TYPES.stream().anyMatch(m -> m.isCompatibleWith(mimeType)));
+
+	/**
+	 * Configure "streaming" media types for which flushing should be performed
+	 * automatically vs at the end of the stream.
+	 * <p>By default this is set to {@link MediaType#APPLICATION_STREAM_JSON}.
+	 * @param mediaTypes one or more media types to add to the list
+	 * @see HttpEncoder#getStreamingMediaTypes()
+	 */
+	public void setStreamingMediaTypes(List<MediaType> mediaTypes) {
+		this.streamingMediaTypes.clear();
+		this.streamingMediaTypes.addAll(mediaTypes);
 	}
 
 	@Override
@@ -86,6 +101,13 @@ public class Jackson2JsonEncoder extends Jackson2CodecSupport implements ServerH
 		return JSON_MIME_TYPES;
 	}
 
+
+	@Override
+	public boolean canEncode(ResolvableType elementType, MimeType mimeType) {
+		return this.mapper.canSerialize(elementType.getRawClass()) &&
+				(mimeType == null || JSON_MIME_TYPES.stream().anyMatch(m -> m.isCompatibleWith(mimeType)));
+	}
+
 	@Override
 	public Flux<DataBuffer> encode(Publisher<?> inputStream, DataBufferFactory bufferFactory,
 			ResolvableType elementType, MimeType mimeType, Map<String, Object> hints) {
@@ -98,7 +120,7 @@ public class Jackson2JsonEncoder extends Jackson2CodecSupport implements ServerH
 			return Flux.from(inputStream).map(value ->
 					encodeValue(value, mimeType, bufferFactory, elementType, hints));
 		}
-		else if (APPLICATION_STREAM_JSON.isCompatibleWith(mimeType)) {
+		else if (MediaType.APPLICATION_STREAM_JSON.isCompatibleWith(mimeType)) {
 			return Flux.from(inputStream).map(value -> {
 				DataBuffer buffer = encodeValue(value, mimeType, bufferFactory, elementType, hints);
 				buffer.write(new byte[]{'\n'});
@@ -147,7 +169,12 @@ public class Jackson2JsonEncoder extends Jackson2CodecSupport implements ServerH
 	}
 
 
-	// ServerHttpEncoder...
+	// HttpEncoder...
+
+	@Override
+	public List<MediaType> getStreamingMediaTypes() {
+		return Collections.unmodifiableList(this.streamingMediaTypes);
+	}
 
 	@Override
 	public Map<String, Object> getEncodeHints(ResolvableType actualType, ResolvableType elementType,