Spring/spring-framework
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Polishing in ResponseBodyEmitterReturnValueHandler

Browse Source 
See gh-33194
This commit is contained in:
rstoyanchev
2024-08-01 16:23:35 +03:00
parent f3ab3905b8
commit 184bb7c23c
4 changed files with 61 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -172,7 +172,7 @@ class ReactiveTypeHandler {
new TextEmitterSubscriber(emitter, this.taskExecutor).connect(adapter, returnValue);
return emitter;
}
MediaType streamingResponseType = findConcreteStreamingMediaType(mediaTypes);
MediaType streamingResponseType = findConcreteJsonStreamMediaType(mediaTypes);
if (streamingResponseType != null) {
ResponseBodyEmitter emitter = getEmitter(streamingResponseType);
new JsonEmitterSubscriber(emitter, this.taskExecutor).connect(adapter, returnValue);
@@ -203,7 +203,7 @@ class ReactiveTypeHandler {
*/
@SuppressWarnings("deprecation")
@Nullable
static MediaType findConcreteStreamingMediaType(Collection<MediaType> acceptedMediaTypes) {
static MediaType findConcreteJsonStreamMediaType(Collection<MediaType> acceptedMediaTypes) {
for (MediaType acceptedType : acceptedMediaTypes) {
if (WILDCARD_SUBTYPE_SUFFIXED_BY_NDJSON.includes(acceptedType)) {
if (acceptedType.isConcrete()) {

77
spring-webmvc/src/main/java/org/springframework/web/servlet/mvc/method/annotation/ResponseBodyEmitterReturnValueHandler.java
View File

@@ -49,12 +49,25 @@ import org.springframework.web.method.support.HandlerMethodReturnValueHandler;
import org.springframework.web.method.support.ModelAndViewContainer;
/**
* Handler for return values of type {@link ResponseBodyEmitter} and subclasses
* such as {@link SseEmitter} including the same types wrapped with
* {@link ResponseEntity}.
* Handler for return values of type:
* <ul>
* <li>{@link ResponseBodyEmitter} including sub-class {@link SseEmitter} and others.
* <li>Reactive return types known to {@link ReactiveAdapterRegistry}.
* <li>Any of the above wrapped with {@link ResponseEntity}.
* </ul>
*
* <p>As of 5.0 also supports reactive return value types for any reactive
* library with registered adapters in {@link ReactiveAdapterRegistry}.
* <p>Single-value reactive types are adapted to {@link DeferredResult}.
* Multi-value reactive types are adapted to {@link ResponseBodyEmitter} or
* {@link SseEmitter} as follows:
* <ul>
* <li>SSE stream, if the element type is
* {@link org.springframework.http.codec.ServerSentEvent} or if negotiated by
* content type.
* <li>Text stream for a {@link org.reactivestreams.Publisher} of
* {@link CharSequence}.
* <li>A JSON stream if negotiated by content type to
* {@link MediaType#APPLICATION_NDJSON}.
* </ul>
*
* @author Rossen Stoyanchev
* @since 4.2
@@ -153,7 +166,7 @@ public class ResponseBodyEmitterReturnValueHandler implements HandlerMethodRetur
else {
emitter = this.reactiveHandler.handleValue(returnValue, returnType, mavContainer, webRequest);
if (emitter == null) {
// Not streaming: write headers without committing response..
// We're not streaming; write headers without committing response
outputMessage.getHeaders().forEach((headerName, headerValues) -> {
for (String headerValue : headerValues) {
response.addHeader(headerName, headerValue);
@@ -164,18 +177,17 @@ public class ResponseBodyEmitterReturnValueHandler implements HandlerMethodRetur
}
emitter.extendResponse(outputMessage);
// At this point we know we're streaming..
// We are streaming
ShallowEtagHeaderFilter.disableContentCaching(request);
// Wrap the response to ignore further header changes
// Headers will be flushed at the first write
// Ignore further header changes; response is committed after first event
outputMessage = new StreamingServletServerHttpResponse(outputMessage);
HttpMessageConvertingHandler handler;
try {
DeferredResult<?> deferredResult = new DeferredResult<>(emitter.getTimeout());
WebAsyncUtils.getAsyncManager(webRequest).startDeferredResultProcessing(deferredResult, mavContainer);
handler = new HttpMessageConvertingHandler(outputMessage, deferredResult);
DeferredResult<?> result = new DeferredResult<>(emitter.getTimeout());
WebAsyncUtils.getAsyncManager(webRequest).startDeferredResultProcessing(result, mavContainer);
handler = new HttpMessageConvertingHandler(outputMessage, result);
}
catch (Throwable ex) {
emitter.initializeWithError(ex);
@@ -186,6 +198,26 @@ public class ResponseBodyEmitterReturnValueHandler implements HandlerMethodRetur
}
/**
* Wrap to silently ignore header changes HttpMessageConverter's that would
* otherwise cause HttpHeaders to raise exceptions.
*/
private static class StreamingServletServerHttpResponse extends DelegatingServerHttpResponse {
private final HttpHeaders mutableHeaders = new HttpHeaders();
public StreamingServletServerHttpResponse(ServerHttpResponse delegate) {
super(delegate);
this.mutableHeaders.putAll(delegate.getHeaders());
}
@Override
public HttpHeaders getHeaders() {
return this.mutableHeaders;
}
}
/**
* ResponseBodyEmitter.Handler that writes with HttpMessageConverter's.
*/
@@ -257,25 +289,4 @@ public class ResponseBodyEmitterReturnValueHandler implements HandlerMethodRetur
}
}
/**
* Wrap to silently ignore header changes HttpMessageConverter's that would
* otherwise cause HttpHeaders to raise exceptions.
*/
private static class StreamingServletServerHttpResponse extends DelegatingServerHttpResponse {
private final HttpHeaders mutableHeaders = new HttpHeaders();
public StreamingServletServerHttpResponse(ServerHttpResponse delegate) {
super(delegate);
this.mutableHeaders.putAll(delegate.getHeaders());
}
@Override
public HttpHeaders getHeaders() {
return this.mutableHeaders;
}
}
}