Add Server-Sent Events support

Flux<SseEvent> is Spring Web Reactive equivalent to Spring MVC
SseEmitter type. It allows to send Server-Sent Events in a reactive way.
Sending Flux<String> or Flux<Pojo> is equivalent to sending
Flux<SseEvent> with the data property set to the String or
Pojo value. For example:

@RestController
public class SseController {

	@RequestMapping("/sse/string")
	Flux<String> string() {
		return Flux.interval(Duration.ofSeconds(1)).map(l -> "foo " + l);
	}

	@RequestMapping("/sse/person")
	Flux<Person> person() {
		return Flux.interval(Duration.ofSeconds(1)).map(l -> new Person(Long.toString(l), "foo", "bar"));
	}

	@RequestMapping("/sse-raw")
	Flux<SseEvent> sse() {
		return Flux.interval(Duration.ofSeconds(1)).map(l -> {
			SseEvent event = new SseEvent();
			event.setId(Long.toString(l));
			event.setData("foo\nbar");
			event.setComment("bar\nbaz");
			return event;
		});
	}
}

spring-web-reactive/src/main/java/org/springframework/core/codec/support/SseEventEncoder.java

+/*
+ * Copyright 2002-2016 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.core.codec.support;
+
+import java.util.List;
+import java.util.Optional;
+
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+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.core.io.buffer.FlushingDataBuffer;
+import org.springframework.util.Assert;
+import org.springframework.util.MimeType;
+import org.springframework.web.reactive.sse.SseEvent;
+
+/**
+ * An encoder for {@link SseEvent}s that also supports any other kind of {@link Object}
+ * (in that case, the object will be the data of the {@link SseEvent}).
+ * @author Sebastien Deleuze
+ */
+public class SseEventEncoder extends AbstractEncoder<Object> {
+
+	private final Encoder<String> stringEncoder;
+
+	private final List<Encoder<?>> dataEncoders;
+
+
+	public SseEventEncoder(Encoder<String> stringEncoder, List<Encoder<?>> dataEncoders) {
+		super(new MimeType("text", "event-stream"));
+		Assert.notNull(stringEncoder, "'stringEncoder' must not be null");
+		Assert.notNull(dataEncoders, "'dataEncoders' must not be null");
+		this.stringEncoder = stringEncoder;
+		this.dataEncoders = dataEncoders;
+	}
+
+	@Override
+	public Flux<DataBuffer> encode(Publisher<?> inputStream, DataBufferFactory bufferFactory, ResolvableType type, MimeType sseMimeType, Object... hints) {
+
+		return Flux.from(inputStream).flatMap(input -> {
+			SseEvent event = (SseEvent.class.equals(type.getRawClass()) ? (SseEvent)input : new SseEvent(input));
+
+			StringBuilder sb = new StringBuilder();
+
+			if (event.getId() != null) {
+				sb.append("id:");
+				sb.append(event.getId());
+				sb.append("\n");
+			}
+
+			if (event.getName() != null) {
+				sb.append("event:");
+				sb.append(event.getName());
+				sb.append("\n");
+			}
+
+			if (event.getReconnectTime() != null) {
+				sb.append("retry:");
+				sb.append(event.getReconnectTime().toString());
+				sb.append("\n");
+			}
+
+			if (event.getComment() != null) {
+				sb.append(":");
+				sb.append(event.getComment().replaceAll("\\n", "\n:"));
+				sb.append("\n");
+			}
+
+			Object data = event.getData();
+			Flux<DataBuffer> dataBuffer = Flux.empty();
+			MimeType stringMimeType = this.stringEncoder.getEncodableMimeTypes().get(0);
+			MimeType mimeType = (event.getMimeType() == null ?
+	                (data instanceof String ? stringMimeType : new MimeType("*")) : event.getMimeType());
+			if (data != null) {
+				sb.append("data:");
+				if (data instanceof String && mimeType.isCompatibleWith(stringMimeType)) {
+					sb.append(((String)data).replaceAll("\\n", "\ndata:")).append("\n");
+				}
+				else {
+					Optional<Encoder<?>> encoder = dataEncoders
+						.stream()
+						.filter(e -> e.canEncode(ResolvableType.forClass(data.getClass()), mimeType))
+						.findFirst();
+
+					if (encoder.isPresent()) {
+						dataBuffer = ((Encoder<Object>)encoder.get())
+								.encode(Mono.just(data), bufferFactory, ResolvableType.forClass(data.getClass()), mimeType)
+								.concatWith(encodeString("\n", bufferFactory, stringMimeType));
+					}
+					else {
+						throw new CodecException("No suitable encoder found!");
+					}
+				}
+			}
+
+			return Flux
+					.concat(encodeString(sb.toString(), bufferFactory, stringMimeType), dataBuffer)
+					.reduce((buf1, buf2) -> buf1.write(buf2))
+					.concatWith(encodeString("\n", bufferFactory, stringMimeType).map(b -> new FlushingDataBuffer(b)));
+		});
+
+	}
+
+	private Flux<DataBuffer> encodeString(String str, DataBufferFactory bufferFactory, MimeType mimeType) {
+		return stringEncoder.encode(Mono.just(str), bufferFactory, ResolvableType.forClass(String.class), mimeType);
+	}
+
+}

