Polish + minor refactoring of SSE reader and writer
Instead of accepting List<Encoder|Decoder> and then look for the first to support JSON, always expect a single JSON [Encoder|Decoder] and use that unconditionally. When writing use the nested ResolvableType instead of the Class of the actual value which should better support generics. Remove the SSE hint and pass "text/event-stream" as the media type instead to serve as a hint. We are expecting a JSON encoder and using it unconditionally in any case so this should be good enough.
This commit is contained in:
Rossen Stoyanchev
@@ -29,7 +29,6 @@ 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.Decoder;
|
||||
import org.springframework.core.codec.StringDecoder;
|
||||
import org.springframework.core.io.buffer.DataBuffer;
|
||||
@@ -39,7 +38,6 @@ import org.springframework.core.io.buffer.DefaultDataBufferFactory;
|
||||
import org.springframework.http.MediaType;
|
||||
import org.springframework.http.ReactiveHttpInputMessage;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.MimeTypeUtils;
|
||||
|
||||
import static java.util.stream.Collectors.joining;
|
||||
|
||||
@@ -61,18 +59,22 @@ public class ServerSentEventHttpMessageReader implements HttpMessageReader<Objec
|
||||
private static final StringDecoder stringDecoder = new StringDecoder(false);
|
||||
|
||||
|
||||
private final List<Decoder<?>> dataDecoders;
|
||||
private final Decoder<?> decoder;
|
||||
|
||||
|
||||
public ServerSentEventHttpMessageReader() {
|
||||
this.dataDecoders = Collections.emptyList();
|
||||
/**
|
||||
* Constructor with JSON {@code Encoder} for encoding objects.
|
||||
*/
|
||||
public ServerSentEventHttpMessageReader(Decoder<?> decoder) {
|
||||
Assert.notNull(decoder, "Decoder must not be null");
|
||||
this.decoder = decoder;
|
||||
}
|
||||
|
||||
public ServerSentEventHttpMessageReader(List<Decoder<?>> dataDecoders) {
|
||||
Assert.notNull(dataDecoders, "'dataDecoders' must not be null");
|
||||
this.dataDecoders = new ArrayList<>(dataDecoders);
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<MediaType> getReadableMediaTypes() {
|
||||
return Collections.singletonList(MediaType.TEXT_EVENT_STREAM);
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean canRead(ResolvableType elementType, MediaType mediaType) {
|
||||
@@ -80,18 +82,13 @@ public class ServerSentEventHttpMessageReader implements HttpMessageReader<Objec
|
||||
ServerSentEvent.class.isAssignableFrom(elementType.getRawClass());
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<MediaType> getReadableMediaTypes() {
|
||||
return Collections.singletonList(MediaType.TEXT_EVENT_STREAM);
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public Flux<Object> read(ResolvableType elementType, ReactiveHttpInputMessage message,
|
||||
Map<String, Object> hints) {
|
||||
|
||||
boolean hasSseWrapper = ServerSentEvent.class.isAssignableFrom(elementType.getRawClass());
|
||||
ResolvableType dataType = (hasSseWrapper ? elementType.getGeneric(0) : elementType);
|
||||
boolean shouldWrap = ServerSentEvent.class.isAssignableFrom(elementType.getRawClass());
|
||||
ResolvableType valueType = shouldWrap ? elementType.getGeneric(0) : elementType;
|
||||
|
||||
return Flux.from(message.getBody())
|
||||
.concatMap(ServerSentEventHttpMessageReader::splitOnNewline)
|
||||
@@ -103,8 +100,8 @@ public class ServerSentEventHttpMessageReader implements HttpMessageReader<Objec
|
||||
.bufferUntil(line -> line.equals("\n"))
|
||||
.concatMap(rawLines -> {
|
||||
String[] lines = rawLines.stream().collect(joining()).split("\\r?\\n");
|
||||
ServerSentEvent<Object> event = buildEvent(lines, dataType, hints);
|
||||
return (hasSseWrapper ? Mono.just(event) : Mono.justOrEmpty(event.data()));
|
||||
ServerSentEvent<Object> event = buildEvent(lines, valueType, hints);
|
||||
return (shouldWrap ? Mono.just(event) : Mono.justOrEmpty(event.data()));
|
||||
})
|
||||
.cast(Object.class);
|
||||
}
|
||||
@@ -126,7 +123,8 @@ public class ServerSentEventHttpMessageReader implements HttpMessageReader<Objec
|
||||
return Flux.fromIterable(results);
|
||||
}
|
||||
|
||||
private ServerSentEvent<Object> buildEvent(String[] lines, ResolvableType dataType, Map<String, Object> hints) {
|
||||
private ServerSentEvent<Object> buildEvent(String[] lines, ResolvableType valueType,
|
||||
Map<String, Object> hints) {
|
||||
|
||||
ServerSentEvent.Builder<Object> sseBuilder = ServerSentEvent.builder();
|
||||
StringBuilder mutableData = new StringBuilder();
|
||||
@@ -151,7 +149,7 @@ public class ServerSentEventHttpMessageReader implements HttpMessageReader<Objec
|
||||
}
|
||||
if (mutableData.length() > 0) {
|
||||
String data = mutableData.toString();
|
||||
sseBuilder.data(decodeData(data, dataType, hints));
|
||||
sseBuilder.data(decodeData(data, valueType, hints));
|
||||
}
|
||||
if (mutableComment.length() > 0) {
|
||||
String comment = mutableComment.toString();
|
||||
@@ -169,11 +167,8 @@ public class ServerSentEventHttpMessageReader implements HttpMessageReader<Objec
|
||||
byte[] bytes = data.getBytes(StandardCharsets.UTF_8);
|
||||
Mono<DataBuffer> input = Mono.just(bufferFactory.wrap(bytes));
|
||||
|
||||
return this.dataDecoders.stream()
|
||||
.filter(e -> e.canDecode(dataType, MimeTypeUtils.APPLICATION_JSON))
|
||||
.findFirst()
|
||||
.orElseThrow(() -> new CodecException("No suitable decoder found!"))
|
||||
.decodeToMono(input, dataType, MimeTypeUtils.APPLICATION_JSON, hints)
|
||||
return this.decoder
|
||||
.decodeToMono(input, dataType, MediaType.TEXT_EVENT_STREAM, hints)
|
||||
.block(Duration.ZERO);
|
||||
}
|
||||
|
||||
@@ -181,7 +176,7 @@ public class ServerSentEventHttpMessageReader implements HttpMessageReader<Objec
|
||||
public Mono<Object> readMono(ResolvableType elementType, ReactiveHttpInputMessage message,
|
||||
Map<String, Object> hints) {
|
||||
|
||||
// Let's give StringDecoder a chance since SSE is ordered ahead of it
|
||||
// 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();
|
||||
|
||||
@@ -17,19 +17,16 @@
|
||||
package org.springframework.http.codec;
|
||||
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
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;
|
||||
@@ -38,39 +35,37 @@ import org.springframework.http.ReactiveHttpOutputMessage;
|
||||
import org.springframework.http.server.reactive.ServerHttpRequest;
|
||||
import org.springframework.http.server.reactive.ServerHttpResponse;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.MimeTypeUtils;
|
||||
|
||||
/**
|
||||
* Writer that supports a stream of {@link ServerSentEvent}s and also plain
|
||||
* {@link Object}s which is the same as an {@link ServerSentEvent} with data
|
||||
* only.
|
||||
* {@code ServerHttpMessageWriter} for {@code "text/event-stream"} responses.
|
||||
*
|
||||
* @author Sebastien Deleuze
|
||||
* @author Arjen Poutsma
|
||||
* @author Rossen Stoyanchev
|
||||
* @since 5.0
|
||||
*/
|
||||
public class ServerSentEventHttpMessageWriter implements ServerHttpMessageWriter<Object> {
|
||||
|
||||
private static final List<MediaType> WRITABLE_MEDIA_TYPES =
|
||||
Collections.singletonList(MediaType.TEXT_EVENT_STREAM);
|
||||
|
||||
|
||||
private final Encoder<?> encoder;
|
||||
|
||||
|
||||
/**
|
||||
* Server-Sent Events hint key expecting a {@link Boolean} value which when set to true
|
||||
* will adapt the content in order to comply with Server-Sent Events recommendation.
|
||||
* For example, it will append "data:" after each line break with data encoders
|
||||
* supporting it.
|
||||
* @see <a href="https://www.w3.org/TR/eventsource/">Server-Sent Events W3C recommendation</a>
|
||||
* Constructor with JSON {@code Encoder} for encoding objects. Support for
|
||||
* {@code String} event data is built-in.
|
||||
*/
|
||||
public static final String SSE_CONTENT_HINT = ServerSentEventHttpMessageWriter.class.getName() + ".sseContent";
|
||||
|
||||
|
||||
private final List<Encoder<?>> dataEncoders;
|
||||
|
||||
|
||||
public ServerSentEventHttpMessageWriter() {
|
||||
this.dataEncoders = Collections.emptyList();
|
||||
public ServerSentEventHttpMessageWriter(Encoder<?> encoder) {
|
||||
Assert.notNull(encoder, "'encoder' must not be null");
|
||||
this.encoder = encoder;
|
||||
}
|
||||
|
||||
public ServerSentEventHttpMessageWriter(List<Encoder<?>> dataEncoders) {
|
||||
Assert.notNull(dataEncoders, "'dataEncoders' must not be null");
|
||||
this.dataEncoders = new ArrayList<>(dataEncoders);
|
||||
|
||||
@Override
|
||||
public List<MediaType> getWritableMediaTypes() {
|
||||
return WRITABLE_MEDIA_TYPES;
|
||||
}
|
||||
|
||||
|
||||
@@ -81,61 +76,35 @@ public class ServerSentEventHttpMessageWriter implements ServerHttpMessageWriter
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<MediaType> getWritableMediaTypes() {
|
||||
return Collections.singletonList(MediaType.TEXT_EVENT_STREAM);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Mono<Void> write(Publisher<?> inputStream, ResolvableType elementType, MediaType mediaType,
|
||||
public Mono<Void> write(Publisher<?> input, ResolvableType elementType, MediaType mediaType,
|
||||
ReactiveHttpOutputMessage message, Map<String, Object> hints) {
|
||||
|
||||
message.getHeaders().setContentType(MediaType.TEXT_EVENT_STREAM);
|
||||
|
||||
DataBufferFactory bufferFactory = message.bufferFactory();
|
||||
Flux<Publisher<DataBuffer>> body = encode(inputStream, bufferFactory, elementType, hints);
|
||||
|
||||
return message.writeAndFlushWith(body);
|
||||
return message.writeAndFlushWith(encode(input, message.bufferFactory(), elementType, hints));
|
||||
}
|
||||
|
||||
private Flux<Publisher<DataBuffer>> encode(Publisher<?> inputStream, DataBufferFactory bufferFactory,
|
||||
ResolvableType type, Map<String, Object> hints) {
|
||||
private Flux<Publisher<DataBuffer>> encode(Publisher<?> input, DataBufferFactory factory,
|
||||
ResolvableType elementType, Map<String, Object> hints) {
|
||||
|
||||
Map<String, Object> hintsWithSse = new HashMap<>(hints);
|
||||
hintsWithSse.put(SSE_CONTENT_HINT, true);
|
||||
return Flux.from(inputStream)
|
||||
.map(o -> toSseEvent(o, type))
|
||||
.map(sse -> {
|
||||
StringBuilder sb = new StringBuilder();
|
||||
sse.id().ifPresent(id -> writeField("id", id, sb));
|
||||
sse.event().ifPresent(event -> writeField("event", event, sb));
|
||||
sse.retry().ifPresent(retry -> writeField("retry", retry.toMillis(), sb));
|
||||
sse.comment().ifPresent(comment -> {
|
||||
comment = comment.replaceAll("\\n", "\n:");
|
||||
sb.append(':').append(comment).append("\n");
|
||||
});
|
||||
Flux<DataBuffer> dataBuffer = sse.data()
|
||||
.<Flux<DataBuffer>>map(data -> {
|
||||
sb.append("data:");
|
||||
if (data instanceof String) {
|
||||
String stringData = ((String) data).replaceAll("\\n", "\ndata:");
|
||||
sb.append(stringData).append('\n');
|
||||
return Flux.empty();
|
||||
}
|
||||
else {
|
||||
return applyEncoder(data, bufferFactory, hintsWithSse);
|
||||
}
|
||||
}).orElse(Flux.empty());
|
||||
ResolvableType valueType = ServerSentEvent.class.isAssignableFrom(elementType.getRawClass()) ?
|
||||
elementType.getGeneric(0) : elementType;
|
||||
|
||||
return Flux.concat(encodeString(sb.toString(), bufferFactory), dataBuffer,
|
||||
encodeString("\n", bufferFactory));
|
||||
});
|
||||
return Flux.from(input).map(element -> {
|
||||
|
||||
}
|
||||
ServerSentEvent<?> sse = element instanceof ServerSentEvent ?
|
||||
(ServerSentEvent<?>) element : ServerSentEvent.builder().data(element).build();
|
||||
|
||||
private ServerSentEvent<?> toSseEvent(Object data, ResolvableType type) {
|
||||
return ServerSentEvent.class.isAssignableFrom(type.getRawClass())
|
||||
? (ServerSentEvent<?>) data
|
||||
: ServerSentEvent.builder().data(data).build();
|
||||
StringBuilder sb = new StringBuilder();
|
||||
sse.id().ifPresent(v -> writeField("id", v, sb));
|
||||
sse.event().ifPresent(v -> writeField("event", v, sb));
|
||||
sse.retry().ifPresent(v -> writeField("retry", v.toMillis(), sb));
|
||||
sse.comment().ifPresent(v -> sb.append(':').append(v.replaceAll("\\n", "\n:")).append("\n"));
|
||||
sse.data().ifPresent(v -> sb.append("data:"));
|
||||
|
||||
return Flux.concat(encodeText(sb, factory),
|
||||
encodeData(sse, valueType, factory, hints),
|
||||
encodeText("\n", factory));
|
||||
});
|
||||
}
|
||||
|
||||
private void writeField(String fieldName, Object fieldValue, StringBuilder stringBuilder) {
|
||||
@@ -146,40 +115,50 @@ public class ServerSentEventHttpMessageWriter implements ServerHttpMessageWriter
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private <T> Flux<DataBuffer> applyEncoder(Object data, DataBufferFactory bufferFactory, Map<String, Object> hints) {
|
||||
ResolvableType elementType = ResolvableType.forClass(data.getClass());
|
||||
Optional<Encoder<?>> encoder = dataEncoders
|
||||
.stream()
|
||||
.filter(e -> e.canEncode(elementType, MimeTypeUtils.APPLICATION_JSON))
|
||||
.findFirst();
|
||||
return ((Encoder<T>) encoder.orElseThrow(() -> new CodecException("No suitable encoder found!")))
|
||||
.encode(Mono.just((T) data), bufferFactory, elementType, MimeTypeUtils.APPLICATION_JSON, hints)
|
||||
.concatWith(encodeString("\n", bufferFactory));
|
||||
private <T> Flux<DataBuffer> encodeData(ServerSentEvent<?> event, ResolvableType valueType,
|
||||
DataBufferFactory factory, Map<String, Object> hints) {
|
||||
|
||||
Object data = event.data().orElse(null);
|
||||
if (data == null) {
|
||||
return Flux.empty();
|
||||
}
|
||||
|
||||
if (data instanceof String) {
|
||||
String text = (String) data;
|
||||
return Flux.from(encodeText(text.replaceAll("\\n", "\ndata:") + "\n", factory));
|
||||
}
|
||||
|
||||
return ((Encoder<T>) this.encoder)
|
||||
.encode(Mono.just((T) data), factory, valueType, MediaType.TEXT_EVENT_STREAM, hints)
|
||||
.concatWith(encodeText("\n", factory));
|
||||
}
|
||||
|
||||
private Mono<DataBuffer> encodeString(String str, DataBufferFactory bufferFactory) {
|
||||
byte[] bytes = str.getBytes(StandardCharsets.UTF_8);
|
||||
private Mono<DataBuffer> encodeText(CharSequence text, DataBufferFactory bufferFactory) {
|
||||
byte[] bytes = text.toString().getBytes(StandardCharsets.UTF_8);
|
||||
DataBuffer buffer = bufferFactory.allocateBuffer(bytes.length).write(bytes);
|
||||
return Mono.just(buffer);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Mono<Void> write(Publisher<?> inputStream, ResolvableType actualType, ResolvableType elementType,
|
||||
public Mono<Void> write(Publisher<?> input, ResolvableType actualType, ResolvableType elementType,
|
||||
MediaType mediaType, ServerHttpRequest request, ServerHttpResponse response,
|
||||
Map<String, Object> hints) {
|
||||
|
||||
Map<String, Object> allHints = this.dataEncoders.stream()
|
||||
.filter(encoder -> encoder instanceof ServerHttpEncoder)
|
||||
.map(encoder -> (ServerHttpEncoder<?>) encoder)
|
||||
.map(encoder -> encoder.getEncodeHints(actualType, elementType, mediaType, request, response))
|
||||
.reduce(new HashMap<>(), (t, u) -> {
|
||||
t.putAll(u);
|
||||
return t;
|
||||
});
|
||||
|
||||
Map<String, Object> allHints = new HashMap<>();
|
||||
allHints.putAll(getEncodeHints(actualType, elementType, mediaType, request, response));
|
||||
allHints.putAll(hints);
|
||||
|
||||
return write(inputStream, elementType, mediaType, response, allHints);
|
||||
return write(input, elementType, mediaType, response, allHints);
|
||||
}
|
||||
|
||||
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;
|
||||
return httpEncoder.getEncodeHints(actualType, elementType, mediaType, request, response);
|
||||
}
|
||||
return Collections.emptyMap();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -28,7 +28,6 @@ import com.fasterxml.jackson.core.util.DefaultPrettyPrinter;
|
||||
import com.fasterxml.jackson.databind.JavaType;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.fasterxml.jackson.databind.ObjectWriter;
|
||||
import com.fasterxml.jackson.databind.SerializationConfig;
|
||||
import com.fasterxml.jackson.databind.SerializationFeature;
|
||||
import com.fasterxml.jackson.databind.type.TypeFactory;
|
||||
import org.reactivestreams.Publisher;
|
||||
@@ -42,7 +41,6 @@ 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.ServerSentEventHttpMessageWriter;
|
||||
import org.springframework.http.converter.json.Jackson2ObjectMapperBuilder;
|
||||
import org.springframework.http.server.reactive.ServerHttpRequest;
|
||||
import org.springframework.http.server.reactive.ServerHttpResponse;
|
||||
@@ -97,20 +95,24 @@ public class Jackson2JsonEncoder extends Jackson2CodecSupport implements ServerH
|
||||
Assert.notNull(elementType, "'elementType' must not be null");
|
||||
|
||||
if (inputStream instanceof Mono) {
|
||||
return Flux.from(inputStream).map(value -> encodeValue(value, bufferFactory, elementType, hints));
|
||||
return Flux.from(inputStream).map(value ->
|
||||
encodeValue(value, mimeType, bufferFactory, elementType, hints));
|
||||
}
|
||||
else if (APPLICATION_STREAM_JSON.isCompatibleWith(mimeType)) {
|
||||
return Flux.from(inputStream).map(value -> {
|
||||
DataBuffer buffer = encodeValue(value, bufferFactory, elementType, hints);
|
||||
DataBuffer buffer = encodeValue(value, mimeType, bufferFactory, elementType, hints);
|
||||
buffer.write(new byte[]{'\n'});
|
||||
return buffer;
|
||||
});
|
||||
}
|
||||
ResolvableType listType = ResolvableType.forClassWithGenerics(List.class, elementType);
|
||||
return Flux.from(inputStream).collectList().map(list -> encodeValue(list, bufferFactory, listType, hints)).flux();
|
||||
else {
|
||||
ResolvableType listType = ResolvableType.forClassWithGenerics(List.class, elementType);
|
||||
return Flux.from(inputStream).collectList().map(list ->
|
||||
encodeValue(list, mimeType, bufferFactory, listType, hints)).flux();
|
||||
}
|
||||
}
|
||||
|
||||
private DataBuffer encodeValue(Object value, DataBufferFactory bufferFactory,
|
||||
private DataBuffer encodeValue(Object value, MimeType mimeType, DataBufferFactory bufferFactory,
|
||||
ResolvableType elementType, Map<String, Object> hints) {
|
||||
|
||||
TypeFactory typeFactory = this.mapper.getTypeFactory();
|
||||
@@ -126,9 +128,9 @@ public class Jackson2JsonEncoder extends Jackson2CodecSupport implements ServerH
|
||||
writer = writer.forType(javaType);
|
||||
}
|
||||
|
||||
SerializationConfig config = writer.getConfig();
|
||||
Boolean sseHint = (Boolean) hints.get(ServerSentEventHttpMessageWriter.SSE_CONTENT_HINT);
|
||||
if (Boolean.TRUE.equals(sseHint) && config.isEnabled(SerializationFeature.INDENT_OUTPUT)) {
|
||||
if (MediaType.TEXT_EVENT_STREAM.isCompatibleWith(mimeType) &&
|
||||
writer.getConfig().isEnabled(SerializationFeature.INDENT_OUTPUT)) {
|
||||
|
||||
writer = writer.with(this.ssePrettyPrinter);
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user