diff --git a/spring-webmvc/src/main/java/org/springframework/web/servlet/function/AsyncServerResponse.java b/spring-webmvc/src/main/java/org/springframework/web/servlet/function/AsyncServerResponse.java new file mode 100644 index 0000000000..cf34740522 --- /dev/null +++ b/spring-webmvc/src/main/java/org/springframework/web/servlet/function/AsyncServerResponse.java @@ -0,0 +1,348 @@ +/* + * 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. + * You may obtain a copy of the License at + * + * https://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.servlet.function; + +import java.io.IOException; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Function; + +import javax.servlet.AsyncContext; +import javax.servlet.AsyncListener; +import javax.servlet.ServletContext; +import javax.servlet.ServletException; +import javax.servlet.ServletRequest; +import javax.servlet.ServletResponse; +import javax.servlet.http.Cookie; +import javax.servlet.http.HttpServletRequest; +import javax.servlet.http.HttpServletRequestWrapper; +import javax.servlet.http.HttpServletResponse; + +import org.reactivestreams.Publisher; +import org.reactivestreams.Subscriber; +import org.reactivestreams.Subscription; + +import org.springframework.core.ReactiveAdapter; +import org.springframework.core.ReactiveAdapterRegistry; +import org.springframework.http.HttpHeaders; +import org.springframework.http.HttpStatus; +import org.springframework.lang.Nullable; +import org.springframework.util.Assert; +import org.springframework.util.ClassUtils; +import org.springframework.util.MultiValueMap; +import org.springframework.web.servlet.ModelAndView; + +/** + * Implementation of {@link ServerResponse} based on a {@link CompletableFuture}. + * + * @author Arjen Poutsma + * @since 5.3 + * @see ServerResponse#async(Object) + */ +final class AsyncServerResponse extends ErrorHandlingServerResponse { + + static final boolean reactiveStreamsPresent = ClassUtils.isPresent( + "org.reactivestreams.Publisher", AsyncServerResponse.class.getClassLoader()); + + + private final CompletableFuture futureResponse; + + + private AsyncServerResponse(CompletableFuture futureResponse) { + this.futureResponse = futureResponse; + } + + @Override + public HttpStatus statusCode() { + return delegate(ServerResponse::statusCode); + } + + @Override + public int rawStatusCode() { + return delegate(ServerResponse::rawStatusCode); + } + + @Override + public HttpHeaders headers() { + return delegate(ServerResponse::headers); + } + + @Override + public MultiValueMap cookies() { + return delegate(ServerResponse::cookies); + } + + private R delegate(Function function) { + ServerResponse response = this.futureResponse.getNow(null); + if (response != null) { + return function.apply(response); + } + else { + throw new IllegalStateException("Future ServerResponse has not yet completed"); + } + } + + @Nullable + @Override + public ModelAndView writeTo(HttpServletRequest request, HttpServletResponse response, Context context) { + + SharedAsyncContextHttpServletRequest sharedRequest = new SharedAsyncContextHttpServletRequest(request); + AsyncContext asyncContext = sharedRequest.startAsync(request, response); + this.futureResponse.whenComplete((futureResponse, futureThrowable) -> { + try { + if (futureResponse != null) { + ModelAndView mav = futureResponse.writeTo(sharedRequest, response, context); + Assert.state(mav == null, "Asynchronous, rendering ServerResponse implementations are not " + + "supported in WebMvc.fn. Please use WebFlux.fn instead."); + } + else if (futureThrowable != null) { + handleError(futureThrowable, request, response, context); + } + } + catch (Throwable throwable) { + try { + handleError(throwable, request, response, context); + } + catch (ServletException | IOException ex) { + logger.warn("Asynchronous execution resulted in exception", ex); + } + } + finally { + asyncContext.complete(); + } + }); + return null; + } + + + @SuppressWarnings({"unchecked"}) + public static ServerResponse create(Object o) { + Assert.notNull(o, "Argument to async must not be null"); + + if (o instanceof CompletableFuture) { + CompletableFuture futureResponse = (CompletableFuture) o; + return new AsyncServerResponse(futureResponse); + } + else if (reactiveStreamsPresent) { + ReactiveAdapter adapter = ReactiveAdapterRegistry.getSharedInstance().getAdapter(o.getClass()); + if (adapter != null) { + Publisher publisher = adapter.toPublisher(o); + CompletableFuture futureResponse = new CompletableFuture<>(); + publisher.subscribe(new ToFutureSubscriber(futureResponse)); + return new AsyncServerResponse(futureResponse); + } + } + throw new IllegalArgumentException("Asynchronous type not supported: " + o.getClass()); + } + + + /** + * Subscriber that exposes the first result it receives via a CompletableFuture. + */ + private static final class ToFutureSubscriber implements Subscriber { + + private final CompletableFuture future; + + @Nullable + private Subscription subscription; + + + public ToFutureSubscriber(CompletableFuture future) { + this.future = future; + } + + @Override + public void onSubscribe(Subscription s) { + if (this.subscription == null) { + this.subscription = s; + s.request(1); + } + else { + s.cancel(); + } + } + + @Override + public void onNext(ServerResponse serverResponse) { + if (!this.future.isDone()) { + this.future.complete(serverResponse); + } + } + + @Override + public void onError(Throwable t) { + if (!this.future.isDone()) { + this.future.completeExceptionally(t); + } + } + + @Override + public void onComplete() { + if (!this.future.isDone()) { + this.future.completeExceptionally(new IllegalStateException("Did not receive ServerResponse")); + } + } + } + + + /** + * HttpServletRequestWrapper that shares its AsyncContext between this + * AsyncServerResponse class and other, subsequent ServerResponse + * implementations, keeping track of how many contexts where + * started with startAsync(). This way, we make sure that + * {@link AsyncContext#complete()} only completes for the response that + * finishes last, and is not closed prematurely. + */ + private static final class SharedAsyncContextHttpServletRequest extends HttpServletRequestWrapper { + + private final AsyncContext asyncContext; + + private final AtomicInteger startedContexts; + + public SharedAsyncContextHttpServletRequest(HttpServletRequest request) { + super(request); + this.asyncContext = request.startAsync(); + this.startedContexts = new AtomicInteger(0); + } + + private SharedAsyncContextHttpServletRequest(HttpServletRequest request, AsyncContext asyncContext, + AtomicInteger startedContexts) { + super(request); + this.asyncContext = asyncContext; + this.startedContexts = startedContexts; + } + + @Override + public AsyncContext startAsync() throws IllegalStateException { + this.startedContexts.incrementAndGet(); + return new SharedAsyncContext(this.asyncContext, this, this.asyncContext.getResponse(), + this.startedContexts); + } + + @Override + public AsyncContext startAsync(ServletRequest servletRequest, ServletResponse servletResponse) + throws IllegalStateException { + this.startedContexts.incrementAndGet(); + SharedAsyncContextHttpServletRequest sharedRequest; + if (servletRequest instanceof SharedAsyncContextHttpServletRequest) { + sharedRequest = (SharedAsyncContextHttpServletRequest) servletRequest; + } + else { + sharedRequest = new SharedAsyncContextHttpServletRequest((HttpServletRequest) servletRequest, + this.asyncContext, this.startedContexts); + } + return new SharedAsyncContext(this.asyncContext, sharedRequest, servletResponse, this.startedContexts); + } + + @Override + public AsyncContext getAsyncContext() { + return new SharedAsyncContext(this.asyncContext, this, this.asyncContext.getResponse(), this.startedContexts); + } + + + private static final class SharedAsyncContext implements AsyncContext { + + private final AsyncContext delegate; + + private final AtomicInteger openContexts; + + private final ServletRequest request; + + private final ServletResponse response; + + + public SharedAsyncContext(AsyncContext delegate, SharedAsyncContextHttpServletRequest request, + ServletResponse response, AtomicInteger usageCount) { + + this.delegate = delegate; + this.request = request; + this.response = response; + this.openContexts = usageCount; + } + + @Override + public void complete() { + if (this.openContexts.decrementAndGet() == 0) { + this.delegate.complete(); + } + } + + @Override + public ServletRequest getRequest() { + return this.request; + } + + @Override + public ServletResponse getResponse() { + return this.response; + } + + @Override + public boolean hasOriginalRequestAndResponse() { + return this.delegate.hasOriginalRequestAndResponse(); + } + + @Override + public void dispatch() { + this.delegate.dispatch(); + } + + @Override + public void dispatch(String path) { + this.delegate.dispatch(path); + } + + @Override + public void dispatch(ServletContext context, String path) { + this.delegate.dispatch(context, path); + } + + @Override + public void start(Runnable run) { + this.delegate.start(run); + } + + @Override + public void addListener(AsyncListener listener) { + this.delegate.addListener(listener); + } + + @Override + public void addListener(AsyncListener listener, + ServletRequest servletRequest, + ServletResponse servletResponse) { + + this.delegate.addListener(listener, servletRequest, servletResponse); + } + + @Override + public T createListener(Class clazz) throws ServletException { + return this.delegate.createListener(clazz); + } + + @Override + public void setTimeout(long timeout) { + this.delegate.setTimeout(timeout); + } + + @Override + public long getTimeout() { + return this.delegate.getTimeout(); + } + } + } +} diff --git a/spring-webmvc/src/main/java/org/springframework/web/servlet/function/DefaultEntityResponseBuilder.java b/spring-webmvc/src/main/java/org/springframework/web/servlet/function/DefaultEntityResponseBuilder.java index 25e0003f52..868b2d6a7a 100644 --- a/spring-webmvc/src/main/java/org/springframework/web/servlet/function/DefaultEntityResponseBuilder.java +++ b/spring-webmvc/src/main/java/org/springframework/web/servlet/function/DefaultEntityResponseBuilder.java @@ -41,6 +41,8 @@ import org.reactivestreams.Subscriber; import org.reactivestreams.Subscription; import org.springframework.core.ParameterizedTypeReference; +import org.springframework.core.ReactiveAdapter; +import org.springframework.core.ReactiveAdapterRegistry; import org.springframework.core.io.InputStreamResource; import org.springframework.core.io.Resource; import org.springframework.core.io.support.ResourceRegion; @@ -56,7 +58,6 @@ import org.springframework.http.converter.HttpMessageConverter; import org.springframework.http.server.ServletServerHttpResponse; import org.springframework.lang.Nullable; import org.springframework.util.Assert; -import org.springframework.util.ClassUtils; import org.springframework.util.LinkedMultiValueMap; import org.springframework.util.MultiValueMap; import org.springframework.web.HttpMediaTypeNotAcceptableException; @@ -71,9 +72,6 @@ import org.springframework.web.servlet.ModelAndView; */ final class DefaultEntityResponseBuilder implements EntityResponse.Builder { - private static final boolean reactiveStreamsPresent = ClassUtils.isPresent( - "org.reactivestreams.Publisher", DefaultEntityResponseBuilder.class.getClassLoader()); - private static final Type RESOURCE_REGION_LIST_TYPE = new ParameterizedTypeReference>() { }.getType(); @@ -209,13 +207,14 @@ final class DefaultEntityResponseBuilder implements EntityResponse.Builder return new CompletionStageEntityResponse(this.status, this.headers, this.cookies, completionStage, this.entityType); } - else if (reactiveStreamsPresent && PublisherEntityResponse.isPublisher(this.entity)) { - Publisher publisher = (Publisher) this.entity; - return new PublisherEntityResponse(this.status, this.headers, this.cookies, publisher, this.entityType); - } - else { - return new DefaultEntityResponse<>(this.status, this.headers, this.cookies, this.entity, this.entityType); + else if (AsyncServerResponse.reactiveStreamsPresent) { + ReactiveAdapter adapter = ReactiveAdapterRegistry.getSharedInstance().getAdapter(this.entity.getClass()); + if (adapter != null) { + Publisher publisher = adapter.toPublisher(this.entity); + return new PublisherEntityResponse(this.status, this.headers, this.cookies, publisher, this.entityType); + } } + return new DefaultEntityResponse<>(this.status, this.headers, this.cookies, this.entity, this.entityType); } @@ -325,7 +324,7 @@ final class DefaultEntityResponseBuilder implements EntityResponse.Builder } protected void tryWriteEntityWithMessageConverters(Object entity, HttpServletRequest request, - HttpServletResponse response, ServerResponse.Context context) { + HttpServletResponse response, ServerResponse.Context context) throws ServletException, IOException { try { writeEntityWithMessageConverters(entity, request, response, context); } @@ -376,6 +375,9 @@ final class DefaultEntityResponseBuilder implements EntityResponse.Builder handleError(throwable, servletRequest, servletResponse, context); } } + catch (ServletException | IOException ex) { + logger.warn("Asynchronous execution resulted in exception", ex); + } finally { asyncContext.complete(); } @@ -385,6 +387,9 @@ final class DefaultEntityResponseBuilder implements EntityResponse.Builder } + /** + * {@link EntityResponse} implementation for asynchronous {@link Publisher} bodies. + */ private static class PublisherEntityResponse extends DefaultEntityResponse> { public PublisherEntityResponse(int statusCode, HttpHeaders headers, @@ -403,11 +408,6 @@ final class DefaultEntityResponseBuilder implements EntityResponse.Builder return null; } - public static boolean isPublisher(Object entity) { - return (entity instanceof Publisher); - } - - @SuppressWarnings("SubscriberImplementation") private class ProducingSubscriber implements Subscriber { @@ -438,13 +438,23 @@ final class DefaultEntityResponseBuilder implements EntityResponse.Builder public void onNext(T element) { HttpServletRequest servletRequest = (HttpServletRequest) this.asyncContext.getRequest(); HttpServletResponse servletResponse = (HttpServletResponse) this.asyncContext.getResponse(); - tryWriteEntityWithMessageConverters(element, servletRequest, servletResponse, this.context); + try { + tryWriteEntityWithMessageConverters(element, servletRequest, servletResponse, this.context); + } + catch (ServletException | IOException ex) { + onError(ex); + } } @Override public void onError(Throwable t) { - handleError(t, (HttpServletRequest) this.asyncContext.getRequest(), - (HttpServletResponse) this.asyncContext.getResponse(), this.context); + try { + handleError(t, (HttpServletRequest) this.asyncContext.getRequest(), + (HttpServletResponse) this.asyncContext.getResponse(), this.context); + } + catch (ServletException | IOException ex) { + logger.warn("Asynchronous execution resulted in exception", ex); + } this.asyncContext.complete(); } diff --git a/spring-webmvc/src/main/java/org/springframework/web/servlet/function/DefaultServerResponseBuilder.java b/spring-webmvc/src/main/java/org/springframework/web/servlet/function/DefaultServerResponseBuilder.java index aeabdc9128..dcba35f9e3 100644 --- a/spring-webmvc/src/main/java/org/springframework/web/servlet/function/DefaultServerResponseBuilder.java +++ b/spring-webmvc/src/main/java/org/springframework/web/servlet/function/DefaultServerResponseBuilder.java @@ -17,21 +17,17 @@ package org.springframework.web.servlet.function; import java.io.IOException; -import java.io.UncheckedIOException; import java.net.URI; import java.time.Instant; import java.time.ZonedDateTime; -import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; import java.util.EnumSet; import java.util.LinkedHashSet; -import java.util.List; import java.util.Map; import java.util.Set; import java.util.function.BiFunction; import java.util.function.Consumer; -import java.util.function.Predicate; import javax.servlet.ServletException; import javax.servlet.http.Cookie; @@ -231,17 +227,17 @@ class DefaultServerResponseBuilder implements ServerResponse.BodyBuilder { /** * Abstract base class for {@link ServerResponse} implementations. */ - abstract static class AbstractServerResponse implements ServerResponse { + abstract static class AbstractServerResponse extends ErrorHandlingServerResponse { private static final Set SAFE_METHODS = EnumSet.of(HttpMethod.GET, HttpMethod.HEAD); + final int statusCode; private final HttpHeaders headers; private final MultiValueMap cookies; - private final List> errorHandlers = new ArrayList<>(); protected AbstractServerResponse( int statusCode, HttpHeaders headers, MultiValueMap cookies) { @@ -252,14 +248,6 @@ class DefaultServerResponseBuilder implements ServerResponse.BodyBuilder { CollectionUtils.unmodifiableMultiValueMap(new LinkedMultiValueMap<>(cookies)); } - protected void addErrorHandler(Predicate predicate, - BiFunction errorHandler) { - - Assert.notNull(predicate, "Predicate must not be null"); - Assert.notNull(errorHandler, "ErrorHandler must not be null"); - this.errorHandlers.add(new ErrorHandler<>(predicate, errorHandler)); - } - @Override public final HttpStatus statusCode() { @@ -338,55 +326,6 @@ class DefaultServerResponseBuilder implements ServerResponse.BodyBuilder { HttpServletRequest request, HttpServletResponse response, Context context) throws ServletException, IOException; - @Nullable - protected ModelAndView handleError(Throwable t, HttpServletRequest servletRequest, - HttpServletResponse servletResponse, Context context) { - - return this.errorHandlers.stream() - .filter(errorHandler -> errorHandler.test(t)) - .findFirst() - .map(errorHandler -> { - ServerRequest serverRequest = (ServerRequest) - servletRequest.getAttribute(RouterFunctions.REQUEST_ATTRIBUTE); - ServerResponse serverResponse = errorHandler.handle(t, serverRequest); - try { - return serverResponse.writeTo(servletRequest, servletResponse, context); - } - catch (ServletException ex) { - throw new IllegalStateException(ex); - } - catch (IOException ex) { - throw new UncheckedIOException(ex); - } - }) - .orElseThrow(() -> new IllegalStateException(t)); - } - - - private static class ErrorHandler { - - private final Predicate predicate; - - private final BiFunction - responseProvider; - - public ErrorHandler(Predicate predicate, - BiFunction responseProvider) { - - Assert.notNull(predicate, "Predicate must not be null"); - Assert.notNull(responseProvider, "ResponseProvider must not be null"); - this.predicate = predicate; - this.responseProvider = responseProvider; - } - - public boolean test(Throwable t) { - return this.predicate.test(t); - } - - public T handle(Throwable t, ServerRequest serverRequest) { - return this.responseProvider.apply(t, serverRequest); - } - } } @@ -394,8 +333,7 @@ class DefaultServerResponseBuilder implements ServerResponse.BodyBuilder { private final BiFunction writeFunction; - public WriterFunctionResponse( - int statusCode, HttpHeaders headers, MultiValueMap cookies, + public WriterFunctionResponse(int statusCode, HttpHeaders headers, MultiValueMap cookies, BiFunction writeFunction) { super(statusCode, headers, cookies); diff --git a/spring-webmvc/src/main/java/org/springframework/web/servlet/function/EntityResponse.java b/spring-webmvc/src/main/java/org/springframework/web/servlet/function/EntityResponse.java index 9a5a170ac4..e2bb1c543b 100644 --- a/spring-webmvc/src/main/java/org/springframework/web/servlet/function/EntityResponse.java +++ b/spring-webmvc/src/main/java/org/springframework/web/servlet/function/EntityResponse.java @@ -52,7 +52,7 @@ public interface EntityResponse extends ServerResponse { /** * Create a builder with the given object. * @param t the object that represents the body of the response - * @param the type of element contained in the publisher + * @param the type of element contained in the entity * @return the created builder */ static Builder fromObject(T t) { @@ -63,7 +63,7 @@ public interface EntityResponse extends ServerResponse { * Create a builder with the given object and type reference. * @param t the object that represents the body of the response * @param entityType the type of the entity, used to capture the generic type - * @param the type of element contained in the publisher + * @param the type of element contained in the entity * @return the created builder */ static Builder fromObject(T t, ParameterizedTypeReference entityType) { diff --git a/spring-webmvc/src/main/java/org/springframework/web/servlet/function/ErrorHandlingServerResponse.java b/spring-webmvc/src/main/java/org/springframework/web/servlet/function/ErrorHandlingServerResponse.java new file mode 100644 index 0000000000..09785c5cf9 --- /dev/null +++ b/spring-webmvc/src/main/java/org/springframework/web/servlet/function/ErrorHandlingServerResponse.java @@ -0,0 +1,95 @@ +/* + * 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. + * You may obtain a copy of the License at + * + * https://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.servlet.function; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.function.BiFunction; +import java.util.function.Predicate; + +import javax.servlet.ServletException; +import javax.servlet.http.HttpServletRequest; +import javax.servlet.http.HttpServletResponse; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.lang.Nullable; +import org.springframework.util.Assert; +import org.springframework.web.servlet.ModelAndView; + +/** + * Base class for {@link ServerResponse} implementations with error handling. + * + * @author Arjen Poutsma + * @since 5.3 + */ +abstract class ErrorHandlingServerResponse implements ServerResponse { + + protected final Log logger = LogFactory.getLog(getClass()); + + private final List> errorHandlers = new ArrayList<>(); + + + protected final void addErrorHandler(Predicate predicate, + BiFunction errorHandler) { + + Assert.notNull(predicate, "Predicate must not be null"); + Assert.notNull(errorHandler, "ErrorHandler must not be null"); + this.errorHandlers.add(new ErrorHandler<>(predicate, errorHandler)); + } + + @Nullable + protected ModelAndView handleError(Throwable t, HttpServletRequest servletRequest, + HttpServletResponse servletResponse, Context context) throws ServletException, IOException { + + for (ErrorHandler errorHandler : this.errorHandlers) { + if (errorHandler.test(t)) { + ServerRequest serverRequest = (ServerRequest) + servletRequest.getAttribute(RouterFunctions.REQUEST_ATTRIBUTE); + ServerResponse serverResponse = errorHandler.handle(t, serverRequest); + return serverResponse.writeTo(servletRequest, servletResponse, context); + } + } + throw new ServletException(t); + } + + + private static class ErrorHandler { + + private final Predicate predicate; + + private final BiFunction responseProvider; + + public ErrorHandler(Predicate predicate, BiFunction responseProvider) { + Assert.notNull(predicate, "Predicate must not be null"); + Assert.notNull(responseProvider, "ResponseProvider must not be null"); + this.predicate = predicate; + this.responseProvider = responseProvider; + } + + public boolean test(Throwable t) { + return this.predicate.test(t); + } + + public T handle(Throwable t, ServerRequest serverRequest) { + return this.responseProvider.apply(t, serverRequest); + } + } + +} diff --git a/spring-webmvc/src/main/java/org/springframework/web/servlet/function/HandlerFilterFunction.java b/spring-webmvc/src/main/java/org/springframework/web/servlet/function/HandlerFilterFunction.java index 18e119011e..f27667592a 100644 --- a/spring-webmvc/src/main/java/org/springframework/web/servlet/function/HandlerFilterFunction.java +++ b/spring-webmvc/src/main/java/org/springframework/web/servlet/function/HandlerFilterFunction.java @@ -112,9 +112,8 @@ public interface HandlerFilterFunction { try { T t = next.handle(request); - if (t instanceof DefaultServerResponseBuilder.AbstractServerResponse) { - ((DefaultServerResponseBuilder.AbstractServerResponse) t) - .addErrorHandler(predicate, errorHandler); + if (t instanceof ErrorHandlingServerResponse) { + ((ErrorHandlingServerResponse) t).addErrorHandler(predicate, errorHandler); } return t; } diff --git a/spring-webmvc/src/main/java/org/springframework/web/servlet/function/ServerResponse.java b/spring-webmvc/src/main/java/org/springframework/web/servlet/function/ServerResponse.java index 2f0e6c1e4f..025317adb0 100644 --- a/spring-webmvc/src/main/java/org/springframework/web/servlet/function/ServerResponse.java +++ b/spring-webmvc/src/main/java/org/springframework/web/servlet/function/ServerResponse.java @@ -24,6 +24,7 @@ import java.util.Collection; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionStage; import java.util.function.BiFunction; import java.util.function.Consumer; @@ -36,6 +37,7 @@ import javax.servlet.http.HttpServletResponse; import org.reactivestreams.Publisher; import org.springframework.core.ParameterizedTypeReference; +import org.springframework.core.ReactiveAdapterRegistry; import org.springframework.http.CacheControl; import org.springframework.http.HttpHeaders; import org.springframework.http.HttpMethod; @@ -216,6 +218,31 @@ public interface ServerResponse { return status(HttpStatus.UNPROCESSABLE_ENTITY); } + /** + * Create a (built) response with the given asynchronous response. + * Parameter {@code asyncResponse} can be a + * {@link CompletableFuture CompletableFuture<ServerResponse>} or + * {@link Publisher Publisher<ServerResponse>} (or any + * asynchronous producer of a single {@code ServerResponse} that can be + * adapted via the {@link ReactiveAdapterRegistry}). + * + *

This method can be used to set the response status code, headers, and + * body based on an asynchronous result. If only the body is asynchronous, + * {@link BodyBuilder#body(Object)} can be used instead. + * + *

Note that + * {@linkplain RenderingResponse rendering responses}, as returned by + * {@link BodyBuilder#render}, are not supported as value + * for {@code asyncResponse}. Use WebFlux.fn for asynchronous rendering. + * @param asyncResponse a {@code CompletableFuture} or + * {@code Publisher} + * @return the asynchronous response + * @since 5.3 + */ + static ServerResponse async(Object asyncResponse) { + return AsyncServerResponse.create(asyncResponse); + } + /** * Defines a builder that adds headers to the response. @@ -374,10 +401,13 @@ public interface ServerResponse { BodyBuilder contentType(MediaType contentType); /** - * Set the body of the response to the given {@code Object} and return it. + * Set the body of the response to the given {@code Object} and return + * it. * - *

Asynchronous response bodies are supported by providing a {@link CompletionStage} or - * {@link Publisher} as body. + *

Asynchronous response bodies are supported by providing a + * {@link CompletionStage} or {@link Publisher} as body (or any + * asynchronous producer of a single entity that can be adapted via the + * {@link ReactiveAdapterRegistry}). * @param body the body of the response * @return the built response */