spring-web-reactive/src/main/java/org/springframework/http/converter/reactive/SseHttpMessageConverter.java

+/*
+ * Copyright 2002-2016 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.http.converter.reactive;
+
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.core.ResolvableType;
+import org.springframework.core.codec.Encoder;
+import org.springframework.core.codec.support.JacksonJsonEncoder;
+import org.springframework.core.codec.support.SseEventEncoder;
+import org.springframework.core.codec.support.StringEncoder;
+import org.springframework.http.MediaType;
+import org.springframework.http.ReactiveHttpOutputMessage;
+import org.springframework.web.reactive.sse.SseEvent;
+
+/**
+ * Implementation of {@link HttpMessageConverter} that can stream Server-Sent Events
+ * response.
+ *
+ * It allows to write {@code Flux<ServerSentEvent>}, which is Spring Web Reactive equivalent
+ * to Spring MVC {@code SseEmitter}.
+ *
+ * Sending {@code Flux<String>} or {@code Flux<Pojo>} is equivalent to sending
+ * {@code Flux<SseEvent>} with the {@code data} property set to the {@code String} or
+ * {@code Pojo} value.
+ *
+ * @author Sebastien Deleuze
+ * @see SseEvent
+ * @see <a href="https://www.w3.org/TR/eventsource/">Server-Sent Events W3C recommandation</a>
+ */
+public class SseHttpMessageConverter extends CodecHttpMessageConverter<Object> {
+
+	/**
+	 * Default constructor that creates a new instance configured with {@link StringEncoder}
+	 * and {@link JacksonJsonEncoder} encoders.
+	 */
+	public SseHttpMessageConverter() {
+		this(new StringEncoder(), Arrays.asList(new JacksonJsonEncoder()));
+	}
+
+	public SseHttpMessageConverter(Encoder<String> stringEncoder, List<Encoder<?>> dataEncoders) {
+		// 1 SseEvent element = 1 DataBuffer element so flush after each element
+		super(new SseEventEncoder(stringEncoder, dataEncoders), null);
+	}
+
+	@Override
+	public Mono<Void> write(Publisher<?> inputStream, ResolvableType type,
+			MediaType contentType, ReactiveHttpOutputMessage outputMessage) {
+
+		outputMessage.getHeaders().add("Content-Type", "text/event-stream");
+		// Keep the SSE connection open even for cold stream in order to avoid unexpected Browser reconnection
+		return super.write(Flux.from(inputStream).concatWith(Flux.never()), type, contentType, outputMessage);
+	}
+
+}

spring-web-reactive/src/main/java/org/springframework/web/reactive/sse/SseEvent.java

