diff --git a/spring-web-reactive/src/main/java/org/springframework/web/reactive/function/AbstractHttpMessageWriterResponse.java b/spring-web-reactive/src/main/java/org/springframework/web/reactive/function/AbstractHttpMessageWriterResponse.java index 0a67bdd5f4..cd18b4bf63 100644 --- a/spring-web-reactive/src/main/java/org/springframework/web/reactive/function/AbstractHttpMessageWriterResponse.java +++ b/spring-web-reactive/src/main/java/org/springframework/web/reactive/function/AbstractHttpMessageWriterResponse.java @@ -16,6 +16,7 @@ package org.springframework.web.reactive.function; +import java.util.function.Supplier; import java.util.stream.Stream; import org.reactivestreams.Publisher; @@ -56,8 +57,9 @@ abstract class AbstractHttpMessageWriterResponse extends AbstractResponse } private Stream> messageWriterStream(ServerWebExchange exchange) { - return exchange.>>getAttribute(Router.HTTP_MESSAGE_WRITERS_ATTRIBUTE) - .orElseThrow(() -> new IllegalStateException("Could not find HttpMessageWriters in ServerWebExchange")); + return exchange.>>>getAttribute(Router.HTTP_MESSAGE_WRITERS_ATTRIBUTE) + .orElseThrow(() -> new IllegalStateException("Could not find HttpMessageWriters in ServerWebExchange")) + .get(); } } diff --git a/spring-web-reactive/src/main/java/org/springframework/web/reactive/function/DefaultRequest.java b/spring-web-reactive/src/main/java/org/springframework/web/reactive/function/DefaultRequest.java index 6ae47f7a50..f6c5788352 100644 --- a/spring-web-reactive/src/main/java/org/springframework/web/reactive/function/DefaultRequest.java +++ b/spring-web-reactive/src/main/java/org/springframework/web/reactive/function/DefaultRequest.java @@ -25,6 +25,7 @@ import java.util.Map; import java.util.Optional; import java.util.OptionalLong; import java.util.function.Function; +import java.util.function.Supplier; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -200,8 +201,9 @@ class DefaultRequest implements Request { } private Stream> messageReaderStream(ServerWebExchange exchange) { - return exchange.>>getAttribute(Router.HTTP_MESSAGE_READERS_ATTRIBUTE) - .orElseThrow(() -> new IllegalStateException("Could not find HttpMessageReaders in ServerWebExchange")); + return exchange.>>>getAttribute(Router.HTTP_MESSAGE_READERS_ATTRIBUTE) + .orElseThrow(() -> new IllegalStateException("Could not find HttpMessageReaders in ServerWebExchange")) + .get(); } @SuppressWarnings("unchecked") diff --git a/spring-web-reactive/src/main/java/org/springframework/web/reactive/function/DefaultResponseBuilder.java b/spring-web-reactive/src/main/java/org/springframework/web/reactive/function/DefaultResponseBuilder.java index c488b8d3f0..5f5b7d079e 100644 --- a/spring-web-reactive/src/main/java/org/springframework/web/reactive/function/DefaultResponseBuilder.java +++ b/spring-web-reactive/src/main/java/org/springframework/web/reactive/function/DefaultResponseBuilder.java @@ -137,7 +137,7 @@ class DefaultResponseBuilder implements Response.BodyBuilder { } @Override - public Response> stream(Publisher publisher, Class elementClass) { + public > Response stream(S publisher, Class elementClass) { Assert.notNull(publisher, "'publisher' must not be null"); Assert.notNull(elementClass, "'elementClass' must not be null"); return new PublisherResponse<>(this.statusCode, this.headers, publisher, elementClass); diff --git a/spring-web-reactive/src/main/java/org/springframework/web/reactive/function/PublisherResponse.java b/spring-web-reactive/src/main/java/org/springframework/web/reactive/function/PublisherResponse.java index 5324c77f4f..aadabb52b4 100644 --- a/spring-web-reactive/src/main/java/org/springframework/web/reactive/function/PublisherResponse.java +++ b/spring-web-reactive/src/main/java/org/springframework/web/reactive/function/PublisherResponse.java @@ -26,22 +26,22 @@ import org.springframework.web.server.ServerWebExchange; /** * @author Arjen Poutsma */ -class PublisherResponse extends AbstractHttpMessageWriterResponse> { +class PublisherResponse> extends AbstractHttpMessageWriterResponse { - private final Publisher body; + private final S body; private final ResolvableType bodyType; public PublisherResponse(int statusCode, HttpHeaders headers, - Publisher body, Class aClass) { + S body, Class aClass) { super(statusCode, headers); this.body = body; this.bodyType = ResolvableType.forClass(aClass); } @Override - public Publisher body() { + public S body() { return this.body; } diff --git a/spring-web-reactive/src/main/java/org/springframework/web/reactive/function/Response.java b/spring-web-reactive/src/main/java/org/springframework/web/reactive/function/Response.java index b03d8576ae..83381b7242 100644 --- a/spring-web-reactive/src/main/java/org/springframework/web/reactive/function/Response.java +++ b/spring-web-reactive/src/main/java/org/springframework/web/reactive/function/Response.java @@ -308,7 +308,7 @@ public interface Response { * @param the type of the elements contained in the publisher * @return the built response */ - Response> stream(Publisher publisher, Class elementClass); + > Response stream(S publisher, Class elementClass); /** * Set the body of the response to the given {@link Resource} and return it. diff --git a/spring-web-reactive/src/main/java/org/springframework/web/reactive/function/Router.java b/spring-web-reactive/src/main/java/org/springframework/web/reactive/function/Router.java index 17fe839d37..9912d8fee4 100644 --- a/spring-web-reactive/src/main/java/org/springframework/web/reactive/function/Router.java +++ b/spring-web-reactive/src/main/java/org/springframework/web/reactive/function/Router.java @@ -63,14 +63,14 @@ public abstract class Router { public static final String REQUEST_ATTRIBUTE = Router.class.getName() + ".request"; /** - * Name of the {@link ServerWebExchange} attribute that contains the + * Name of the {@link ServerWebExchange} attribute that contains a {@link Supplier} to the * {@linkplain Stream stream} of {@link HttpMessageReader}s obtained * from the {@linkplain Configuration#messageReaders() configuration}. */ public static final String HTTP_MESSAGE_READERS_ATTRIBUTE = Router.class.getName() + ".httpMessageReaders"; /** - * Name of the {@link ServerWebExchange} attribute that contains the + * Name of the {@link ServerWebExchange} attribute that contains a {@link Supplier} to the * {@linkplain Stream stream} of {@link HttpMessageWriter}s obtained * from the {@linkplain Configuration#messageWriters() configuration}. */ @@ -213,8 +213,8 @@ public abstract class Router { Configuration configuration) { Map attributes = exchange.getAttributes(); attributes.put(REQUEST_ATTRIBUTE, request); - attributes.put(HTTP_MESSAGE_READERS_ATTRIBUTE, configuration.messageReaders().get()); - attributes.put(HTTP_MESSAGE_WRITERS_ATTRIBUTE, configuration.messageWriters().get()); + attributes.put(HTTP_MESSAGE_READERS_ATTRIBUTE, configuration.messageReaders()); + attributes.put(HTTP_MESSAGE_WRITERS_ATTRIBUTE, configuration.messageWriters()); } @SuppressWarnings("unchecked") diff --git a/spring-web-reactive/src/test/java/org/springframework/web/reactive/function/BodyResponseTests.java b/spring-web-reactive/src/test/java/org/springframework/web/reactive/function/BodyResponseTests.java index affb00c75b..c305b0b788 100644 --- a/spring-web-reactive/src/test/java/org/springframework/web/reactive/function/BodyResponseTests.java +++ b/spring-web-reactive/src/test/java/org/springframework/web/reactive/function/BodyResponseTests.java @@ -17,6 +17,9 @@ package org.springframework.web.reactive.function; import java.util.Collections; +import java.util.Set; +import java.util.function.Supplier; +import java.util.stream.Stream; import org.junit.Test; @@ -24,6 +27,7 @@ import org.springframework.core.codec.CharSequenceEncoder; import org.springframework.http.HttpHeaders; import org.springframework.http.HttpMethod; import org.springframework.http.codec.EncoderHttpMessageWriter; +import org.springframework.http.codec.HttpMessageWriter; import org.springframework.mock.http.server.reactive.test.MockServerHttpRequest; import org.springframework.mock.http.server.reactive.test.MockServerHttpResponse; import org.springframework.web.server.ServerWebExchange; @@ -53,8 +57,9 @@ public class BodyResponseTests { MockServerHttpRequest request = new MockServerHttpRequest(HttpMethod.GET, "http://localhost"); MockServerHttpResponse response = new MockServerHttpResponse(); ServerWebExchange exchange = new DefaultServerWebExchange(request, response, new MockWebSessionManager()); - exchange.getAttributes().put(Router.HTTP_MESSAGE_WRITERS_ATTRIBUTE, Collections - .singleton(new EncoderHttpMessageWriter(new CharSequenceEncoder())).stream()); + Set> messageWriters = Collections.singleton(new EncoderHttpMessageWriter(new CharSequenceEncoder())); + exchange.getAttributes().put(Router.HTTP_MESSAGE_WRITERS_ATTRIBUTE, + (Supplier>>) messageWriters::stream); publisherResponse.writeTo(exchange).block(); diff --git a/spring-web-reactive/src/test/java/org/springframework/web/reactive/function/DefaultRequestTests.java b/spring-web-reactive/src/test/java/org/springframework/web/reactive/function/DefaultRequestTests.java index 394eb03049..c415fe9d84 100644 --- a/spring-web-reactive/src/test/java/org/springframework/web/reactive/function/DefaultRequestTests.java +++ b/spring-web-reactive/src/test/java/org/springframework/web/reactive/function/DefaultRequestTests.java @@ -26,6 +26,9 @@ import java.util.List; import java.util.Map; import java.util.Optional; import java.util.OptionalLong; +import java.util.Set; +import java.util.function.Supplier; +import java.util.stream.Stream; import org.junit.Before; import org.junit.Test; @@ -41,6 +44,7 @@ import org.springframework.http.HttpMethod; import org.springframework.http.HttpRange; import org.springframework.http.MediaType; import org.springframework.http.codec.DecoderHttpMessageReader; +import org.springframework.http.codec.HttpMessageReader; import org.springframework.http.server.reactive.ServerHttpRequest; import org.springframework.http.server.reactive.ServerHttpResponse; import org.springframework.util.LinkedMultiValueMap; @@ -155,14 +159,16 @@ public class DefaultRequestTests { when(mockRequest.getHeaders()).thenReturn(httpHeaders); when(mockRequest.getBody()).thenReturn(body); + Set> messageReaders = Collections + .singleton(new DecoderHttpMessageReader(new StringDecoder())); when(mockExchange.getAttribute(Router.HTTP_MESSAGE_READERS_ATTRIBUTE)) - .thenReturn(Optional.of(Collections - .singleton(new DecoderHttpMessageReader(new StringDecoder())) - .stream())); + .thenReturn(Optional.of( + (Supplier>>) messageReaders::stream)); assertEquals(body, defaultRequest.body().stream()); Mono resultMono = defaultRequest.body().convertToMono(String.class); assertEquals("foo", resultMono.block()); } + } \ No newline at end of file diff --git a/spring-web-reactive/src/test/java/org/springframework/web/reactive/function/PublisherHandlerFunctionIntegrationTests.java b/spring-web-reactive/src/test/java/org/springframework/web/reactive/function/PublisherHandlerFunctionIntegrationTests.java index 00f0d440de..de93778147 100644 --- a/spring-web-reactive/src/test/java/org/springframework/web/reactive/function/PublisherHandlerFunctionIntegrationTests.java +++ b/spring-web-reactive/src/test/java/org/springframework/web/reactive/function/PublisherHandlerFunctionIntegrationTests.java @@ -16,6 +16,7 @@ package org.springframework.web.reactive.function; +import java.net.URI; import java.util.List; import org.junit.Before; @@ -27,10 +28,13 @@ import reactor.core.publisher.Mono; import org.springframework.core.ParameterizedTypeReference; import org.springframework.http.HttpMethod; import org.springframework.http.HttpStatus; +import org.springframework.http.RequestEntity; import org.springframework.http.ResponseEntity; import org.springframework.web.client.RestTemplate; import static org.junit.Assert.assertEquals; +import static org.springframework.web.reactive.function.RequestPredicates.GET; +import static org.springframework.web.reactive.function.RequestPredicates.POST; import static org.springframework.web.reactive.function.Router.route; /** @@ -49,8 +53,9 @@ public class PublisherHandlerFunctionIntegrationTests @Override protected RoutingFunction routingFunction() { PersonHandler personHandler = new PersonHandler(); - return route(RequestPredicates.GET("/mono"), personHandler::mono) - .and(route(RequestPredicates.GET("/flux"), personHandler::flux)); + return route(GET("/mono"), personHandler::mono) + .and(route(POST("/mono"), personHandler::postMono)) + .and(route(GET("/flux"), personHandler::flux)); } @@ -76,6 +81,17 @@ public class PublisherHandlerFunctionIntegrationTests assertEquals("Jane", body.get(1).getName()); } + @Test + public void postMono() { + URI uri = URI.create("http://localhost:" + port + "/mono"); + Person person = new Person("Jack"); + RequestEntity requestEntity = RequestEntity.post(uri).body(person); + ResponseEntity result = restTemplate.exchange(requestEntity, Person.class); + + assertEquals(HttpStatus.OK, result.getStatusCode()); + assertEquals("Jack", result.getBody().getName()); + } + private static class PersonHandler { @@ -84,6 +100,11 @@ public class PublisherHandlerFunctionIntegrationTests return Response.ok().stream(Mono.just(person), Person.class); } + public Response> postMono(Request request) { + Mono personMono = request.body().convertToMono(Person.class); + return Response.ok().stream(personMono, Person.class); + } + public Response> flux(Request request) { Person person1 = new Person("John"); Person person2 = new Person("Jane"); diff --git a/spring-web-reactive/src/test/java/org/springframework/web/reactive/function/PublisherResponseTests.java b/spring-web-reactive/src/test/java/org/springframework/web/reactive/function/PublisherResponseTests.java index 2b032af427..65868a2ac1 100644 --- a/spring-web-reactive/src/test/java/org/springframework/web/reactive/function/PublisherResponseTests.java +++ b/spring-web-reactive/src/test/java/org/springframework/web/reactive/function/PublisherResponseTests.java @@ -17,6 +17,9 @@ package org.springframework.web.reactive.function; import java.util.Collections; +import java.util.Set; +import java.util.function.Supplier; +import java.util.stream.Stream; import org.junit.Test; import org.reactivestreams.Publisher; @@ -26,6 +29,7 @@ import org.springframework.core.codec.CharSequenceEncoder; import org.springframework.http.HttpHeaders; import org.springframework.http.HttpMethod; import org.springframework.http.codec.EncoderHttpMessageWriter; +import org.springframework.http.codec.HttpMessageWriter; import org.springframework.mock.http.server.reactive.test.MockServerHttpRequest; import org.springframework.mock.http.server.reactive.test.MockServerHttpResponse; import org.springframework.web.server.ServerWebExchange; @@ -42,7 +46,7 @@ public class PublisherResponseTests { private final Publisher publisher = Flux.just("foo", "bar"); - private final PublisherResponse publisherResponse = + private final PublisherResponse> publisherResponse = new PublisherResponse<>(200, new HttpHeaders(), publisher, String.class); @Test @@ -55,8 +59,9 @@ public class PublisherResponseTests { MockServerHttpRequest request = new MockServerHttpRequest(HttpMethod.GET, "http://localhost"); MockServerHttpResponse response = new MockServerHttpResponse(); ServerWebExchange exchange = new DefaultServerWebExchange(request, response, new MockWebSessionManager()); - exchange.getAttributes().put(Router.HTTP_MESSAGE_WRITERS_ATTRIBUTE, Collections - .singleton(new EncoderHttpMessageWriter(new CharSequenceEncoder())).stream()); + Set> messageWriters = Collections.singleton(new EncoderHttpMessageWriter(new CharSequenceEncoder())); + exchange.getAttributes().put(Router.HTTP_MESSAGE_WRITERS_ATTRIBUTE, + (Supplier>>) messageWriters::stream); publisherResponse.writeTo(exchange).block();