Add NDJSON and deprecate application/stream+json

Closes gh-21283

spring-web/src/jmh/java/org/springframework/http/MediaTypeBenchmark.java

@@ -88,7 +88,7 @@ public class MediaTypeBenchmark {
 					"application/problem+json",
 					"application/xhtml+xml",
 					"application/rss+xml",
-					"application/stream+json",
+					"application/x-ndjson",
 					"application/xml;q=0.9",
 					"application/atom+xml",
 					"application/cbor",

spring-web/src/main/java/org/springframework/http/MediaType.java

@@ -216,16 +216,36 @@ public class MediaType extends MimeType implements Serializable {
 	 */
 	public static final String APPLICATION_RSS_XML_VALUE = "application/rss+xml";
 
+	/**
+	 * Public constant media type for {@code application/x-ndjson}.
+	 * @since 5.3
+	 */
+	public static final MediaType APPLICATION_NDJSON;
+
+	/**
+	 * A String equivalent of {@link MediaType#APPLICATION_NDJSON}.
+	 * @since 5.3
+	 */
+	public static final String APPLICATION_NDJSON_VALUE = "application/x-ndjson";
+
 	/**
 	 * Public constant media type for {@code application/stream+json}.
+	 * @deprecated as of 5.3, see notice on {@link #APPLICATION_STREAM_JSON_VALUE}.
 	 * @since 5.0
 	 */
+	@Deprecated
 	public static final MediaType APPLICATION_STREAM_JSON;
 
 	/**
 	 * A String equivalent of {@link MediaType#APPLICATION_STREAM_JSON}.
+	 * @deprecated as of 5.3 since it originates from the W3C Activity Streams
+	 * specification which has a more specific purpose and has been since
+	 * replaced with a different mime type. Use {@link #APPLICATION_NDJSON} as
+	 * a replacement or any other line-delimited JSON format (e.g. JSON Lines,
+	 * JSON Text Sequences).
 	 * @since 5.0
 	 */
+	@Deprecated
 	public static final String APPLICATION_STREAM_JSON_VALUE = "application/stream+json";
 
 	/**
@@ -378,6 +398,7 @@ public class MediaType extends MimeType implements Serializable {
 		APPLICATION_FORM_URLENCODED = new MediaType("application", "x-www-form-urlencoded");
 		APPLICATION_JSON = new MediaType("application", "json");
 		APPLICATION_JSON_UTF8 = new MediaType("application", "json", StandardCharsets.UTF_8);
+		APPLICATION_NDJSON = new MediaType("application", "x-ndjson");
 		APPLICATION_OCTET_STREAM = new MediaType("application", "octet-stream");
 		APPLICATION_PDF = new MediaType("application", "pdf");
 		APPLICATION_PROBLEM_JSON = new MediaType("application", "problem+json");

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

@@ -67,15 +67,9 @@ public abstract class AbstractJackson2Encoder extends Jackson2CodecSupport imple
 
 	private static final byte[] NEWLINE_SEPARATOR = {'\n'};
 
-	private static final Map<MediaType, byte[]> STREAM_SEPARATORS;
-
 	private static final Map<String, JsonEncoding> ENCODINGS;
 
 	static {
-		STREAM_SEPARATORS = new HashMap<>(4);
-		STREAM_SEPARATORS.put(MediaType.APPLICATION_STREAM_JSON, NEWLINE_SEPARATOR);
-		STREAM_SEPARATORS.put(MediaType.parseMediaType("application/stream+x-jackson-smile"), new byte[0]);
-
 		ENCODINGS = new HashMap<>(JsonEncoding.values().length + 1);
 		for (JsonEncoding encoding : JsonEncoding.values()) {
 			ENCODINGS.put(encoding.getJavaName(), encoding);
@@ -98,9 +92,6 @@ public abstract class AbstractJackson2Encoder extends Jackson2CodecSupport imple
 	/**
 	 * 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 HttpMessageEncoder#getStreamingMediaTypes()
 	 */
 	public void setStreamingMediaTypes(List<MediaType> mediaTypes) {
 		this.streamingMediaTypes.clear();
@@ -138,7 +129,7 @@ public abstract class AbstractJackson2Encoder extends Jackson2CodecSupport imple
 					.flux();
 		}
 		else {
-			byte[] separator = streamSeparator(mimeType);
+			byte[] separator = getStreamingMediaTypeSeparator(mimeType);
 			if (separator != null) { // streaming
 				try {
 					ObjectWriter writer = createObjectWriter(elementType, mimeType, hints);
@@ -268,11 +259,18 @@ public abstract class AbstractJackson2Encoder extends Jackson2CodecSupport imple
 		return writer;
 	}
 
+	/**
+	 * Return the separator to use for the given mime type.
+	 * <p>By default, this method returns new line {@code "\n"} if the given
+	 * mime type is one of the configured {@link #setStreamingMediaTypes(List)
+	 * streaming} mime types.
+	 * @since 5.3
+	 */
 	@Nullable
-	private byte[] streamSeparator(@Nullable MimeType mimeType) {
+	protected byte[] getStreamingMediaTypeSeparator(@Nullable MimeType mimeType) {
 		for (MediaType streamingMediaType : this.streamingMediaTypes) {
 			if (streamingMediaType.isCompatibleWith(mimeType)) {
-				return STREAM_SEPARATORS.getOrDefault(streamingMediaType, NEWLINE_SEPARATOR);
+				return NEWLINE_SEPARATOR;
 			}
 		}
 		return null;

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

@@ -35,6 +35,7 @@ import org.springframework.core.MethodParameter;
 import org.springframework.core.ResolvableType;
 import org.springframework.core.codec.Hints;
 import org.springframework.http.HttpLogging;
+import org.springframework.http.MediaType;
 import org.springframework.http.server.reactive.ServerHttpRequest;
 import org.springframework.http.server.reactive.ServerHttpResponse;
 import org.springframework.lang.Nullable;
@@ -72,8 +73,9 @@ public abstract class Jackson2CodecSupport {
 
 	private static final List<MimeType> DEFAULT_MIME_TYPES = Collections.unmodifiableList(
 			Arrays.asList(
-					new MimeType("application", "json"),
-					new MimeType("application", "*+json")));
+					MediaType.APPLICATION_JSON,
+					new MediaType("application", "*+json"),
+					MediaType.APPLICATION_NDJSON));
 
 
 	protected final Log logger = HttpLogging.forLogName(getClass());

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2018 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -16,7 +16,7 @@
 
 package org.springframework.http.codec.json;
 
-import java.util.Collections;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 
@@ -54,9 +54,10 @@ public class Jackson2JsonEncoder extends AbstractJackson2Encoder {
 		this(Jackson2ObjectMapperBuilder.json().build());
 	}
 
+	@SuppressWarnings("deprecation")
 	public Jackson2JsonEncoder(ObjectMapper mapper, MimeType... mimeTypes) {
 		super(mapper, mimeTypes);
-		setStreamingMediaTypes(Collections.singletonList(MediaType.APPLICATION_STREAM_JSON));
+		setStreamingMediaTypes(Arrays.asList(MediaType.APPLICATION_NDJSON, MediaType.APPLICATION_STREAM_JSON));
 		this.ssePrettyPrinter = initSsePrettyPrinter();
 	}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -25,6 +25,7 @@ import reactor.core.publisher.Flux;
 
 import org.springframework.http.MediaType;
 import org.springframework.http.converter.json.Jackson2ObjectMapperBuilder;
+import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.MimeType;
 
@@ -43,6 +44,11 @@ public class Jackson2SmileEncoder extends AbstractJackson2Encoder {
 			new MimeType("application", "x-jackson-smile"),
 			new MimeType("application", "*+x-jackson-smile")};
 
+	private static final MimeType STREAM_MIME_TYPE =
+			MediaType.parseMediaType("application/stream+x-jackson-smile");
+
+	private static final byte[] STREAM_SEPARATOR = new byte[0];
+
 
 	public Jackson2SmileEncoder() {
 		this(Jackson2ObjectMapperBuilder.smile().build(), DEFAULT_SMILE_MIME_TYPES);
@@ -54,4 +60,22 @@ public class Jackson2SmileEncoder extends AbstractJackson2Encoder {
 		setStreamingMediaTypes(Collections.singletonList(new MediaType("application", "stream+x-jackson-smile")));
 	}
 
+
+	/**
+	 * Return the separator to use for the given mime type.
+	 * <p>By default, this method returns a single byte 0 if the given
+	 * mime type is one of the configured {@link #setStreamingMediaTypes(List)
+	 * streaming} mime types.
+	 * @since 5.3
+	 */
+	@Nullable
+	@Override
+	protected byte[] getStreamingMediaTypeSeparator(@Nullable MimeType mimeType) {
+		for (MediaType streamingMediaType : getStreamingMediaTypes()) {
+			if (streamingMediaType.isCompatibleWith(mimeType)) {
+				return STREAM_SEPARATOR;
+			}
+		}
+		return null;
+	}
 }

spring-web/src/test/java/org/springframework/http/codec/json/Jackson2JsonDecoderTests.java

@@ -51,6 +51,7 @@ import org.springframework.web.testfixture.xml.Pojo;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 import static org.springframework.http.MediaType.APPLICATION_JSON;
+import static org.springframework.http.MediaType.APPLICATION_NDJSON;
 import static org.springframework.http.MediaType.APPLICATION_STREAM_JSON;
 import static org.springframework.http.MediaType.APPLICATION_XML;
 import static org.springframework.http.codec.json.Jackson2CodecSupport.JSON_VIEW_HINT;
@@ -77,6 +78,7 @@ public class Jackson2JsonDecoderTests extends AbstractDecoderTests<Jackson2JsonD
 	@Test
 	public void canDecode() {
 		assertThat(decoder.canDecode(ResolvableType.forClass(Pojo.class), APPLICATION_JSON)).isTrue();
+		assertThat(decoder.canDecode(ResolvableType.forClass(Pojo.class), APPLICATION_NDJSON)).isTrue();
 		assertThat(decoder.canDecode(ResolvableType.forClass(Pojo.class), APPLICATION_STREAM_JSON)).isTrue();
 		assertThat(decoder.canDecode(ResolvableType.forClass(Pojo.class), null)).isTrue();
 

spring-web/src/test/java/org/springframework/http/codec/json/Jackson2JsonEncoderTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -47,6 +47,7 @@ import static java.util.Collections.singletonMap;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 import static org.springframework.http.MediaType.APPLICATION_JSON;
+import static org.springframework.http.MediaType.APPLICATION_NDJSON;
 import static org.springframework.http.MediaType.APPLICATION_OCTET_STREAM;
 import static org.springframework.http.MediaType.APPLICATION_STREAM_JSON;
 import static org.springframework.http.MediaType.APPLICATION_XML;
@@ -66,6 +67,7 @@ public class Jackson2JsonEncoderTests extends AbstractEncoderTests<Jackson2JsonE
 	public void canEncode() {
 		ResolvableType pojoType = ResolvableType.forClass(Pojo.class);
 		assertThat(this.encoder.canEncode(pojoType, APPLICATION_JSON)).isTrue();
+		assertThat(this.encoder.canEncode(pojoType, APPLICATION_NDJSON)).isTrue();
 		assertThat(this.encoder.canEncode(pojoType, APPLICATION_STREAM_JSON)).isTrue();
 		assertThat(this.encoder.canEncode(pojoType, null)).isTrue();
 

spring-web/src/test/java/org/springframework/http/codec/json/Jackson2TokenizerTests.java

@@ -197,6 +197,43 @@ public class Jackson2TokenizerTests extends AbstractLeakCheckingTests {
 		testTokenize(asList("[1", ",2,", "3]"), asList("1", "2", "3"), true);
 	}
 
+	@Test
+	void tokenizeStream() {
+
+		// NDJSON (Newline Delimited JSON), JSON Lines
+		testTokenize(
+				asList(
+						"{\"id\":1,\"name\":\"Robert\"}",
+						"\n",
+						"{\"id\":2,\"name\":\"Raide\"}",
+						"\n",
+						"{\"id\":3,\"name\":\"Ford\"}"
+				),
+				asList(
+						"{\"id\":1,\"name\":\"Robert\"}",
+						"{\"id\":2,\"name\":\"Raide\"}",
+						"{\"id\":3,\"name\":\"Ford\"}"
+				),
+				true);
+
+		// JSON Sequence with newline separator
+		testTokenize(
+				asList(
+						"\n",
+						"{\"id\":1,\"name\":\"Robert\"}",
+						"\n",
+						"{\"id\":2,\"name\":\"Raide\"}",
+						"\n",
+						"{\"id\":3,\"name\":\"Ford\"}"
+				),
+				asList(
+						"{\"id\":1,\"name\":\"Robert\"}",
+						"{\"id\":2,\"name\":\"Raide\"}",
+						"{\"id\":3,\"name\":\"Ford\"}"
+				),
+				true);
+	}
+
 	private void testTokenize(List<String> input, List<String> output, boolean tokenize) {
 		StepVerifier.FirstStep<String> builder = StepVerifier.create(decode(input, tokenize, -1));
 		output.forEach(expected -> builder.assertNext(actual -> {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -35,8 +35,8 @@ import org.springframework.web.server.adapter.WebHttpHandlerBuilder;
 import org.springframework.web.testfixture.http.server.reactive.bootstrap.AbstractHttpHandlerIntegrationTests;
 import org.springframework.web.testfixture.http.server.reactive.bootstrap.HttpServer;
 
-import static org.springframework.http.MediaType.APPLICATION_STREAM_JSON;
-import static org.springframework.http.MediaType.APPLICATION_STREAM_JSON_VALUE;
+import static org.springframework.http.MediaType.APPLICATION_NDJSON;
+import static org.springframework.http.MediaType.APPLICATION_NDJSON_VALUE;
 
 /**
  * @author Sebastien Deleuze
@@ -71,7 +71,7 @@ class JacksonStreamingIntegrationTests extends AbstractHttpHandlerIntegrationTes
 
 		Flux<Person> result = this.webClient.get()
 				.uri("/stream")
-				.accept(APPLICATION_STREAM_JSON)
+				.accept(APPLICATION_NDJSON)
 				.retrieve()
 				.bodyToFlux(Person.class);
 
@@ -105,7 +105,7 @@ class JacksonStreamingIntegrationTests extends AbstractHttpHandlerIntegrationTes
 	static class JacksonStreamingController {
 
 		@GetMapping(value = "/stream",
-				produces = { APPLICATION_STREAM_JSON_VALUE, "application/stream+x-jackson-smile" })
+				produces = { APPLICATION_NDJSON_VALUE, "application/stream+x-jackson-smile" })
 		Flux<Person> person() {
 			return testInterval(Duration.ofMillis(100), 50).map(l -> new Person("foo " + l));
 		}

spring-webflux/src/test/java/org/springframework/web/reactive/result/view/HttpMessageWriterViewTests.java

@@ -53,10 +53,11 @@ public class HttpMessageWriterViewTests {
 
 
 	@Test
-	public void supportedMediaTypes() throws Exception {
-		assertThat(this.view.getSupportedMediaTypes()).isEqualTo(Arrays.asList(
+	public void supportedMediaTypes() {
+		assertThat(this.view.getSupportedMediaTypes()).containsExactly(
 				MediaType.APPLICATION_JSON,
-				MediaType.parseMediaType("application/*+json")));
+				MediaType.parseMediaType("application/*+json"),
+				MediaType.APPLICATION_NDJSON);
 	}
 
 	@Test

spring-webmvc/src/main/java/org/springframework/web/servlet/mvc/method/annotation/ReactiveTypeHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -19,6 +19,7 @@ package org.springframework.web.servlet.mvc.method.annotation;
 import java.io.IOException;
 import java.time.Duration;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
 import java.util.Optional;
@@ -73,8 +74,12 @@ class ReactiveTypeHandler {
 
 	private static final long STREAMING_TIMEOUT_VALUE = -1;
 
+	@SuppressWarnings("deprecation")
+	private static final List<MediaType> JSON_STREAMING_MEDIA_TYPES =
+			Arrays.asList(MediaType.APPLICATION_NDJSON, MediaType.APPLICATION_STREAM_JSON);
+
+	private static final Log logger = LogFactory.getLog(ReactiveTypeHandler.class);
 
-	private static Log logger = LogFactory.getLog(ReactiveTypeHandler.class);
 
 	private final ReactiveAdapterRegistry adapterRegistry;
 
@@ -144,11 +149,15 @@ class ReactiveTypeHandler {
 				new TextEmitterSubscriber(emitter, this.taskExecutor).connect(adapter, returnValue);
 				return emitter;
 			}
-			if (mediaTypes.stream().anyMatch(MediaType.APPLICATION_STREAM_JSON::includes)) {
-				logExecutorWarning(returnType);
-				ResponseBodyEmitter emitter = getEmitter(MediaType.APPLICATION_STREAM_JSON);
-				new JsonEmitterSubscriber(emitter, this.taskExecutor).connect(adapter, returnValue);
-				return emitter;
+			for (MediaType type : mediaTypes) {
+				for (MediaType streamingType : JSON_STREAMING_MEDIA_TYPES) {
+					if (streamingType.includes(type)) {
+						logExecutorWarning(returnType);
+						ResponseBodyEmitter emitter = getEmitter(streamingType);
+						new JsonEmitterSubscriber(emitter, this.taskExecutor).connect(adapter, returnValue);
+						return emitter;
+					}
+				}
 			}
 		}
 

spring-webmvc/src/main/resources/org/springframework/web/servlet/config/spring-mvc.xsd

@@ -250,7 +250,7 @@
 	By default, a SimpleAsyncTaskExecutor is used which does not re-use threads and is not recommended for production.
 
 	As of 5.0 this executor is also used when a controller returns a reactive type that does streaming
-	(e.g. "text/event-stream" or "application/stream+json") for the blocking writes to the
+	(e.g. "text/event-stream" or "application/x-ndjson") for the blocking writes to the
 	"javax.servlet.ServletOutputStream".
 
 								]]></xsd:documentation>

spring-webmvc/src/test/java/org/springframework/web/servlet/mvc/method/annotation/ReactiveTypeHandlerTests.java

@@ -225,7 +225,7 @@ public class ReactiveTypeHandlerTests {
 	@Test
 	public void writeStreamJson() throws Exception {
 
-		this.servletRequest.addHeader("Accept", "application/stream+json");
+		this.servletRequest.addHeader("Accept", "application/x-ndjson");
 
 		Sinks.StandaloneFluxSink<Bar> sink = Sinks.unicast();
 		ResponseBodyEmitter emitter = handleValue(sink.asFlux(), Flux.class, forClass(Bar.class));
@@ -243,7 +243,7 @@ public class ReactiveTypeHandlerTests {
 		sink.next(bar2);
 		sink.complete();
 
-		assertThat(message.getHeaders().getContentType().toString()).isEqualTo("application/stream+json");
+		assertThat(message.getHeaders().getContentType().toString()).isEqualTo("application/x-ndjson");
 		assertThat(emitterHandler.getValues()).isEqualTo(Arrays.asList(bar1, "\n", bar2, "\n"));
 	}
 

src/docs/asciidoc/testing-webtestclient.adoc

@@ -378,7 +378,7 @@ You can also use https://github.com/jayway/JsonPath[JSONPath] expressions, as fo
 [[webtestclient-stream]]
 === Streaming Responses
 
-To test infinite streams (for example, `"text/event-stream"` or `"application/stream+json"`),
+To test infinite streams (for example, `"text/event-stream"` or `"application/x-ndjson"`),
 you need to exit the chained API (by using `returnResult`), immediately after the response status
 and header assertions, as the following example shows:
 

src/docs/asciidoc/web/webflux.adoc

@@ -755,9 +755,9 @@ into ``TokenBuffer``'s each representing a JSON object.
 * When decoding to a single-value publisher (e.g. `Mono`), there is one `TokenBuffer`.
 * When decoding to a multi-value publisher (e.g. `Flux`), each `TokenBuffer` is passed to
 the `ObjectMapper` as soon as enough bytes are received for a fully formed object. The
-input content can be a JSON array, or
-https://en.wikipedia.org/wiki/JSON_streaming[line-delimited JSON] if the content-type is
-`application/stream+json`.
+input content can be a JSON array, or any
+https://en.wikipedia.org/wiki/JSON_streaming[line-delimited JSON] format such as NDJSON,
+JSON Lines, or JSON Text Sequences.
 
 The `Jackson2Encoder` works as follows:
 
@@ -766,9 +766,10 @@ The `Jackson2Encoder` works as follows:
 * For a multi-value publisher with `application/json`, by default collect the values with
 `Flux#collectToList()` and then serialize the resulting collection.
 * For a multi-value publisher with a streaming media type such as
-`application/stream+json` or `application/stream+x-jackson-smile`, encode, write, and
+`application/x-ndjson` or `application/stream+x-jackson-smile`, encode, write, and
 flush each value individually using a
-https://en.wikipedia.org/wiki/JSON_streaming[line-delimited JSON] format.
+https://en.wikipedia.org/wiki/JSON_streaming[line-delimited JSON] format. Other
+streaming media types may be registered with the encoder.
 * For SSE the `Jackson2Encoder` is invoked per event and the output is flushed to ensure
 delivery without delay.
 
@@ -852,7 +853,7 @@ To configure all three in WebFlux, you'll need to supply a pre-configured instan
 [.small]#<<web.adoc#mvc-ann-async-http-streaming, Web MVC>>#
 
 When streaming to the HTTP response (for example, `text/event-stream`,
-`application/stream+json`), it is important to send data periodically, in order to
+`application/x-ndjson`), it is important to send data periodically, in order to
 reliably detect a disconnected client sooner rather than later. Such a send could be a
 comment-only, empty SSE event or any other "no-op" data that would effectively serve as
 a heartbeat.

src/docs/asciidoc/web/webmvc.adoc

@@ -4581,7 +4581,7 @@ Reactive return values are handled as follows:
 
 * A single-value promise is adapted to, similar to using `DeferredResult`. Examples
 include `Mono` (Reactor) or `Single` (RxJava).
-* A multi-value stream with a streaming media type (such as `application/stream+json`
+* A multi-value stream with a streaming media type (such as `application/x-ndjson`
 or `text/event-stream`) is adapted to, similar to using `ResponseBodyEmitter` or
 `SseEmitter`. Examples include `Flux` (Reactor) or `Observable` (RxJava).
 Applications can also return `Flux<ServerSentEvent>` or `Observable<ServerSentEvent>`.