+/*
+ * Copyright 2002-2016 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.web.reactive.sse;
+
+import org.springframework.http.converter.reactive.SseHttpMessageConverter;
+import org.springframework.util.MimeType;
+
+/**
+ * Represent a Server-Sent Event.
+ *
+ * <p>{@code Flux<SseEvent>} is Spring Web Reactive equivalent to Spring MVC
+ * {@code SseEmitter} type. It allows to send Server-Sent Events in a reactive way.
+ *
+ * @author Sebastien Deleuze
+ * @see SseHttpMessageConverter
+ * @see <a href="https://www.w3.org/TR/eventsource/">Server-Sent Events W3C recommandation</a>
+ */
+public class SseEvent {
+
+	private String id;
+
+	private String name;
+
+	private Object data;
+
+	private MimeType mimeType;
+
+	private Long reconnectTime;
+
+	private String comment;
+
+	/**
+	 * Create an empty instance.
+	 */
+	public SseEvent() {
+	}
+
+	/**
+	 * Create an instance with the provided {@code data}.
+	 */
+	public SseEvent(Object data) {
+		this.data = data;
+	}
+
+	/**
+	 * Create an instance with the provided {@code data} and {@code mediaType}.
+	 */
+	public SseEvent(Object data, MimeType mimeType) {
+		this.data = data;
+		this.mimeType = mimeType;
+	}
+
+	/**
+	 * Set the {@code id} SSE field
+	 */
+	public void setId(String id) {
+		this.id = id;
+	}
+
+	/**
+	 * @see #setId(String)
+	 */
+	public String getId() {
+		return id;
+	}
+
+	/**
+	 * Set the {@code event} SSE field
+	 */
+	public void setName(String name) {
+		this.name = name;
+	}
+
+	/**
+	 * @see #setName(String)
+	 */
+	public String getName() {
+		return name;
+	}
+
+	/**
+	 * Set {@code data} SSE field. If a multiline {@code String} is provided, it will be
+	 * turned into multiple {@code data} field lines by  as
+	 * defined in Server-Sent Events W3C recommandation.
+	 *
+	 * If no {@code mediaType} is defined, default {@link SseHttpMessageConverter} will:
+	 *  - Turn single line {@code String} to a single {@code data} field
+	 *  - Turn multiline line {@code String} to multiple {@code data} fields
+	 *  - Serialize other {@code Object} as JSON
+	 *
+	 * @see #setMimeType(MimeType)
+	 */
+	public void setData(Object data) {
+		this.data = data;
+	}
+
+	/**
+	 * @see #setData(Object)
+	 */
+	public Object getData() {
+		return data;
+	}
+
+	/**
+	 * Set the {@link MimeType} used to serialize the {@code data}.
+	 * {@link SseHttpMessageConverter} should be configured with the relevant encoder to be
+	 * able to serialize it.
+	 */
+	public void setMimeType(MimeType mimeType) {
+		this.mimeType = mimeType;
+	}
+
+	/**
+	 * @see #setMimeType(MimeType)
+	 */
+	public MimeType getMimeType() {
+		return mimeType;
+	}
+
+	/**
+	 * Set the {@code retry} SSE field
+	 */
+	public void setReconnectTime(Long reconnectTime) {
+		this.reconnectTime = reconnectTime;
+	}
+
+	/**
+	 * @see #setReconnectTime(Long)
+	 */
+	public Long getReconnectTime() {
+		return reconnectTime;
+	}
+
+	/**
+	 * Set SSE comment. If a multiline comment is provided, it will be turned into multiple
+	 * SSE comment lines by {@link SseHttpMessageConverter} as defined in Server-Sent Events W3C
+	 * recommandation.
+	 */
+	public void setComment(String comment) {
+		this.comment = comment;
+	}
+
+	/**
+	 * @see #setComment(String)
+	 */
+	public String getComment() {
+		return comment;
+	}
+
+}

spring-web-reactive/src/test/java/org/springframework/core/codec/support/SseEventEncoderTests.java

+/*
+ * Copyright 2002-2016 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.core.codec.support;
+
+import java.util.Arrays;
+
+import org.junit.Test;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.test.TestSubscriber;
+
+import org.springframework.core.ResolvableType;
+import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.util.MimeType;
+import org.springframework.web.reactive.sse.SseEvent;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+
+/**
+ * @author Sebastien Deleuze
+ */
+public class SseEventEncoderTests extends AbstractDataBufferAllocatingTestCase {
+
+	@Test
+	public void nullMimeType() {
+		SseEventEncoder encoder = new SseEventEncoder(new StringEncoder(), Arrays.asList(new JacksonJsonEncoder()));
+		assertTrue(encoder.canEncode(ResolvableType.forClass(Object.class), null));
+	}
+
+	@Test
+	public void unsupportedMimeType() {
+		SseEventEncoder encoder = new SseEventEncoder(new StringEncoder(), Arrays.asList(new JacksonJsonEncoder()));
+		assertFalse(encoder.canEncode(ResolvableType.forClass(Object.class), new MimeType("foo", "bar")));
+	}
+
+	@Test
+	public void supportedMimeType() {
+		SseEventEncoder encoder = new SseEventEncoder(new StringEncoder(), Arrays.asList(new JacksonJsonEncoder()));
+		assertTrue(encoder.canEncode(ResolvableType.forClass(Object.class), new MimeType("text", "event-stream")));
+	}
+
+	@Test
+	public void encodeServerSentEvent() {
+		SseEventEncoder encoder = new SseEventEncoder(new StringEncoder(), Arrays.asList(new JacksonJsonEncoder()));
+		SseEvent event = new SseEvent();
+		event.setId("c42");
+		event.setName("foo");
+		event.setComment("bla\nbla bla\nbla bla bla");
+		event.setReconnectTime(123L);
+		Mono<SseEvent> source = Mono.just(event);
+		Flux<DataBuffer> output = encoder.encode(source, this.dataBufferFactory,
+						ResolvableType.forClass(SseEvent.class), new MimeType("text", "event-stream"));
+		TestSubscriber
+				.subscribe(output)
+				.assertNoError()
+				.assertValuesWith(
+						stringConsumer(
+								"id:c42\n" +
+								"event:foo\n" +
+								"retry:123\n" +
+								":bla\n:bla bla\n:bla bla bla\n"),
+						stringConsumer("\n")
+				);
+	}
+
+	@Test
+	public void encodeString() {
+		SseEventEncoder encoder = new SseEventEncoder(new StringEncoder(), Arrays.asList(new JacksonJsonEncoder()));
+		Flux<String> source = Flux.just("foo", "bar");
+		Flux<DataBuffer> output = encoder.encode(source, this.dataBufferFactory,
+				ResolvableType.forClass(String.class), new MimeType("text", "event-stream"));
+		TestSubscriber
+				.subscribe(output)
+				.assertNoError()
+				.assertValuesWith(
+						stringConsumer("data:foo\n"),
+						stringConsumer("\n"),
+						stringConsumer("data:bar\n"),
+						stringConsumer("\n")
+				);
+	}
+
+	@Test
+	public void encodeMultilineString() {
+		SseEventEncoder encoder = new SseEventEncoder(new StringEncoder(), Arrays.asList(new JacksonJsonEncoder()));
+		Flux<String> source = Flux.just("foo\nbar", "foo\nbaz");
+		Flux<DataBuffer> output = encoder.encode(source, this.dataBufferFactory,
+				ResolvableType.forClass(String.class), new MimeType("text", "event-stream"));
+		TestSubscriber
+				.subscribe(output)
+				.assertNoError()
+				.assertValuesWith(
+						stringConsumer("data:foo\ndata:bar\n"),
+						stringConsumer("\n"),
+						stringConsumer("data:foo\ndata:baz\n"),
+						stringConsumer("\n")
+				);
+	}
+
+
+	@Test
+	public void encodePojo() {
+		SseEventEncoder encoder = new SseEventEncoder(new StringEncoder(), Arrays.asList(new JacksonJsonEncoder()));
+		Flux<Pojo> source = Flux.just(new Pojo("foofoo", "barbar"), new Pojo("foofoofoo", "barbarbar"));
+		Flux<DataBuffer> output = encoder.encode(source, this.dataBufferFactory,
+						ResolvableType.forClass(Pojo.class), new MimeType("text", "event-stream"));
+		TestSubscriber
+				.subscribe(output)
+				.assertNoError()
+				.assertValuesWith(
+						stringConsumer("data:{\"foo\":\"foofoo\",\"bar\":\"barbar\"}\n"),
+						stringConsumer("\n"),
+						stringConsumer("data:{\"foo\":\"foofoofoo\",\"bar\":\"barbarbar\"}\n"),
+						stringConsumer("\n")
+				);
+	}
+
+}

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

+/*
+ * Copyright 2002-2016 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.web.reactive.result.method.annotation;
+
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.List;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runners.Parameterized;
+import static org.springframework.web.client.reactive.HttpRequestBuilders.get;
+import static org.springframework.web.client.reactive.WebResponseExtractors.bodyStream;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.test.TestSubscriber;
+
+import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.codec.support.ByteBufferDecoder;
+import org.springframework.core.codec.support.JacksonJsonDecoder;
+import org.springframework.core.codec.support.JsonObjectDecoder;
+import org.springframework.core.codec.support.StringDecoder;
+import org.springframework.core.convert.ConversionService;
+import org.springframework.core.convert.support.GenericConversionService;
+import org.springframework.core.convert.support.ReactiveStreamsToCompletableFutureConverter;
+import org.springframework.core.convert.support.ReactiveStreamsToRxJava1Converter;
+import org.springframework.core.io.buffer.DataBufferFactory;
+import org.springframework.core.io.buffer.DefaultDataBufferFactory;
+import org.springframework.http.MediaType;
+import org.springframework.http.client.reactive.ReactorHttpClientRequestFactory;
+import org.springframework.http.converter.reactive.HttpMessageConverter;
+import org.springframework.http.converter.reactive.SseHttpMessageConverter;
+import org.springframework.http.server.reactive.AbstractHttpHandlerIntegrationTests;
+import org.springframework.http.server.reactive.HttpHandler;
+import org.springframework.http.server.reactive.boot.JettyHttpServer;
+import org.springframework.http.server.reactive.boot.ReactorHttpServer;
+import org.springframework.http.server.reactive.boot.RxNettyHttpServer;
+import org.springframework.http.server.reactive.boot.TomcatHttpServer;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.client.reactive.WebClient;
+import org.springframework.web.reactive.DispatcherHandler;
+import org.springframework.web.reactive.result.SimpleResultHandler;
+import org.springframework.web.reactive.sse.SseEvent;
+import org.springframework.web.server.adapter.WebHttpHandlerBuilder;
+
+/**
+ * @author Sebastien Deleuze
+ */
+public class SseIntegrationTests extends AbstractHttpHandlerIntegrationTests {
+
+	// TODO Fix Undertow support and remove this method
+	@Parameterized.Parameters(name = "server [{0}]")
+	public static Object[][] arguments() {
+		return new Object[][] {
+				{new JettyHttpServer()},
+				{new RxNettyHttpServer()},
+				{new ReactorHttpServer()},
+				{new TomcatHttpServer()},
+		};
+	}
+
+	private AnnotationConfigApplicationContext wac;
+
+	private WebClient webClient;
+
+	@Before
+	public void setup() throws Exception {
+		super.setup();
+		this.webClient = new WebClient(new ReactorHttpClientRequestFactory());
+		this.webClient.setMessageDecoders(Arrays.asList(
+				new ByteBufferDecoder(),
+				new StringDecoder(false),
+				new JacksonJsonDecoder(new JsonObjectDecoder())));
+	}
+
+	@Override
+	protected HttpHandler createHttpHandler() {
+		this.wac = new AnnotationConfigApplicationContext();
+		this.wac.register(TestConfiguration.class);
+		this.wac.refresh();
+
+		DispatcherHandler webHandler = new DispatcherHandler();
+		webHandler.setApplicationContext(this.wac);
+
+		return WebHttpHandlerBuilder.webHandler(webHandler).build();
+	}
+
+	@Test
+	public void sseAsString() throws Exception {
+		Mono<String> result = this.webClient
+				.perform(get("http://localhost:" + port + "/sse/string")
+				.accept(new MediaType("text", "event-stream")))
+				.extract(bodyStream(String.class))
+				.take(Duration.ofMillis(500))
+				.reduce((s1, s2) -> s1 + s2);
+
+		TestSubscriber
+				.subscribe(result)
+				.await()
+				.assertValues("data:foo 0\n\ndata:foo 1\n\n");
+	}
+
+	@Test
+	public void sseAsPojo() throws Exception {
+		Mono<String> result = this.webClient
+				.perform(get("http://localhost:" + port + "/sse/person")
+				.accept(new MediaType("text", "event-stream")))
+				.extract(bodyStream(String.class))
+				.take(Duration.ofMillis(500))
+				.reduce((s1, s2) -> s1 + s2);
+
+		TestSubscriber
+				.subscribe(result)
+				.await()
+				.assertValues("data:{\"name\":\"foo 0\"}\n\ndata:{\"name\":\"foo 1\"}\n\n");
+	}
+
+	@Test
+	public void sseAsEvent() throws Exception {
+		Mono<String> result = this.webClient
+				.perform(get("http://localhost:" + port + "/sse/event")
+				.accept(new MediaType("text", "event-stream")))
+				.extract(bodyStream(String.class))
+				.take(Duration.ofMillis(500))
+				.reduce((s1, s2) -> s1 + s2);
+
+		TestSubscriber
+				.subscribe(result)
+				.await()
+				.assertValues("id:0\n:bar\ndata:foo\n\nid:1\n:bar\ndata:foo\n\n");
+	}
+
+	@RestController
+	@SuppressWarnings("unused")
+	static class SseController {
+
+		@RequestMapping("/sse/string")
+		Flux<String> string() {
+			return Flux.interval(Duration.ofMillis(100)).map(l -> "foo " + l).take(2);
+		}
+
+		@RequestMapping("/sse/person")
+		Flux<Person> person() {
+			return Flux.interval(Duration.ofMillis(100)).map(l -> new Person("foo " + l)).take(2);
+		}
+
+		@RequestMapping("/sse/event")
+		Flux<SseEvent> sse() {
+			return Flux.interval(Duration.ofMillis(100)).map(l -> {
+				SseEvent event = new SseEvent();
+				event.setId(Long.toString(l));
+				event.setData("foo");
+				event.setComment("bar");
+				return event;
+			}).take(2);
+		}
+
+	}
+
+	@Configuration
+	@SuppressWarnings("unused")
+	static class TestConfiguration {
+
+		private DataBufferFactory dataBufferFactory = new DefaultDataBufferFactory();
+
+		@Bean
+		public SseController sseController() {
+			return new SseController();
+		}
+
+		@Bean
+		public RequestMappingHandlerMapping handlerMapping() {
+			return new RequestMappingHandlerMapping();
+		}
+
+		@Bean
+		public RequestMappingHandlerAdapter handlerAdapter() {
+			RequestMappingHandlerAdapter handlerAdapter = new RequestMappingHandlerAdapter();
+			handlerAdapter.setConversionService(conversionService());
+			return handlerAdapter;
+		}
+
+		@Bean
+		public ConversionService conversionService() {
+			GenericConversionService service = new GenericConversionService();
+			service.addConverter(new ReactiveStreamsToCompletableFutureConverter());
+			service.addConverter(new ReactiveStreamsToRxJava1Converter());
+			return service;
+		}
+
+		@Bean
+		public ResponseBodyResultHandler responseBodyResultHandler() {
+			List<HttpMessageConverter<?>> converters = Arrays.asList(new SseHttpMessageConverter());
+			return new ResponseBodyResultHandler(converters, conversionService());
+		}
+
+		@Bean
+		public SimpleResultHandler simpleHandlerResultHandler() {
+			return new SimpleResultHandler(conversionService());
+		}
+
+	}
+
+	private static class Person {
+
+		private String name;
+
+		@SuppressWarnings("unused")
+		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) {
+			if (this == o) {
+				return true;
+			}
+			if (o == null || getClass() != o.getClass()) {
+				return false;
+			}
+			Person person = (Person) o;
+			return !(this.name != null ? !this.name.equals(person.name) : person.name != null);
+		}
+
+		@Override
+		public int hashCode() {
+			return this.name != null ? this.name.hashCode() : 0;
+		}
+
+		@Override
+		public String toString() {
+			return "Person{" +
+					"name='" + name + '\'' +
+					'}';
+		}
+	}
+
+}