Introduce new functional web API

This commit introduces a new, functional web programming model in the
org.springframework.web.reactive.function package. The key types
are:

 - Request and Response are new Java 8-DSLs for access to the HTTP
   request and response
 - HandlerFunction represents a function to handle a request to a
   response
 - RoutingFunction maps a request to a HandlerFunction
 - FilterFunction filters a routing as defined by a RoutingFunction
 - RequestPredicate is used by Router to create RoutingFunctions
 - RequestPredicates offers common RequestPredicate instances
This commit is contained in:
Arjen Poutsma
2016-07-27 10:19:26 +02:00
parent 18e491ac0a
commit f1319f58ec
46 changed files with 5242 additions and 0 deletions

View File

@@ -0,0 +1,63 @@
/*
* 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.function;
import java.util.stream.Stream;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Mono;
import org.springframework.core.ResolvableType;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.codec.HttpMessageWriter;
import org.springframework.http.server.reactive.ServerHttpResponse;
import org.springframework.web.server.ServerWebExchange;
/**
* @author Arjen Poutsma
*/
abstract class AbstractHttpMessageWriterResponse<T> extends AbstractResponse<T> {
protected AbstractHttpMessageWriterResponse(int statusCode, HttpHeaders headers) {
super(statusCode, headers);
}
protected <S> Mono<Void> writeToInternal(ServerWebExchange exchange,
Publisher<S> body,
ResolvableType bodyType) {
writeStatusAndHeaders(exchange);
MediaType contentType = exchange.getResponse().getHeaders().getContentType();
ServerHttpResponse response = exchange.getResponse();
return messageWriterStream(exchange)
.filter(messageWriter -> messageWriter.canWrite(bodyType, contentType))
.findFirst()
.map(CastingUtils::cast)
.map(messageWriter -> messageWriter.write(body, bodyType, contentType, response))
.orElseGet(() -> {
response.setStatusCode(HttpStatus.NOT_ACCEPTABLE);
return response.setComplete();
});
}
private Stream<HttpMessageWriter<?>> messageWriterStream(ServerWebExchange exchange) {
return exchange.<Stream<HttpMessageWriter<?>>>getAttribute(Router.HTTP_MESSAGE_WRITERS_ATTRIBUTE)
.orElseThrow(() -> new IllegalStateException("Could not find HttpMessageWriters in ServerWebExchange"));
}
}

View File

@@ -0,0 +1,61 @@
/*
* 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.function;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.server.reactive.ServerHttpResponse;
import org.springframework.web.server.ServerWebExchange;
/**
* @author Arjen Poutsma
*/
abstract class AbstractResponse<T> implements Response<T> {
private final int statusCode;
private final HttpHeaders headers;
protected AbstractResponse(
int statusCode, HttpHeaders headers) {
this.statusCode = statusCode;
this.headers = HttpHeaders.readOnlyHttpHeaders(headers);
}
@Override
public HttpStatus statusCode() {
return HttpStatus.valueOf(this.statusCode);
}
@Override
public HttpHeaders headers() {
return this.headers;
}
protected void writeStatusAndHeaders(ServerWebExchange exchange) {
ServerHttpResponse response = exchange.getResponse();
response.setStatusCode(HttpStatus.valueOf(this.statusCode));
HttpHeaders responseHeaders = response.getHeaders();
if (!this.headers.isEmpty()) {
this.headers.entrySet().stream()
.filter(entry -> !responseHeaders.containsKey(entry.getKey()))
.forEach(entry -> responseHeaders
.put(entry.getKey(), entry.getValue()));
}
}
}

View File

@@ -0,0 +1,50 @@
/*
* 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.function;
import reactor.core.publisher.Mono;
import org.springframework.core.ResolvableType;
import org.springframework.http.HttpHeaders;
import org.springframework.util.Assert;
import org.springframework.web.server.ServerWebExchange;
/**
* @author Arjen Poutsma
*/
class BodyResponse<T> extends AbstractHttpMessageWriterResponse<T> {
private final T body;
public BodyResponse(int statusCode, HttpHeaders headers,
T body) {
super(statusCode, headers);
Assert.notNull(body, "'body' must not be null");
this.body = body;
}
@Override
public T body() {
return this.body;
}
@Override
public Mono<Void> writeTo(ServerWebExchange exchange) {
return writeToInternal(exchange, Mono.just(this.body), ResolvableType.forInstance(this.body));
}
}

View File

@@ -0,0 +1,41 @@
/*
* 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.function;
import org.springframework.http.codec.HttpMessageReader;
import org.springframework.http.codec.HttpMessageWriter;
/**
* @author Arjen Poutsma
*/
@SuppressWarnings("unchecked")
abstract class CastingUtils {
public static <T> HttpMessageReader<T> cast(HttpMessageReader<?> messageReader) {
return (HttpMessageReader<T>) messageReader;
}
public static <T> HttpMessageWriter<T> cast(HttpMessageWriter<?> messageWriter) {
return (HttpMessageWriter<T>) messageWriter;
}
public static <T> HandlerFunction<T> cast(HandlerFunction<?> handlerFunction) {
return (HandlerFunction<T>) handlerFunction;
}
}

View File

@@ -0,0 +1,81 @@
/*
* 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.function;
import java.util.ArrayList;
import java.util.List;
import java.util.function.Supplier;
import java.util.stream.Stream;
import org.springframework.core.codec.ByteBufferDecoder;
import org.springframework.core.codec.ByteBufferEncoder;
import org.springframework.core.codec.CharSequenceEncoder;
import org.springframework.core.codec.StringDecoder;
import org.springframework.http.codec.DecoderHttpMessageReader;
import org.springframework.http.codec.EncoderHttpMessageWriter;
import org.springframework.http.codec.HttpMessageReader;
import org.springframework.http.codec.HttpMessageWriter;
import org.springframework.http.codec.json.Jackson2JsonDecoder;
import org.springframework.http.codec.json.Jackson2JsonEncoder;
import org.springframework.http.codec.xml.Jaxb2XmlDecoder;
import org.springframework.http.codec.xml.Jaxb2XmlEncoder;
import org.springframework.util.ClassUtils;
/**
* A default implementation of configuration.
* @author Arjen Poutsma
*/
class DefaultConfiguration implements Router.Configuration {
private static final boolean jackson2Present =
ClassUtils.isPresent("com.fasterxml.jackson.databind.ObjectMapper",
DefaultConfiguration.class.getClassLoader()) &&
ClassUtils.isPresent("com.fasterxml.jackson.core.JsonGenerator",
DefaultConfiguration.class.getClassLoader());
private static final boolean jaxb2Present =
ClassUtils.isPresent("javax.xml.bind.Binder", DefaultConfiguration.class.getClassLoader());
private final List<HttpMessageReader<?>> messageReaders = new ArrayList<>();
private final List<HttpMessageWriter<?>> messageWriters = new ArrayList<>();
public DefaultConfiguration() {
this.messageReaders.add(new DecoderHttpMessageReader<>(new ByteBufferDecoder()));
this.messageReaders.add(new DecoderHttpMessageReader<>(new StringDecoder()));
this.messageWriters.add(new EncoderHttpMessageWriter<>(new ByteBufferEncoder()));
this.messageWriters.add(new EncoderHttpMessageWriter<>(new CharSequenceEncoder()));
if (jaxb2Present) {
this.messageReaders.add(new DecoderHttpMessageReader<>(new Jaxb2XmlDecoder()));
this.messageWriters.add(new EncoderHttpMessageWriter<>(new Jaxb2XmlEncoder()));
}
if (jackson2Present) {
this.messageReaders.add(new DecoderHttpMessageReader<>(new Jackson2JsonDecoder()));
this.messageWriters.add(new EncoderHttpMessageWriter<>(new Jackson2JsonEncoder()));
}
}
@Override
public Supplier<Stream<HttpMessageReader<?>>> messageReaders() {
return this.messageReaders::stream;
}
@Override
public Supplier<Stream<HttpMessageWriter<?>>> messageWriters() {
return this.messageWriters::stream;
}
}

View File

@@ -0,0 +1,215 @@
/*
* 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.function;
import java.net.InetSocketAddress;
import java.net.URI;
import java.nio.charset.Charset;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.OptionalLong;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.core.ResolvableType;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.HttpRange;
import org.springframework.http.MediaType;
import org.springframework.http.codec.HttpMessageReader;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.web.server.ServerWebExchange;
import org.springframework.web.server.UnsupportedMediaTypeStatusException;
/**
* {@code Request} implementation based on a {@link ServerWebExchange}.
* @author Arjen Poutsma
*/
class DefaultRequest implements Request {
private final ServerWebExchange exchange;
private final Headers headers;
private final Body body;
DefaultRequest(ServerWebExchange exchange) {
this.exchange = exchange;
this.headers = new DefaultHeaders();
this.body = new DefaultBody();
}
@Override
public HttpMethod method() {
return request().getMethod();
}
@Override
public URI uri() {
return request().getURI();
}
@Override
public Headers headers() {
return this.headers;
}
@Override
public Body body() {
return this.body;
}
@Override
public <T> Optional<T> attribute(String name) {
return this.exchange.getAttribute(name);
}
@Override
public Map<String, Object> attributes() {
return this.exchange.getAttributes();
}
@Override
public List<String> queryParams(String name) {
List<String> queryParams = request().getQueryParams().get(name);
return queryParams != null ? queryParams : Collections.emptyList();
}
@Override
public Map<String, String> pathVariables() {
return this.exchange.<Map<String, String>>getAttribute(Router.URI_TEMPLATE_VARIABLES_ATTRIBUTE).
orElseGet(Collections::emptyMap);
}
private ServerHttpRequest request() {
return this.exchange.getRequest();
}
private class DefaultHeaders implements Headers {
private HttpHeaders delegate() {
return request().getHeaders();
}
@Override
public List<MediaType> accept() {
return delegate().getAccept();
}
@Override
public List<Charset> acceptCharset() {
return delegate().getAcceptCharset();
}
@Override
public OptionalLong contentLength() {
return toOptionalLong(delegate().getContentLength());
}
@Override
public Optional<MediaType> contentType() {
return Optional.ofNullable(delegate().getContentType());
}
@Override
public InetSocketAddress host() {
return delegate().getHost();
}
@Override
public List<HttpRange> range() {
return delegate().getRange();
}
@Override
public List<String> header(String headerName) {
List<String> headerValues = delegate().get(headerName);
return headerValues != null ? headerValues : Collections.emptyList();
}
@Override
public HttpHeaders asHttpHeaders() {
return HttpHeaders.readOnlyHttpHeaders(delegate());
}
private OptionalLong toOptionalLong(long value) {
return value != -1 ? OptionalLong.of(value) : OptionalLong.empty();
}
}
private class DefaultBody implements Body {
@Override
public Flux<DataBuffer> stream() {
return request().getBody();
}
@Override
public <T> Flux<T> convertTo(Class<? extends T> aClass) {
ResolvableType elementType = ResolvableType.forClass(aClass);
return convertTo(aClass, reader -> reader.read(elementType, request()));
}
@Override
public <T> Mono<T> convertToMono(Class<? extends T> aClass) {
ResolvableType elementType = ResolvableType.forClass(aClass);
return convertTo(aClass, reader -> reader.readMono(elementType, request()));
}
private <T, S extends Publisher<T>> S convertTo(Class<? extends T> targetClass,
Function<HttpMessageReader<T>, S> readerFunction) {
ResolvableType elementType = ResolvableType.forClass(targetClass);
MediaType contentType = headers.contentType().orElse(MediaType.APPLICATION_OCTET_STREAM);
return messageReaderStream(exchange)
.filter(r -> r.canRead(elementType, contentType))
.findFirst()
.map(CastingUtils::<T>cast)
.map(readerFunction)
.orElseGet(() -> {
List<MediaType> supportedMediaTypes = messageReaderStream(exchange)
.flatMap(messageReader -> messageReader.getReadableMediaTypes().stream())
.collect(Collectors.toList());
return cast(
Mono.<T>error(new UnsupportedMediaTypeStatusException(contentType, supportedMediaTypes)));
});
}
private Stream<HttpMessageReader<?>> messageReaderStream(ServerWebExchange exchange) {
return exchange.<Stream<HttpMessageReader<?>>>getAttribute(Router.HTTP_MESSAGE_READERS_ATTRIBUTE)
.orElseThrow(() -> new IllegalStateException("Could not find HttpMessageReaders in ServerWebExchange"));
}
@SuppressWarnings("unchecked")
private <T, S extends Publisher<T>> S cast(Mono<T> mono) {
return (S) mono;
}
}
}

View File

@@ -0,0 +1,167 @@
/*
* 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.function;
import java.net.URI;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Arrays;
import java.util.LinkedHashSet;
import org.reactivestreams.Publisher;
import org.springframework.core.io.Resource;
import org.springframework.http.CacheControl;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.MediaType;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.util.Assert;
/**
* Default {@link Response.BodyBuilder} implementation.
*
* @author Arjen Poutsma
*/
class DefaultResponseBuilder implements Response.BodyBuilder {
private final int statusCode;
private final HttpHeaders headers = new HttpHeaders();
public DefaultResponseBuilder(int statusCode) {
this.statusCode = statusCode;
}
@Override
public Response.BodyBuilder header(String headerName, String... headerValues) {
for (String headerValue : headerValues) {
this.headers.add(headerName, headerValue);
}
return this;
}
@Override
public Response.BodyBuilder headers(HttpHeaders headers) {
if (headers != null) {
this.headers.putAll(headers);
}
return this;
}
@Override
public Response.BodyBuilder allow(HttpMethod... allowedMethods) {
this.headers.setAllow(new LinkedHashSet<>(Arrays.asList(allowedMethods)));
return this;
}
@Override
public Response.BodyBuilder contentLength(long contentLength) {
this.headers.setContentLength(contentLength);
return this;
}
@Override
public Response.BodyBuilder contentType(MediaType contentType) {
this.headers.setContentType(contentType);
return this;
}
@Override
public Response.BodyBuilder eTag(String eTag) {
if (eTag != null) {
if (!eTag.startsWith("\"") && !eTag.startsWith("W/\"")) {
eTag = "\"" + eTag;
}
if (!eTag.endsWith("\"")) {
eTag = eTag + "\"";
}
}
this.headers.setETag(eTag);
return this;
}
@Override
public Response.BodyBuilder lastModified(ZonedDateTime lastModified) {
ZonedDateTime gmt = lastModified.withZoneSameInstant(ZoneId.of("GMT"));
String headerValue = DateTimeFormatter.RFC_1123_DATE_TIME.format(gmt);
this.headers.set(HttpHeaders.LAST_MODIFIED, headerValue);
return this;
}
@Override
public Response.BodyBuilder location(URI location) {
this.headers.setLocation(location);
return this;
}
@Override
public Response.BodyBuilder cacheControl(CacheControl cacheControl) {
String ccValue = cacheControl.getHeaderValue();
if (ccValue != null) {
this.headers.setCacheControl(cacheControl.getHeaderValue());
}
return this;
}
@Override
public Response.BodyBuilder varyBy(String... requestHeaders) {
this.headers.setVary(Arrays.asList(requestHeaders));
return this;
}
@Override
public Response<Void> build() {
return new EmptyResponse(this.statusCode, this.headers);
}
@Override
public <T> Response<T> body(T body) {
Assert.notNull(body, "'body' must not be null");
return new BodyResponse<>(this.statusCode, this.headers, body);
}
@Override
public <T> Response<Publisher<T>> stream(Publisher<T> publisher, Class<T> 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);
}
@Override
public Response<Resource> resource(Resource resource) {
Assert.notNull(resource, "'resource' must not be null");
return new ResourceResponse(this.statusCode, this.headers, resource);
}
@Override
public <T> Response<Publisher<ServerSentEvent<T>>> sse(
Publisher<ServerSentEvent<T>> eventsPublisher) {
Assert.notNull(eventsPublisher, "'eventsPublisher' must not be null");
return ServerSentEventResponse
.fromSseEvents(this.statusCode, this.headers, eventsPublisher);
}
@Override
public <T> Response<Publisher<T>> sse(Publisher<T> eventsPublisher, Class<T> eventClass) {
Assert.notNull(eventsPublisher, "'eventsPublisher' must not be null");
Assert.notNull(eventClass, "'eventClass' must not be null");
return ServerSentEventResponse
.fromPublisher(this.statusCode, this.headers, eventsPublisher, eventClass);
}
}

View File

@@ -0,0 +1,51 @@
/*
* 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.function;
import reactor.core.publisher.Mono;
import org.springframework.http.HttpHeaders;
import org.springframework.web.server.ServerWebExchange;
/**
* @author Arjen Poutsma
*/
class EmptyResponse extends AbstractResponse<Void> {
public EmptyResponse(int statusCode, HttpHeaders headers) {
super(statusCode, addContentLength(headers));
}
private static HttpHeaders addContentLength(HttpHeaders headers) {
if (headers.getContentLength() == -1) {
headers.setContentLength(0);
}
return headers;
}
@Override
public Void body() {
return null;
}
@Override
public Mono<Void> writeTo(ServerWebExchange exchange) {
writeStatusAndHeaders(exchange);
return exchange.getResponse().setComplete();
}
}

View File

@@ -0,0 +1,45 @@
/*
* 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.function;
import org.springframework.web.reactive.function.support.RequestWrapper;
/**
* Represents a function that filters a {@linkplain HandlerFunction handler function}.
*
* @param <T> the type of the {@linkplain HandlerFunction handler function} to filter
* @param <R> the type of the response of the function
* @author Arjen Poutsma
* @since 5.0
*/
@FunctionalInterface
public interface FilterFunction<T, R> {
/**
* Apply this filter to the given handler function. The given
* {@linkplain HandlerFunction handler function} represents the next entity in the
* chain, and can be {@linkplain HandlerFunction#handle(Request) invoked} in order
* to proceed to this entity, or not invoked to block the chain.
*
* @param request the request
* @param next the next handler or filter function in the chain
* @return the filtered response
* @see RequestWrapper
*/
Response<R> filter(Request request, HandlerFunction<T> next);
}

View File

@@ -0,0 +1,36 @@
/*
* 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.function;
/**
* Represents a function that handles a {@linkplain Request request}.
*
* @param <T> the type of the response of the function
* @author Arjen Poutsma
* @since 5.0
*/
@FunctionalInterface
public interface HandlerFunction<T> {
/**
* Handle the given request.
* @param request the request to handle
* @return the response
*/
Response<T> handle(Request request);
}

View File

@@ -0,0 +1,53 @@
/*
* 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.function;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Mono;
import org.springframework.core.ResolvableType;
import org.springframework.http.HttpHeaders;
import org.springframework.web.server.ServerWebExchange;
/**
* @author Arjen Poutsma
*/
class PublisherResponse<T> extends AbstractHttpMessageWriterResponse<Publisher<T>> {
private final Publisher<T> body;
private final ResolvableType bodyType;
public PublisherResponse(int statusCode, HttpHeaders headers,
Publisher<T> body, Class<T> aClass) {
super(statusCode, headers);
this.body = body;
this.bodyType = ResolvableType.forClass(aClass);
}
@Override
public Publisher<T> body() {
return this.body;
}
@Override
public Mono<Void> writeTo(ServerWebExchange exchange) {
return writeToInternal(exchange, this.body, this.bodyType);
}
}

View File

@@ -0,0 +1,206 @@
/*
* 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.function;
import java.net.InetSocketAddress;
import java.net.URI;
import java.nio.charset.Charset;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.OptionalLong;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.HttpRange;
import org.springframework.http.MediaType;
/**
* Represents an HTTP request, as handled by a {@linkplain HandlerFunction handler function}.
* Access to headers and body is offered by {@link Headers} and {@link Body} respectively.
*
* @author Arjen Poutsma
* @since 5.0
*/
public interface Request {
/**
* Return the HTTP method.
*/
HttpMethod method();
/**
* Return the request URI.
*/
URI uri();
/**
* Return the request path.
*/
default String path() {
return uri().getPath();
}
/**
* Return the headers of this request.
*/
Headers headers();
/**
* Return the body of this request.
*/
Body body();
/**
* Return the request attribute value if present.
* @param name the attribute name
* @param <T> the attribute type
* @return the attribute value
*/
<T> Optional<T> attribute(String name);
/**
* Return a mutable map of attributes for this request.
*/
Map<String, Object> attributes();
/**
* Return the first query parameter with the given name, if present.
* @param name the parameter name
* @return the parameter value
*/
default Optional<String> queryParam(String name) {
List<String> queryParams = this.queryParams(name);
return !queryParams.isEmpty() ? Optional.of(queryParams.get(0)) : Optional.empty();
}
/**
* Return all query parameter with the given name. Returns an empty list if no values could
* be found.
* @param name the parameter name
* @return the parameter values
*/
List<String> queryParams(String name);
/**
* Return the path variable with the given name, if present.
* @param name the variable name
* @return the variable value
*/
default Optional<String> pathVariable(String name) {
return Optional.ofNullable(this.pathVariables().get(name));
}
/**
* Return all path variables.
* @return the path variables
*/
Map<String, String> pathVariables();
/**
* Represents the headers of the HTTP request.
* @see Request#headers()
*/
interface Headers {
/**
* Return the list of acceptable {@linkplain MediaType media types},
* as specified by the {@code Accept} header.
* <p>Returns an empty list when the acceptable media types are unspecified.
*/
List<MediaType> accept();
/**
* Return the list of acceptable {@linkplain Charset charsets},
* as specified by the {@code Accept-Charset} header.
*/
List<Charset> acceptCharset();
/**
* Return the length of the body in bytes, as specified by the
* {@code Content-Length} header.
*/
OptionalLong contentLength();
/**
* Return the {@linkplain MediaType media type} of the body, as specified
* by the {@code Content-Type} header.
*/
Optional<MediaType> contentType();
/**
* Return the value of the required {@code Host} header.
* <p>If the header value does not contain a port, the returned
* {@linkplain InetSocketAddress#getPort() port} will be {@code 0}.
*/
InetSocketAddress host();
/**
* Return the value of the {@code Range} header.
* <p>Returns an empty list when the range is unknown.
*/
List<HttpRange> range();
/**
* Return the header value(s), if any, for the header of the given name.
* <p>Return an empty list if no header values are found.
*
* @param headerName the header name
*/
List<String> header(String headerName);
/**
* Return the headers as a "live" {@link HttpHeaders} instance.
*/
HttpHeaders asHttpHeaders();
}
/**
* Represents the body of the HTTP request.
* @see Request#body()
*/
interface Body {
/**
* Return the request body as a stream of {@linkplain DataBuffer data buffers}.
* @return the request body byte stream
*/
Flux<DataBuffer> stream();
/**
* Converts the body into a multiple-element stream of the given type.
* @param aClass the type
* @param <T> the type of the element contained in the flux
* @return a flux that streams element of the given type
*/
<T> Flux<T> convertTo(Class<? extends T> aClass);
/**
* Converts the body into a single-element stream of the given type.
* @param aClass the type
* @param <T> the type of the element contained in the mono
* @return a flux that streams element of the given type
*/
<T> Mono<T> convertToMono(Class<? extends T> aClass);
}
}

View File

@@ -0,0 +1,91 @@
/*
* 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.function;
import org.springframework.util.Assert;
/**
* Represents a function that evaluates on a given {@link Request}.
* Instances of this function that evaluate on common request properties can be found in {@link RequestPredicates}.
*
* @author Arjen Poutsma
* @since 5.0
* @see RequestPredicates
* @see Router#route(RequestPredicate, HandlerFunction)
* @see Router#subroute(RequestPredicate, RoutingFunction)
*/
@FunctionalInterface
public interface RequestPredicate {
/**
* Evaluates this predicate on the given request.
*
*
* @param request the request to match against
* @return {@code true} if the request matches the predicate; {@code false} otherwise
*/
boolean test(Request request);
/**
* Returns a composed request predicate that tests against both this predicate AND the {@code other} predicate.
* When evaluating the composed predicate, if this predicate is {@code false}, then the {@code other}
* predicate is not evaluated.
*
* @param other a predicate that will be logically-ANDed with this predicate
* @return a predicate composed of this predicate AND the {@code other} predicate
*/
default RequestPredicate and(RequestPredicate other) {
Assert.notNull(other, "'other' must not be null");
return new RequestPredicate() {
@Override
public boolean test(Request t) {
return RequestPredicate.this.test(t) && other.test(t);
}
@Override
public Request subRequest(Request request) {
return other.subRequest(RequestPredicate.this.subRequest(request));
}
};
}
/**
* Return a predicate that represents the logical negation of this predicate.
*
* @return a predicate that represents the logical negation of this predicate
*/
default RequestPredicate negate() {
return (t) -> !test(t);
}
/**
* Returns a composed request predicate that tests against both this predicate OR the {@code other} predicate.
* When evaluating the composed predicate, if this predicate is {@code true}, then the {@code other} predicate
* is not evaluated.
* @param other a predicate that will be logically-ORed with this predicate
* @return a predicate composed of this predicate OR the {@code other} predicate
*/
default RequestPredicate or(RequestPredicate other) {
Assert.notNull(other, "'other' must not be null");
return (t) -> test(t) || other.test(t);
}
default Request subRequest(Request request) {
return request;
}
}

View File

@@ -0,0 +1,285 @@
/*
* 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.function;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.function.Predicate;
import org.springframework.http.HttpMethod;
import org.springframework.http.MediaType;
import org.springframework.util.AntPathMatcher;
import org.springframework.util.Assert;
import org.springframework.util.PathMatcher;
import org.springframework.web.reactive.function.support.RequestWrapper;
/**
* Implementations of {@link RequestPredicate} that implement various useful request matching operations, such as
* matching based on path, HTTP method, etc.
*
* @author Arjen Poutsma
* @since 5.0
*/
public abstract class RequestPredicates {
private static final PathMatcher DEFAULT_PATH_MATCHER = new AntPathMatcher();
/**
* Returns a {@code RequestPredicate} that always matches.
*
* @return a predicate that always matches
*/
public static RequestPredicate all() {
return request -> true;
}
/**
* Return a {@code RequestPredicate} that tests against the given HTTP method.
*
* @param httpMethod the HTTP method to match to
* @return a predicate that tests against the given HTTP method
*/
public static RequestPredicate method(HttpMethod httpMethod) {
return new HttpMethodPredicate(httpMethod);
}
/**
* Return a {@code RequestPredicate} that tests against the given path pattern.
*
* @param pattern the pattern to match to
* @return a predicate that tests against the given path pattern
*/
public static RequestPredicate path(String pattern) {
return path(pattern, DEFAULT_PATH_MATCHER);
}
/**
* Return a {@code RequestPredicate} that tests against the given path pattern using the given matcher.
*
* @param pattern the pattern to match to
* @param pathMatcher the path matcher to use
* @return a predicate that tests against the given path pattern
*/
public static RequestPredicate path(String pattern, PathMatcher pathMatcher) {
return new PathPredicate(pattern, pathMatcher);
}
/**
* Return a {@code RequestPredicate} that tests the request's headers against the given headers predicate.
*
* @param headersPredicate a predicate that tests against the request headers
* @return a predicate that tests against the given header predicate
*/
public static RequestPredicate headers(Predicate<Request.Headers> headersPredicate) {
return new HeaderPredicates(headersPredicate);
}
/**
* Return a {@code RequestPredicate} that tests if the request's
* {@linkplain Request.Headers#contentType() content type} is {@linkplain MediaType#includes(MediaType) included}
* by any of the given media types.
*
* @param mediaTypes the media types to match the request's content type against
* @return a predicate that tests the request's content type against the given media types
*/
public static RequestPredicate contentType(MediaType... mediaTypes) {
Assert.notEmpty(mediaTypes, "'mediaTypes' must not be empty");
Set<MediaType> mediaTypeSet = new HashSet<>(Arrays.asList(mediaTypes));
return headers(headers -> {
MediaType contentType = headers.contentType().orElse(MediaType.APPLICATION_OCTET_STREAM);
return mediaTypeSet.stream()
.anyMatch(mediaType -> mediaType.includes(contentType));
});
}
/**
* Return a {@code RequestPredicate} that tests if the request's
* {@linkplain Request.Headers#accept() accept} header is
* {@linkplain MediaType#isCompatibleWith(MediaType) compatible} with any of the given media types.
*
* @param mediaTypes the media types to match the request's accept header against
* @return a predicate that tests the request's accept header against the given media types
*/
public static RequestPredicate accept(MediaType... mediaTypes) {
Assert.notEmpty(mediaTypes, "'mediaTypes' must not be empty");
Set<MediaType> mediaTypeSet = new HashSet<>(Arrays.asList(mediaTypes));
return headers(headers -> {
List<MediaType> acceptedMediaTypes = headers.accept();
MediaType.sortBySpecificityAndQuality(acceptedMediaTypes);
return acceptedMediaTypes.stream()
.anyMatch(acceptedMediaType -> mediaTypeSet.stream()
.anyMatch(acceptedMediaType::isCompatibleWith));
});
}
/**
* Return a {@code RequestPredicate} that matches if request's HTTP method is {@code GET} and the given
* {@code pattern} matches against the request path.
*
* @param pattern the path pattern to match against
* @return a predicate that matches if the request method is GET and if the given pattern matches against the
* request path
*/
public static RequestPredicate GET(String pattern) {
return method(HttpMethod.GET).and(path(pattern));
}
/**
* Return a {@code RequestPredicate} that matches if request's HTTP method is {@code HEAD} and the given
* {@code pattern} matches against the request path.
*
* @param pattern the path pattern to match against
* @return a predicate that matches if the request method is HEAD and if the given pattern matches against the
* request path
*/
public static RequestPredicate HEAD(String pattern) {
return method(HttpMethod.HEAD).and(path(pattern));
}
/**
* Return a {@code RequestPredicate} that matches if request's HTTP method is {@code POST} and the given
* {@code pattern} matches against the request path.
*
* @param pattern the path pattern to match against
* @return a predicate that matches if the request method is POST and if the given pattern matches against the
* request path
*/
public static RequestPredicate POST(String pattern) {
return method(HttpMethod.POST).and(path(pattern));
}
/**
* Return a {@code RequestPredicate} that matches if request's HTTP method is {@code PUT} and the given
* {@code pattern} matches against the request path.
*
* @param pattern the path pattern to match against
* @return a predicate that matches if the request method is PUT and if the given pattern matches against the
* request path
*/
public static RequestPredicate PUT(String pattern) {
return method(HttpMethod.PUT).and(path(pattern));
}
/**
* Return a {@code RequestPredicate} that matches if request's HTTP method is {@code PATCH} and the given
* {@code pattern} matches against the request path.
*
* @param pattern the path pattern to match against
* @return a predicate that matches if the request method is PATCH and if the given pattern matches against the
* request path
*/
public static RequestPredicate PATCH(String pattern) {
return method(HttpMethod.PATCH).and(path(pattern));
}
/**
* Return a {@code RequestPredicate} that matches if request's HTTP method is {@code DELETE} and the given
* {@code pattern} matches against the request path.
*
* @param pattern the path pattern to match against
* @return a predicate that matches if the request method is DELETE and if the given pattern matches against the
* request path
*/
public static RequestPredicate DELETE(String pattern) {
return method(HttpMethod.DELETE).and(path(pattern));
}
/**
* Return a {@code RequestPredicate} that matches if request's HTTP method is {@code OPTIONS} and the given
* {@code pattern} matches against the request path.
*
* @param pattern the path pattern to match against
* @return a predicate that matches if the request method is OPTIONS and if the given pattern matches against the
* request path
*/
public static RequestPredicate OPTIONS(String pattern) {
return method(HttpMethod.OPTIONS).and(path(pattern));
}
private static class HttpMethodPredicate implements RequestPredicate {
private final HttpMethod httpMethod;
public HttpMethodPredicate(HttpMethod httpMethod) {
Assert.notNull(httpMethod, "'httpMethod' must not be null");
this.httpMethod = httpMethod;
}
@Override
public boolean test(Request request) {
return this.httpMethod == request.method();
}
}
private static class PathPredicate implements RequestPredicate {
private final String pattern;
private final PathMatcher pathMatcher;
public PathPredicate(String pattern, PathMatcher pathMatcher) {
Assert.notNull(pattern, "'pattern' must not be null");
Assert.notNull(pathMatcher, "'pathMatcher' must not be null");
this.pattern = pattern;
this.pathMatcher = pathMatcher;
}
@Override
public boolean test(Request request) {
String path = request.path();
if (this.pathMatcher.match(this.pattern, path)) {
Map<String, String> uriTemplateVariables = this.pathMatcher.extractUriTemplateVariables(this.pattern, path);
request.attributes().put(Router.URI_TEMPLATE_VARIABLES_ATTRIBUTE, uriTemplateVariables);
return true;
}
else {
return false;
}
}
@Override
public Request subRequest(Request request) {
String requestPath = request.path();
String subPath = this.pathMatcher.extractPathWithinPattern(this.pattern, requestPath);
return new RequestWrapper(request) {
@Override
public String path() {
return subPath;
}
};
}
}
private static class HeaderPredicates implements RequestPredicate {
private final Predicate<Request.Headers> headersPredicate;
public HeaderPredicates(Predicate<Request.Headers> headersPredicate) {
Assert.notNull(headersPredicate, "'headersPredicate' must not be null");
this.headersPredicate = headersPredicate;
}
@Override
public boolean test(Request request) {
return this.headersPredicate.test(request.headers());
}
}
}

View File

@@ -0,0 +1,55 @@
/*
* 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.function;
import reactor.core.publisher.Mono;
import org.springframework.core.ResolvableType;
import org.springframework.core.io.Resource;
import org.springframework.http.HttpHeaders;
import org.springframework.http.codec.ResourceHttpMessageWriter;
import org.springframework.web.server.ServerWebExchange;
/**
* @author Arjen Poutsma
*/
class ResourceResponse extends AbstractResponse<Resource> {
private static final ResolvableType RESOURCE_TYPE = ResolvableType.forClass(Resource.class);
private final ResourceHttpMessageWriter messageWriter = new ResourceHttpMessageWriter();
private final Resource resource;
public ResourceResponse(int statusCode, HttpHeaders httpHeaders, Resource resource) {
super(statusCode, httpHeaders);
this.resource = resource;
}
@Override
public Resource body() {
return this.resource;
}
@Override
public Mono<Void> writeTo(ServerWebExchange exchange) {
writeStatusAndHeaders(exchange);
return this.messageWriter
.write(Mono.just(this.resource), RESOURCE_TYPE, null, exchange.getResponse());
}
}

View File

@@ -0,0 +1,343 @@
/*
* 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.function;
import java.net.URI;
import java.time.ZonedDateTime;
import java.util.Set;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Mono;
import org.springframework.core.io.Resource;
import org.springframework.http.CacheControl;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.util.Assert;
import org.springframework.web.server.ServerWebExchange;
/**
* Represents a typed HTTP response, as returned by a {@linkplain HandlerFunction handler function} or
* {@linkplain FilterFunction filter function}.
*
* @author Arjen Poutsma
* @since 5.0
* @param <T> the type of the body that this response contains
*/
public interface Response<T> {
// Static builder methods
/**
* Create a builder with the status code and headers of the given response.
*
* @param other the response to copy the status and headers from
* @return the created builder
*/
static BodyBuilder from(Response<?> other) {
Assert.notNull(other, "'other' must not be null");
DefaultResponseBuilder builder = new DefaultResponseBuilder(other.statusCode().value());
return builder.headers(other.headers());
}
/**
* Create a builder with the given status.
*
* @param status the response status
* @return the created builder
*/
static BodyBuilder status(HttpStatus status) {
Assert.notNull(status, "HttpStatus must not be null");
return new DefaultResponseBuilder(status.value());
}
/**
* Create a builder with the given status.
*
* @param status the response status
* @return the created builder
*/
static BodyBuilder status(int status) {
return new DefaultResponseBuilder(status);
}
/**
* Create a builder with the status set to {@linkplain HttpStatus#OK OK}.
*
* @return the created builder
*/
static BodyBuilder ok() {
return status(HttpStatus.OK);
}
/**
* Create a new builder with a {@linkplain HttpStatus#CREATED CREATED} status
* and a location header set to the given URI.
*
* @param location the location URI
* @return the created builder
*/
static BodyBuilder created(URI location) {
BodyBuilder builder = status(HttpStatus.CREATED);
return builder.location(location);
}
/**
* Create a builder with an {@linkplain HttpStatus#ACCEPTED ACCEPTED} status.
*
* @return the created builder
*/
static BodyBuilder accepted() {
return status(HttpStatus.ACCEPTED);
}
/**
* Create a builder with a {@linkplain HttpStatus#NO_CONTENT NO_CONTENT} status.
*
* @return the created builder
*/
static HeadersBuilder<?> noContent() {
return status(HttpStatus.NO_CONTENT);
}
/**
* Create a builder with a {@linkplain HttpStatus#BAD_REQUEST BAD_REQUEST} status.
*
* @return the created builder
*/
static BodyBuilder badRequest() {
return status(HttpStatus.BAD_REQUEST);
}
/**
* Create a builder with a {@linkplain HttpStatus#NOT_FOUND NOT_FOUND} status.
*
* @return the created builder
*/
static HeadersBuilder<?> notFound() {
return status(HttpStatus.NOT_FOUND);
}
/**
* Create a builder with an
* {@linkplain HttpStatus#UNPROCESSABLE_ENTITY UNPROCESSABLE_ENTITY} status.
*
* @return the created builder
*/
static BodyBuilder unprocessableEntity() {
return status(HttpStatus.UNPROCESSABLE_ENTITY);
}
// Instance methods
/**
* Return the status code of this response.
*/
HttpStatus statusCode();
/**
* Return the headers of this response.
*/
HttpHeaders headers();
/**
* Return the body of this response.
*/
T body();
/**
* Writes this response to the given web exchange.
*
* @param exchange the web exchange to write to
* @return {@code Mono<Void>} to indicate when request handling is complete
*/
Mono<Void> writeTo(ServerWebExchange exchange);
/**
* Defines a builder that adds headers to the response entity.
*
* @param <B> the builder subclass
*/
interface HeadersBuilder<B extends HeadersBuilder<B>> {
/**
* Add the given header value(s) under the given name.
*
* @param headerName the header name
* @param headerValues the header value(s)
* @return this builder
* @see HttpHeaders#add(String, String)
*/
B header(String headerName, String... headerValues);
/**
* Copy the given headers into the entity's headers map.
*
* @param headers the existing HttpHeaders to copy from
* @return this builder
* @see HttpHeaders#add(String, String)
*/
B headers(HttpHeaders headers);
/**
* Set the set of allowed {@link HttpMethod HTTP methods}, as specified
* by the {@code Allow} header.
*
* @param allowedMethods the allowed methods
* @return this builder
* @see HttpHeaders#setAllow(Set)
*/
B allow(HttpMethod... allowedMethods);
/**
* Set the entity tag of the body, as specified by the {@code ETag} header.
*
* @param eTag the new entity tag
* @return this builder
* @see HttpHeaders#setETag(String)
*/
B eTag(String eTag);
/**
* Set the time the resource was last changed, as specified by the
* {@code Last-Modified} header.
* <p>The date should be specified as the number of milliseconds since
* January 1, 1970 GMT.
*
* @param lastModified the last modified date
* @return this builder
* @see HttpHeaders#setLastModified(long)
*/
B lastModified(ZonedDateTime lastModified);
/**
* Set the location of a resource, as specified by the {@code Location} header.
*
* @param location the location
* @return this builder
* @see HttpHeaders#setLocation(URI)
*/
B location(URI location);
/**
* Set the caching directives for the resource, as specified by the HTTP 1.1
* {@code Cache-Control} header.
* <p>A {@code CacheControl} instance can be built like
* {@code CacheControl.maxAge(3600).cachePublic().noTransform()}.
*
* @param cacheControl a builder for cache-related HTTP response headers
* @return this builder
* @see <a href="https://tools.ietf.org/html/rfc7234#section-5.2">RFC-7234 Section 5.2</a>
*/
B cacheControl(CacheControl cacheControl);
/**
* Configure one or more request header names (e.g. "Accept-Language") to
* add to the "Vary" response header to inform clients that the response is
* subject to content negotiation and variances based on the value of the
* given request headers. The configured request header names are added only
* if not already present in the response "Vary" header.
*
* @param requestHeaders request header names
* @return this builder
*/
B varyBy(String... requestHeaders);
/**
* Build the response entity with no body.
*
* @return the response entity
*/
Response<Void> build();
}
/**
* Defines a builder that adds a body to the response entity.
*/
interface BodyBuilder extends HeadersBuilder<BodyBuilder> {
/**
* Set the length of the body in bytes, as specified by the
* {@code Content-Length} header.
*
* @param contentLength the content length
* @return this builder
* @see HttpHeaders#setContentLength(long)
*/
BodyBuilder contentLength(long contentLength);
/**
* Set the {@linkplain MediaType media type} of the body, as specified by the
* {@code Content-Type} header.
*
* @param contentType the content type
* @return this builder
* @see HttpHeaders#setContentType(MediaType)
*/
BodyBuilder contentType(MediaType contentType);
/**
* Set the body of the response to the given object and return it.
*
* @param body the body of the response entity
* @return the built response
*/
<T> Response<T> body(T body);
/**
* Set the body of the response to the given {@link Publisher} and return it.
* @param publisher the publisher to stream to the response body
* @param elementClass the class of elements contained in the publisher
* @param <T> the type of the elements contained in the publisher
* @return the built response
*/
<T> Response<Publisher<T>> stream(Publisher<T> publisher, Class<T> elementClass);
/**
* Set the body of the response to the given {@link Resource} and return it.
* If the resource can be resolved to a {@linkplain Resource#getFile() file}, it will be copied using
* <a href="https://en.wikipedia.org/wiki/Zero-copy">zero-copy</a>
*
* @param resource the resource to write to the response
* @return the built response
*/
Response<Resource> resource(Resource resource);
/**
* Set the body of the response to the given {@link ServerSentEvent} publisher and return it.
* @param eventsPublisher the {@link ServerSentEvent} publisher to stream to the response body
* @param <T> the type of the elements contained in the {@link ServerSentEvent}
* @return the built response
* @see <a href="https://www.w3.org/TR/eventsource/">Server-Sent Events W3C recommendation</a>
*/
<T> Response<Publisher<ServerSentEvent<T>>> sse(Publisher<ServerSentEvent<T>> eventsPublisher);
/**
* Set the body of the response to the given Server-Sent Event {@link Publisher} and return it.
* @param eventsPublisher the publisher to stream to the response body as Server-Sent Events
* @param eventClass the class of event contained in the publisher
* @param <T> the type of the elements contained in the publisher
* @return the built response
* @see <a href="https://www.w3.org/TR/eventsource/">Server-Sent Events W3C recommendation</a>
*/
<T> Response<Publisher<T>> sse(Publisher<T> eventsPublisher, Class<T> eventClass);
}
}

View File

@@ -0,0 +1,278 @@
/*
* 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.function;
import java.util.Map;
import java.util.Optional;
import java.util.function.Supplier;
import java.util.stream.Stream;
import reactor.core.publisher.Mono;
import org.springframework.context.ApplicationContext;
import org.springframework.http.codec.HttpMessageReader;
import org.springframework.http.codec.HttpMessageWriter;
import org.springframework.http.server.reactive.HttpHandler;
import org.springframework.util.Assert;
import org.springframework.web.reactive.HandlerMapping;
import org.springframework.web.server.ServerWebExchange;
import org.springframework.web.server.adapter.HttpWebHandlerAdapter;
/**
* <strong>Central entry point Spring's functional web framework.</strong>
* Exposes routing functionality, such as to
* {@linkplain #route(RequestPredicate, HandlerFunction) create} a {@link RoutingFunction} given a
* {@link RequestPredicate} and {@link HandlerFunction}, and to do further
* {@linkplain #subroute(RequestPredicate, RoutingFunction) subrouting} on an existing routing
* function.
*
* <p>Additionally, this class can {@linkplain #toHttpHandler(RoutingFunction) transform} a
* {@link RoutingFunction} into an {@link HttpHandler}, which can be run in
* {@linkplain org.springframework.http.server.reactive.ServletHttpHandlerAdapter Servlet 3.1+},
* {@linkplain org.springframework.http.server.reactive.ReactorHttpHandlerAdapter Reactor},
* {@linkplain org.springframework.http.server.reactive.RxNettyHttpHandlerAdapter RxNetty}, or
* {@linkplain org.springframework.http.server.reactive.UndertowHttpHandlerAdapter Undertow}.
* Or it {@linkplain #toHandlerMapping(RoutingFunction, Configuration) transform} a
* {@link RoutingFunction} into an {@link HandlerMapping}, which can be run in a
* {@link org.springframework.web.reactive.DispatcherHandler DispatcherHandler}.
*
* @author Arjen Poutsma
* @since 5.0
*/
public abstract class Router {
private static final HandlerFunction<Void> NOT_FOUND_HANDLER = request -> Response.notFound().build();
/**
* Name of the {@link ServerWebExchange} attribute that contains the {@link Request}.
*/
public static final String REQUEST_ATTRIBUTE = Router.class.getName() + ".request";
/**
* Name of the {@link ServerWebExchange} attribute that contains 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
* {@linkplain Stream stream} of {@link HttpMessageWriter}s obtained
* from the {@linkplain Configuration#messageWriters() configuration}.
*/
public static final String HTTP_MESSAGE_WRITERS_ATTRIBUTE = Router.class.getName() + ".httpMessageWriters";
/**
* Name of the {@link ServerWebExchange} attribute that contains the URI
* templates map, mapping variable names to values.
*/
public static final String URI_TEMPLATE_VARIABLES_ATTRIBUTE = Router.class.getName() + ".uriTemplateVariables";
/**
* Route to the given handler function if the given request predicate applies.
*
* @param predicate the predicate to test
* @param handlerFunction the handler function to route to
* @param <T> the type of the handler function
* @return a routing function that routes to {@code handlerFunction} if {@code predicate} evaluates to {@code true}
* @see RequestPredicates
*/
public static <T> RoutingFunction<T> route(RequestPredicate predicate, HandlerFunction<T> handlerFunction) {
Assert.notNull(predicate, "'predicate' must not be null");
Assert.notNull(handlerFunction, "'handlerFunction' must not be null");
return request -> predicate.test(request) ? Optional.of(handlerFunction) : Optional.empty();
}
/**
* Route to the given routing function if the given request predicate applies.
*
* @param predicate the predicate to test
* @param routingFunction the routing function to route to
* @param <T> the type of the handler function
* @return a routing function that routes to {@code routingFunction} if {@code predicate} evaluates to {@code true}
* @see RequestPredicates
*/
public static <T> RoutingFunction<T> subroute(RequestPredicate predicate, RoutingFunction<T> routingFunction) {
Assert.notNull(predicate, "'predicate' must not be null");
Assert.notNull(routingFunction, "'routingFunction' must not be null");
return request -> {
if (predicate.test(request)) {
Request subRequest = predicate.subRequest(request);
return routingFunction.route(subRequest);
}
else {
return Optional.empty();
}
};
}
/**
* Converts the given {@linkplain RoutingFunction routing function} into a {@link HttpHandler}.
* This conversion uses the {@linkplain #defaultConfiguration() default configuration}.
*
* <p>The returned {@code HttpHandler} can be adapted to run in
* {@linkplain org.springframework.http.server.reactive.ServletHttpHandlerAdapter Servlet 3.1+},
* {@linkplain org.springframework.http.server.reactive.ReactorHttpHandlerAdapter Reactor},
* {@linkplain org.springframework.http.server.reactive.RxNettyHttpHandlerAdapter RxNetty}, or
* {@linkplain org.springframework.http.server.reactive.UndertowHttpHandlerAdapter Undertow}.
*
* @param routingFunction the routing function to convert
* @return an http handler that handles HTTP request using the given routing function
*/
public static HttpHandler toHttpHandler(RoutingFunction<?> routingFunction) {
return toHttpHandler(routingFunction, defaultConfiguration());
}
/**
* Converts the given {@linkplain RoutingFunction routing function} into a {@link HttpHandler},
* using the given configuration.
*
* <p>The returned {@code HttpHandler} can be adapted to run in
* {@linkplain org.springframework.http.server.reactive.ServletHttpHandlerAdapter Servlet 3.1+},
* {@linkplain org.springframework.http.server.reactive.ReactorHttpHandlerAdapter Reactor},
* {@linkplain org.springframework.http.server.reactive.RxNettyHttpHandlerAdapter RxNetty}, or
* {@linkplain org.springframework.http.server.reactive.UndertowHttpHandlerAdapter Undertow}.
*
* @param routingFunction the routing function to convert
* @param configuration the configuration to use
* @return an http handler that handles HTTP request using the given routing function
*/
public static HttpHandler toHttpHandler(RoutingFunction<?> routingFunction, Configuration configuration) {
Assert.notNull(routingFunction, "'routingFunction' must not be null");
Assert.notNull(configuration, "'configuration' must not be null");
return new HttpWebHandlerAdapter(exchange -> {
Request request = new DefaultRequest(exchange);
addAttributes(exchange, request, configuration);
HandlerFunction<?> handlerFunction = routingFunction.route(request).orElse(notFound());
Response<?> response = handlerFunction.handle(request);
return response.writeTo(exchange);
});
}
/**
* Converts the given {@linkplain RoutingFunction routing function} into a {@link HandlerMapping}.
* This conversion uses the {@linkplain #defaultConfiguration() default configuration}.
*
* <p>The returned {@code HttpHandler} can be run in a
* {@link org.springframework.web.reactive.DispatcherHandler}.
*
* @param routingFunction the routing function to convert
* @return an handler mapping that maps HTTP request to a handler using the given routing function
* @see org.springframework.web.reactive.function.support.HandlerFunctionAdapter
* @see org.springframework.web.reactive.function.support.ResponseResultHandler
*/
public static HandlerMapping toHandlerMapping(RoutingFunction<?> routingFunction) {
return toHandlerMapping(routingFunction, defaultConfiguration());
}
/**
* Converts the given {@linkplain RoutingFunction routing function} into a {@link HandlerMapping}
* using the given configuration.
*
* <p>The returned {@code HttpHandler} can be run in a
* {@link org.springframework.web.reactive.DispatcherHandler}.
*
* @param routingFunction the routing function to convert
* @return an handler mapping that maps HTTP request to a handler using the given routing function
* @see org.springframework.web.reactive.function.support.HandlerFunctionAdapter
* @see org.springframework.web.reactive.function.support.ResponseResultHandler
*/
public static HandlerMapping toHandlerMapping(RoutingFunction<?> routingFunction, Configuration configuration) {
Assert.notNull(routingFunction, "'routingFunction' must not be null");
Assert.notNull(configuration, "'configuration' must not be null");
return exchange -> {
Request request = new DefaultRequest(exchange);
addAttributes(exchange, request, configuration);
Optional<? extends HandlerFunction<?>> route = routingFunction.route(request);
return Mono.justOrEmpty(route);
};
}
private static void addAttributes(ServerWebExchange exchange, Request request,
Configuration configuration) {
Map<String, Object> 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());
}
@SuppressWarnings("unchecked")
private static <T> HandlerFunction<T> notFound() {
return (HandlerFunction<T>) NOT_FOUND_HANDLER;
}
/**
* Return the default configuration.
*/
public static Configuration defaultConfiguration() {
return new DefaultConfiguration();
}
/**
* Returns a configuration based on the given {@linkplain ApplicationContext application context}.
* This configuration will search for all {@link HttpMessageReader} and {@link HttpMessageWriter}
* instances in the given application context.
* @param applicationContext the application context to base the configuration on
* @return the configuration
*/
public static Configuration toConfiguration(ApplicationContext applicationContext) {
return new Configuration() {
@Override
public Supplier<Stream<HttpMessageReader<?>>> messageReaders() {
return () -> applicationContext.getBeansOfType(HttpMessageReader.class).values().stream()
.map(CastingUtils::cast);
}
@Override
public Supplier<Stream<HttpMessageWriter<?>>> messageWriters() {
return () -> applicationContext.getBeansOfType(HttpMessageWriter.class).values().stream()
.map(CastingUtils::cast);
}
};
}
/**
* Defines the configuration to be used by this {@code Router}.
*/
public interface Configuration {
/**
* Supply a {@linkplain Stream stream} of {@link HttpMessageReader}s to be used for request
* body conversion.
* @return the stream of message readers
*/
Supplier<Stream<HttpMessageReader<?>>> messageReaders();
/**
* Supply a {@linkplain Stream stream} of {@link HttpMessageWriter}s to be used for response
* body conversion.
* @return the stream of message writers
*/
Supplier<Stream<HttpMessageWriter<?>>> messageWriters();
}
}

View File

@@ -0,0 +1,86 @@
/*
* 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.function;
import java.util.Optional;
/**
* Represents a function that routes to a {@linkplain HandlerFunction handler function}.
*
* @param <T> the type of the {@linkplain HandlerFunction handler function} to route to
* @author Arjen Poutsma
* @since 5.0
*/
@FunctionalInterface
public interface RoutingFunction<T> {
/**
* Return the {@linkplain HandlerFunction handler function} that matches the given request.
* @param request the request to route to
* @return an {@code Optional} describing the {@code HandlerFunction} that matches this request,
* or an empty {@code Optional} if there is no match
*/
Optional<HandlerFunction<T>> route(Request request);
/**
* Return a composed routing function that first invokes this function,
* and then invokes the {@code other} function if this route had
* {@linkplain Optional#empty() no result}.
*
* @param other the function to apply when this function has no result
* @return a composed function that first routes with this function and then the {@code other} function if this
* function has no result
*/
default RoutingFunction<T> and(RoutingFunction<T> other) {
return request -> {
Optional<HandlerFunction<T>> result = this.route(request);
return result.isPresent() ? result : other.route(request);
};
}
/**
* Return a composed routing function that first invokes this function,
* and then invokes the {@code other} function if this route had
* {@linkplain Optional#empty() no result}.
*
* @param other the function to apply when this function has no result
* @return a composed function that first routes with this function and then the {@code other} function if this
* function has no result
*/
default <S> RoutingFunction<?> andOther(RoutingFunction<S> other) {
return request -> {
Optional<HandlerFunction<Object>> result = this.route(request).
map(CastingUtils::cast);
return result.isPresent() ? result : other.route(request)
.map(CastingUtils::cast);
};
}
/**
* Filter all {@linkplain HandlerFunction handler functions} routed by this function with the given
* {@linkplain FilterFunction filter function}.
*
* @param filterFunction the filter to apply
* @param <S> the filter return type
* @return the filtered routing function
*/
default <S> RoutingFunction<S> filter(FilterFunction<T, S> filterFunction) {
return request -> this.route(request)
.map(handlerFunction -> filterRequest -> filterFunction.filter(filterRequest, handlerFunction));
}
}

View File

@@ -0,0 +1,81 @@
/*
* 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.function;
import java.util.Collections;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Mono;
import org.springframework.core.ResolvableType;
import org.springframework.http.HttpHeaders;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.http.codec.ServerSentEventHttpMessageWriter;
import org.springframework.http.codec.json.Jackson2JsonEncoder;
import org.springframework.util.ClassUtils;
import org.springframework.web.server.ServerWebExchange;
/**
* @author Arjen Poutsma
*/
class ServerSentEventResponse<T> extends AbstractResponse<Publisher<T>> {
private static final boolean jackson2Present =
ClassUtils.isPresent("com.fasterxml.jackson.databind.ObjectMapper",
DefaultConfiguration.class.getClassLoader()) &&
ClassUtils.isPresent("com.fasterxml.jackson.core.JsonGenerator",
DefaultConfiguration.class.getClassLoader());
private static final ResolvableType SERVER_SIDE_EVENT_TYPE = ResolvableType.forClass(ServerSentEvent.class);
private final ServerSentEventHttpMessageWriter messageWriter;
private final Publisher<T> eventsPublisher;
private final ResolvableType eventType;
private ServerSentEventResponse(int statusCode, HttpHeaders headers, Publisher<T> eventsPublisher, ResolvableType eventType) {
super(statusCode, headers);
this.eventsPublisher = eventsPublisher;
this.eventType = eventType;
this.messageWriter =
jackson2Present ? new ServerSentEventHttpMessageWriter(Collections.singletonList(new Jackson2JsonEncoder())) :
new ServerSentEventHttpMessageWriter();
}
public static <S> ServerSentEventResponse<S> fromPublisher(int statusCode, HttpHeaders headers, Publisher<S> eventsPublisher, Class<? extends S> eventType) {
return new ServerSentEventResponse<>(statusCode, headers, eventsPublisher, ResolvableType.forClass(eventType));
}
public static <S> ServerSentEventResponse<ServerSentEvent<S>> fromSseEvents(int statusCode, HttpHeaders headers, Publisher<ServerSentEvent<S>> eventsPublisher) {
return new ServerSentEventResponse<>(statusCode, headers, eventsPublisher,
SERVER_SIDE_EVENT_TYPE);
}
@Override
public Publisher<T> body() {
return this.eventsPublisher;
}
@Override
public Mono<Void> writeTo(ServerWebExchange exchange) {
writeStatusAndHeaders(exchange);
return this.messageWriter.write(this.eventsPublisher, this.eventType, null, exchange.getResponse());
}
}

View File

@@ -0,0 +1,4 @@
/**
* Types that make up Spring's functional web framework.
*/
package org.springframework.web.reactive.function;

View File

@@ -0,0 +1,69 @@
/*
* 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.function.support;
import java.lang.reflect.Method;
import reactor.core.publisher.Mono;
import org.springframework.core.MethodParameter;
import org.springframework.web.reactive.HandlerAdapter;
import org.springframework.web.reactive.HandlerResult;
import org.springframework.web.reactive.function.HandlerFunction;
import org.springframework.web.reactive.function.Request;
import org.springframework.web.reactive.function.Response;
import org.springframework.web.reactive.function.Router;
import org.springframework.web.server.ServerWebExchange;
/**
* {@code HandlerAdapter} implementation that supports {@link HandlerFunction}s.
*
* @author Arjen Poutsma
* @since 5.0
*/
public class HandlerFunctionAdapter implements HandlerAdapter {
private static final MethodParameter HANDLER_FUNCTION_RETURN_TYPE;
static {
try {
Method method = HandlerFunction.class.getMethod("handle", Request.class);
HANDLER_FUNCTION_RETURN_TYPE = new MethodParameter(method, -1);
}
catch (NoSuchMethodException ex) {
throw new Error(ex);
}
}
@Override
public boolean supports(Object handler) {
return handler instanceof HandlerFunction;
}
@Override
public Mono<HandlerResult> handle(ServerWebExchange exchange, Object handler) {
HandlerFunction<?> handlerFunction = (HandlerFunction<?>) handler;
Request request =
exchange.<Request>getAttribute(Router.REQUEST_ATTRIBUTE)
.orElseThrow(() -> new IllegalStateException("Could not find Request in exchange attributes"));
Response<?> response = handlerFunction.handle(request);
HandlerResult handlerResult =
new HandlerResult(handlerFunction, response, HANDLER_FUNCTION_RETURN_TYPE);
return Mono.just(handlerResult);
}
}

View File

@@ -0,0 +1,216 @@
/*
* 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.function.support;
import java.net.InetSocketAddress;
import java.net.URI;
import java.nio.charset.Charset;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.OptionalLong;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.HttpRange;
import org.springframework.http.MediaType;
import org.springframework.util.Assert;
import org.springframework.web.reactive.function.HandlerFunction;
import org.springframework.web.reactive.function.Request;
/**
* Implementation of the {@link Request} interface that can be subclassed to adapt the request to a
* {@link HandlerFunction handler function}. All methods default to calling through to the wrapped request.
*
* @author Arjen Poutsma
* @since 5.0
*/
public class RequestWrapper implements Request {
private final Request request;
/**
* Create a new {@code RequestWrapper} that wraps the given request.
*
* @param request the request to wrap
*/
public RequestWrapper(Request request) {
Assert.notNull(request, "'request' must not be null");
this.request = request;
}
/**
* Return the wrapped request.
*/
public Request request() {
return this.request;
}
@Override
public HttpMethod method() {
return this.request.method();
}
@Override
public URI uri() {
return this.request.uri();
}
@Override
public String path() {
return this.request.path();
}
@Override
public Headers headers() {
return this.request.headers();
}
@Override
public Body body() {
return this.request.body();
}
@Override
public <T> Optional<T> attribute(String name) {
return this.request.attribute(name);
}
@Override
public Map<String, Object> attributes() {
return this.request.attributes();
}
@Override
public Optional<String> queryParam(String name) {
return this.request.queryParam(name);
}
@Override
public List<String> queryParams(String name) {
return this.request.queryParams(name);
}
@Override
public Optional<String> pathVariable(String name) {
return this.request.pathVariable(name);
}
@Override
public Map<String, String> pathVariables() {
return this.request.pathVariables();
}
/**
* Implementation of the {@link Headers} interface that can be subclassed to adapt the headers to a
* {@link HandlerFunction handler function}. All methods default to calling through to the wrapped headers.
*/
public static class HeadersWrapper implements Request.Headers {
private final Headers headers;
/**
* Create a new {@code HeadersWrapper} that wraps the given request.
*
* @param headers the headers to wrap
*/
public HeadersWrapper(Headers headers) {
Assert.notNull(headers, "'headers' must not be null");
this.headers = headers;
}
@Override
public List<MediaType> accept() {
return this.headers.accept();
}
@Override
public List<Charset> acceptCharset() {
return this.headers.acceptCharset();
}
@Override
public OptionalLong contentLength() {
return this.headers.contentLength();
}
@Override
public Optional<MediaType> contentType() {
return this.headers.contentType();
}
@Override
public InetSocketAddress host() {
return this.headers.host();
}
@Override
public List<HttpRange> range() {
return this.headers.range();
}
@Override
public List<String> header(String headerName) {
return this.headers.header(headerName);
}
@Override
public HttpHeaders asHttpHeaders() {
return this.headers.asHttpHeaders();
}
}
/**
* Implementation of the {@link Body} interface that can be subclassed to adapt the headers to a
* {@link HandlerFunction handler function}. All methods default to calling through to the wrapped body.
*/
public static class BodyWrapper implements Request.Body {
private final Body body;
/**
* Create a new {@code DelegatingBody} that wraps the given body.
*
* @param body the body to wrap
*/
public BodyWrapper(Body body) {
Assert.notNull(body, "'body' must not be null");
this.body = body;
}
@Override
public Flux<DataBuffer> stream() {
return this.body.stream();
}
@Override
public <T> Flux<T> convertTo(Class<? extends T> aClass) {
return this.body.convertTo(aClass);
}
@Override
public <T> Mono<T> convertToMono(Class<? extends T> aClass) {
return this.body.convertToMono(aClass);
}
}
}

View File

@@ -0,0 +1,46 @@
/*
* 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.function.support;
import reactor.core.publisher.Mono;
import org.springframework.web.reactive.HandlerResult;
import org.springframework.web.reactive.HandlerResultHandler;
import org.springframework.web.reactive.function.Response;
import org.springframework.web.server.ServerWebExchange;
/**
* {@code HandlerResultHandler} implementation that supports {@link Response}s.
*
* @author Arjen Poutsma
* @since 5.0
*/
public class ResponseResultHandler implements HandlerResultHandler {
@Override
public boolean supports(HandlerResult result) {
Object returnValue = result.getReturnValue().orElse(null);
return returnValue instanceof Response;
}
@Override
public Mono<Void> handleResult(ServerWebExchange exchange, HandlerResult result) {
Response<?> response = (Response<?>) result.getReturnValue().orElseThrow(
IllegalStateException::new);
return response.writeTo(exchange);
}
}

View File

@@ -0,0 +1,7 @@
/**
* Classes supporting the {@code org.springframework.web.reactive.function} package.
* Contains a {@code HandlerAdapter} that supports {@code HandlerFunction}s,
* a {@code HandlerResultHandler} that supports {@code Response}s, and
* a {@code Request} wrapper to adapt a request.
*/
package org.springframework.web.reactive.function.support;

View File

@@ -0,0 +1,35 @@
/*
* 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.function;
import org.springframework.http.server.reactive.AbstractHttpHandlerIntegrationTests;
import org.springframework.http.server.reactive.HttpHandler;
/**
* @author Arjen Poutsma
*/
public abstract class AbstractRoutingFunctionIntegrationTests
extends AbstractHttpHandlerIntegrationTests {
@Override
protected final HttpHandler createHttpHandler() {
RoutingFunction<?> routingFunction = routingFunction();
return Router.toHttpHandler(routingFunction);
}
protected abstract RoutingFunction<?> routingFunction();
}

View File

@@ -0,0 +1,65 @@
/*
* 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.function;
import java.net.URI;
import java.util.Collections;
import org.junit.Test;
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.server.reactive.MockServerHttpRequest;
import org.springframework.http.server.reactive.MockServerHttpResponse;
import org.springframework.web.server.ServerWebExchange;
import org.springframework.web.server.adapter.DefaultServerWebExchange;
import org.springframework.web.server.session.MockWebSessionManager;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
/**
* @author Arjen Poutsma
*/
public class BodyResponseTests {
private final String body = "foo";
private final BodyResponse<String> publisherResponse =
new BodyResponse<>(200, new HttpHeaders(), body);
@Test
public void body() throws Exception {
assertEquals(body, publisherResponse.body());
}
@Test
public void writeTo() throws Exception {
MockServerHttpRequest request = new MockServerHttpRequest(HttpMethod.GET, URI.create("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<CharSequence>(new CharSequenceEncoder())).stream());
publisherResponse.writeTo(exchange).block();
assertNotNull(response.getBody());
}
}

View File

@@ -0,0 +1,72 @@
/*
* 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.function;
import org.junit.Before;
import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DefaultDataBufferFactory;
import org.springframework.web.reactive.function.support.RequestWrapper;
import static org.junit.Assert.assertSame;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
/**
* @author Arjen Poutsma
*/
public class BodyWrapperTests {
private Request.Body mockBody;
private RequestWrapper.BodyWrapper wrapper;
@Before
public void setUp() throws Exception {
mockBody = mock(Request.Body.class);
wrapper = new RequestWrapper.BodyWrapper(mockBody);
}
@Test
public void stream() throws Exception {
DataBuffer buffer = new DefaultDataBufferFactory().allocateBuffer();
Flux<DataBuffer> flux = Flux.just(buffer);
when(mockBody.stream()).thenReturn(flux);
assertSame(flux, wrapper.stream());
}
@Test
public void convertTo() throws Exception {
Flux<String> flux = Flux.just("foo", "bar");
when(mockBody.convertTo(String.class)).thenReturn(flux);
assertSame(flux, wrapper.convertTo(String.class));
}
@Test
public void convertToMono() throws Exception {
Mono<String> mono = Mono.just("foo");
when(mockBody.convertToMono(String.class)).thenReturn(mono);
assertSame(mono, wrapper.convertToMono(String.class));
}
}

View File

@@ -0,0 +1,40 @@
/*
* 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.function;
import org.junit.Test;
import static org.junit.Assert.assertEquals;
/**
* @author Arjen Poutsma
*/
public class DefaultConfigurationTests {
private DefaultConfiguration configuration = new DefaultConfiguration();
@Test
public void messageReaders() throws Exception {
assertEquals(4, configuration.messageReaders().get().count());
}
@Test
public void messageWriters() throws Exception {
assertEquals(4, configuration.messageWriters().get().count());
}
}

View File

@@ -0,0 +1,170 @@
/*
* 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.function;
import java.net.InetSocketAddress;
import java.net.URI;
import java.nio.ByteBuffer;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.OptionalLong;
import org.junit.Before;
import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.core.codec.StringDecoder;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DefaultDataBuffer;
import org.springframework.core.io.buffer.DefaultDataBufferFactory;
import org.springframework.http.HttpHeaders;
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.server.reactive.ServerHttpRequest;
import org.springframework.http.server.reactive.ServerHttpResponse;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
import org.springframework.web.server.ServerWebExchange;
import static org.junit.Assert.assertEquals;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
/**
* @author Arjen Poutsma
*/
public class DefaultRequestTests {
private ServerHttpRequest mockRequest;
private ServerHttpResponse mockResponse;
private ServerWebExchange mockExchange;
private DefaultRequest defaultRequest;
@Before
public void createMocks() {
mockRequest = mock(ServerHttpRequest.class);
mockResponse = mock(ServerHttpResponse.class);
mockExchange = mock(ServerWebExchange.class);
when(mockExchange.getRequest()).thenReturn(mockRequest);
when(mockExchange.getResponse()).thenReturn(mockResponse);
defaultRequest = new DefaultRequest(mockExchange);
}
@Test
public void method() throws Exception {
HttpMethod method = HttpMethod.HEAD;
when(mockRequest.getMethod()).thenReturn(method);
assertEquals(method, defaultRequest.method());
}
@Test
public void uri() throws Exception {
URI uri = URI.create("https://example.com");
when(mockRequest.getURI()).thenReturn(uri);
assertEquals(uri, defaultRequest.uri());
}
@Test
public void attributes() throws Exception {
Map<String, Object> attributes = new LinkedHashMap<>();
when(mockExchange.getAttributes()).thenReturn(attributes);
assertEquals(attributes, defaultRequest.attributes());
}
@Test
public void queryParams() throws Exception {
MultiValueMap<String, String> queryParams = new LinkedMultiValueMap<>();
queryParams.set("foo", "bar");
when(mockRequest.getQueryParams()).thenReturn(queryParams);
assertEquals(Optional.of("bar"), defaultRequest.queryParam("foo"));
}
@Test
public void pathVariables() throws Exception {
Map<String, String> pathVariables = Collections.singletonMap("foo", "bar");
when(mockExchange.getAttribute(Router.URI_TEMPLATE_VARIABLES_ATTRIBUTE)).thenReturn(Optional.of(pathVariables));
assertEquals(pathVariables, defaultRequest.pathVariables());
}
@Test
public void header() throws Exception {
HttpHeaders httpHeaders = new HttpHeaders();
List<MediaType> accept =
Collections.singletonList(MediaType.APPLICATION_JSON);
httpHeaders.setAccept(accept);
List<Charset> acceptCharset = Collections.singletonList(StandardCharsets.UTF_8);
httpHeaders.setAcceptCharset(acceptCharset);
long contentLength = 42L;
httpHeaders.setContentLength(contentLength);
MediaType contentType = MediaType.TEXT_PLAIN;
httpHeaders.setContentType(contentType);
InetSocketAddress host = InetSocketAddress.createUnresolved("localhost", 80);
httpHeaders.setHost(host);
List<HttpRange> range = Collections.singletonList(HttpRange.createByteRange(0, 42));
httpHeaders.setRange(range);
when(mockRequest.getHeaders()).thenReturn(httpHeaders);
Request.Headers headers = defaultRequest.headers();
assertEquals(accept, headers.accept());
assertEquals(acceptCharset, headers.acceptCharset());
assertEquals(OptionalLong.of(contentLength), headers.contentLength());
assertEquals(Optional.of(contentType), headers.contentType());
assertEquals(httpHeaders, headers.asHttpHeaders());
}
@Test
public void body() throws Exception {
DefaultDataBufferFactory factory = new DefaultDataBufferFactory();
DefaultDataBuffer dataBuffer =
factory.wrap(ByteBuffer.wrap("foo".getBytes(StandardCharsets.UTF_8)));
Flux<DataBuffer> body = Flux.just(dataBuffer);
HttpHeaders httpHeaders = new HttpHeaders();
httpHeaders.setContentType(MediaType.TEXT_PLAIN);
when(mockRequest.getHeaders()).thenReturn(httpHeaders);
when(mockRequest.getBody()).thenReturn(body);
when(mockExchange.getAttribute(Router.HTTP_MESSAGE_READERS_ATTRIBUTE))
.thenReturn(Optional.of(Collections
.singleton(new DecoderHttpMessageReader<String>(new StringDecoder()))
.stream()));
assertEquals(body, defaultRequest.body().stream());
Mono<String> resultMono = defaultRequest.body().convertToMono(String.class);
assertEquals("foo", resultMono.block());
}
}

View File

@@ -0,0 +1,209 @@
/*
* 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.function;
import java.util.List;
import java.util.function.Supplier;
import java.util.stream.Stream;
import org.junit.Before;
import org.junit.Test;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.http.HttpMethod;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.http.codec.HttpMessageReader;
import org.springframework.http.codec.HttpMessageWriter;
import org.springframework.http.server.reactive.AbstractHttpHandlerIntegrationTests;
import org.springframework.http.server.reactive.HttpHandler;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.reactive.DispatcherHandler;
import org.springframework.web.reactive.HandlerAdapter;
import org.springframework.web.reactive.HandlerMapping;
import org.springframework.web.reactive.config.WebReactiveConfiguration;
import org.springframework.web.reactive.function.support.HandlerFunctionAdapter;
import org.springframework.web.reactive.function.support.ResponseResultHandler;
import org.springframework.web.server.adapter.WebHttpHandlerBuilder;
import static org.junit.Assert.assertEquals;
import static org.springframework.web.reactive.function.Router.route;
/**
* Tests the use of {@link HandlerFunction} and {@link RoutingFunction} in a
* {@link DispatcherHandler}.
* @author Arjen Poutsma
*/
public class DispatcherHandlerIntegrationTests extends AbstractHttpHandlerIntegrationTests {
private AnnotationConfigApplicationContext wac;
private RestTemplate restTemplate;
@Before
public void createRestTemplate() {
this.restTemplate = new RestTemplate();
}
@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 mono() throws Exception {
ResponseEntity<Person> result =
restTemplate.getForEntity("http://localhost:" + port + "/mono", Person.class);
assertEquals(HttpStatus.OK, result.getStatusCode());
assertEquals("John", result.getBody().getName());
}
@Test
public void flux() throws Exception {
ParameterizedTypeReference<List<Person>> reference = new ParameterizedTypeReference<List<Person>>() {};
ResponseEntity<List<Person>> result =
restTemplate.exchange("http://localhost:" + port + "/flux", HttpMethod.GET, null, reference);
assertEquals(HttpStatus.OK, result.getStatusCode());
List<Person> body = result.getBody();
assertEquals(2, body.size());
assertEquals("John", body.get(0).getName());
assertEquals("Jane", body.get(1).getName());
}
@Configuration
static class TestConfiguration extends WebReactiveConfiguration {
@Bean
public PersonHandler personHandler() {
return new PersonHandler();
}
@Bean
public HandlerAdapter handlerAdapter() {
return new HandlerFunctionAdapter();
}
@Bean
public HandlerMapping handlerMapping(RoutingFunction<?> routingFunction,
ApplicationContext applicationContext) {
return Router.toHandlerMapping(routingFunction,
new Router.Configuration() {
@Override
public Supplier<Stream<HttpMessageReader<?>>> messageReaders() {
return () -> getMessageReaders().stream();
}
@Override
public Supplier<Stream<HttpMessageWriter<?>>> messageWriters() {
return () -> getMessageWriters().stream();
}
});
}
@Bean
public RoutingFunction<?> routingFunction() {
PersonHandler personHandler = personHandler();
return route(RequestPredicates.GET("/mono"), personHandler::mono)
.and(route(RequestPredicates.GET("/flux"), personHandler::flux));
}
@Bean
public ResponseResultHandler responseResultHandler() {
return new ResponseResultHandler();
}
}
private static class PersonHandler {
public Response<Publisher<Person>> mono(Request request) {
Person person = new Person("John");
return Response.ok().stream(Mono.just(person), Person.class);
}
public Response<Publisher<Person>> flux(Request request) {
Person person1 = new Person("John");
Person person2 = new Person("Jane");
return Response.ok().stream(Flux.just(person1, person2), Person.class);
}
}
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 + '\'' +
'}';
}
}
}

View File

@@ -0,0 +1,163 @@
/*
* 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.function;
import java.time.Duration;
import java.util.Objects;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.http.MediaType;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.http.server.reactive.HttpHandler;
import org.springframework.http.server.reactive.bootstrap.HttpServer;
import org.springframework.http.server.reactive.bootstrap.ReactorHttpServer;
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;
import static org.springframework.web.reactive.function.Router.toHttpHandler;
/**
* @author Arjen Poutsma
*/
public class Driver {
public static void main(String[] args) throws Exception {
PersonHandler handler = new PersonHandler();
RoutingFunction<Publisher<Person>> personRoute =
route(GET("/person/{id}"), handler::person)
.and(route(GET("/people"), handler::people))
.filter((request, next) -> {
System.out.println("In publisher filter");
Response<Publisher<Person>> handlerResponse = next.handle(request);
Publisher<Person> s = Flux.from(handlerResponse.body())
.map(person -> new Person(person.name.toUpperCase()));
return Response.from(handlerResponse).stream(s, Person.class);
});
RoutingFunction<Publisher<String>> stringRoute = route(POST("/string"), request -> {
Flux<String> requestBody = request.body().convertTo(String.class);
Flux<String> responseBody = Flux.concat(Mono.just("Hello "), requestBody);
return Response.ok().stream(responseBody, String.class);
});
RoutingFunction<?> sseRoute = route(GET("/sse"), request -> {
Flux<ServerSentEvent<Person>> eventFlux = Flux.interval(Duration.ofMillis(100)).map(l -> {
Person person = new Person("Person " + l);
return ServerSentEvent.<Person>builder().data(person)
.id(Long.toString(l))
.comment("bar")
.build();
}).take(20);
return Response.ok().sse(eventFlux);
}).andOther(route(GET("/sse-string"), request -> {
Flux<String> flux = Flux.interval(Duration.ofMillis(100)).map(l -> Long.toString(l)).take(20);
return Response.ok().sse(flux, String.class);
}));
createServer(toHttpHandler(
personRoute.andOther(stringRoute).andOther(sseRoute)
.filter((request, next) -> {
System.out.println("Before");
Response<?> result = next.handle(request);
System.out.println("After");
return result;
})
));
System.out.println("Press ENTER to exit.");
System.in.read();
}
private static HttpServer createServer(HttpHandler httpHandler) throws Exception {
HttpServer server = new ReactorHttpServer();
int port = 8080;
server.setPort(port);
server.setHandler(httpHandler);
server.afterPropertiesSet();
server.start();
System.out.println("Server started at http://localhost:" + port + "/");
return server;
}
private static class PersonHandler {
public Response<Publisher<Person>> person(Request r) {
System.out.println("r.pathVariable(id) = " + r.pathVariable("id"));
return Response.ok().contentType(MediaType.APPLICATION_JSON)
.stream(Mono.just(new Person("John")), Person.class);
}
public Response<Publisher<Person>> people(Request r) {
return Response.ok().contentType(MediaType.APPLICATION_JSON)
.stream(Flux.just(new Person("Jane"), new Person("John")), Person.class);
}
}
private static class Person {
private String name;
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 instanceof Person) {
Person other = (Person) o;
return Objects.equals(this.name, other.name);
}
return false;
}
@Override
public int hashCode() {
return Objects.hashCode(this.name);
}
@Override
public String toString() {
return "Person{" +
"name='" + name + '\'' +
'}';
}
}
}

View File

@@ -0,0 +1,71 @@
/*
* 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.function;
import org.junit.Test;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.server.reactive.MockServerHttpResponse;
import org.springframework.web.server.ServerWebExchange;
import static org.junit.Assert.*;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
/**
* @author Arjen Poutsma
*/
public class EmptyResponseTests {
@Test
public void statusCode() throws Exception {
HttpStatus statusCode = HttpStatus.ACCEPTED;
EmptyResponse emptyResponse = new EmptyResponse(statusCode.value(), new HttpHeaders());
assertSame(statusCode, emptyResponse.statusCode());
}
@Test
public void headers() throws Exception {
HttpHeaders headers = new HttpHeaders();
EmptyResponse emptyResponse = new EmptyResponse(200, headers);
assertEquals(headers, emptyResponse.headers());
}
@Test
public void body() throws Exception {
EmptyResponse emptyResponse = new EmptyResponse(200, new HttpHeaders());
assertNull(emptyResponse.body());
}
@Test
public void writeTo() throws Exception {
HttpHeaders headers = new HttpHeaders();
headers.add("MyKey", "MyValue");
EmptyResponse emptyResponse = new EmptyResponse(201, headers);
ServerWebExchange exchange = mock(ServerWebExchange.class);
MockServerHttpResponse response = new MockServerHttpResponse();
when(exchange.getResponse()).thenReturn(response);
emptyResponse.writeTo(exchange).block();
assertEquals(201, response.getStatusCode().value());
assertEquals("MyValue", response.getHeaders().getFirst("MyKey"));
assertNull(response.getBody());
}
}

View File

@@ -0,0 +1,120 @@
/*
* 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.function;
import java.net.InetSocketAddress;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
import java.util.OptionalLong;
import org.junit.Before;
import org.junit.Test;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpRange;
import org.springframework.http.MediaType;
import org.springframework.web.reactive.function.support.RequestWrapper;
import static org.junit.Assert.assertSame;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
/**
* @author Arjen Poutsma
*/
public class HeadersWrapperTest {
private Request.Headers mockHeaders;
private RequestWrapper.HeadersWrapper wrapper;
@Before
public void createWrapper() {
mockHeaders = mock(Request.Headers.class);
wrapper = new RequestWrapper.HeadersWrapper(mockHeaders);
}
@Test
public void accept() throws Exception {
List<MediaType> accept = Collections.singletonList(MediaType.APPLICATION_JSON);
when(mockHeaders.accept()).thenReturn(accept);
assertSame(accept, wrapper.accept());
}
@Test
public void acceptCharset() throws Exception {
List<Charset> acceptCharset = Collections.singletonList(StandardCharsets.UTF_8);
when(mockHeaders.acceptCharset()).thenReturn(acceptCharset);
assertSame(acceptCharset, wrapper.acceptCharset());
}
@Test
public void contentLength() throws Exception {
OptionalLong contentLength = OptionalLong.of(42L);
when(mockHeaders.contentLength()).thenReturn(contentLength);
assertSame(contentLength, wrapper.contentLength());
}
@Test
public void contentType() throws Exception {
Optional<MediaType> contentType = Optional.of(MediaType.APPLICATION_JSON);
when(mockHeaders.contentType()).thenReturn(contentType);
assertSame(contentType, wrapper.contentType());
}
@Test
public void host() throws Exception {
InetSocketAddress host = InetSocketAddress.createUnresolved("example.com", 42);
when(mockHeaders.host()).thenReturn(host);
assertSame(host, wrapper.host());
}
@Test
public void range() throws Exception {
List<HttpRange> range = Collections.singletonList(HttpRange.createByteRange(42));
when(mockHeaders.range()).thenReturn(range);
assertSame(range, wrapper.range());
}
@Test
public void header() throws Exception {
String name = "foo";
List<String> value = Collections.singletonList("bar");
when(mockHeaders.header(name)).thenReturn(value);
assertSame(value, wrapper.header(name));
}
@Test
public void asHttpHeaders() throws Exception {
HttpHeaders httpHeaders = new HttpHeaders();
when(mockHeaders.asHttpHeaders()).thenReturn(httpHeaders);
assertSame(httpHeaders, wrapper.asHttpHeaders());
}
}

View File

@@ -0,0 +1,367 @@
/*
* 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.function;
import java.net.InetSocketAddress;
import java.net.URI;
import java.nio.charset.Charset;
import java.time.Instant;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.OptionalLong;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.HttpRange;
import org.springframework.http.MediaType;
import org.springframework.util.Assert;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
/**
* @author Arjen Poutsma
*/
public class MockRequest implements Request {
private final HttpMethod method;
private final URI uri;
private final MockHeaders headers;
private final MockBody body;
private final Map<String, Object> attributes;
private final MultiValueMap<String, String> queryParams;
private final Map<String, String> pathVariables;
private MockRequest(HttpMethod method, URI uri,
MockHeaders headers, MockBody body, Map<String, Object> attributes,
MultiValueMap<String, String> queryParams,
Map<String, String> pathVariables) {
this.method = method;
this.uri = uri;
this.headers = headers;
this.body = body;
this.attributes = attributes;
this.queryParams = queryParams;
this.pathVariables = pathVariables;
}
public static Builder builder() {
return new BuilderImpl();
}
@Override
public HttpMethod method() {
return this.method;
}
@Override
public URI uri() {
return this.uri;
}
@Override
public Headers headers() {
return this.headers;
}
@Override
public Body body() {
return this.body;
}
@SuppressWarnings("unchecked")
@Override
public <T> Optional<T> attribute(String name) {
return Optional.ofNullable((T) this.attributes.get(name));
}
@Override
public Map<String, Object> attributes() {
return this.attributes;
}
@Override
public List<String> queryParams(String name) {
return Collections.unmodifiableList(this.queryParams.get(name));
}
@Override
public Map<String, String> pathVariables() {
return Collections.unmodifiableMap(this.pathVariables);
}
public interface Builder {
Builder method(HttpMethod method);
Builder uri(URI uri);
Builder header(String key, String value);
Builder headers(HttpHeaders headers);
Builder attribute(String name, Object value);
Builder attributes(Map<String, Object> attributes);
Builder queryParam(String key, String value);
Builder queryParams(MultiValueMap<String, String> queryParams);
Builder pathVariable(String key, String value);
Builder pathVariables(Map<String, String> pathVariables);
<T> MockRequest body(Flux<T> body);
<T> MockRequest body(Mono<T> body);
MockRequest build();
}
private static class BuilderImpl implements Builder {
private HttpMethod method = HttpMethod.GET;
private URI uri = URI.create("http://localhost");
private MockHeaders headers = new MockHeaders(new HttpHeaders());
private Map<String, Object> attributes = new LinkedHashMap<>();
private MultiValueMap<String, String> queryParams = new LinkedMultiValueMap<>();
private Map<String, String> pathVariables = new LinkedHashMap<>();
@Override
public Builder method(HttpMethod method) {
Assert.notNull(method, "'method' must not be null");
this.method = method;
return this;
}
@Override
public Builder uri(URI uri) {
Assert.notNull(uri, "'uri' must not be null");
this.uri = uri;
return this;
}
@Override
public Builder header(String key, String value) {
Assert.notNull(key, "'key' must not be null");
Assert.notNull(value, "'value' must not be null");
this.headers.header(key, value);
return this;
}
@Override
public Builder headers(HttpHeaders headers) {
Assert.notNull(headers, "'headers' must not be null");
this.headers = new MockHeaders(headers);
return this;
}
@Override
public Builder attribute(String name, Object value) {
Assert.notNull(name, "'name' must not be null");
Assert.notNull(value, "'value' must not be null");
this.attributes.put(name, value);
return this;
}
@Override
public Builder attributes(Map<String, Object> attributes) {
Assert.notNull(attributes, "'attributes' must not be null");
this.attributes = attributes;
return this;
}
@Override
public Builder queryParam(String key, String value) {
Assert.notNull(key, "'key' must not be null");
Assert.notNull(value, "'value' must not be null");
this.queryParams.add(key, value);
return this;
}
@Override
public Builder queryParams(MultiValueMap<String, String> queryParams) {
Assert.notNull(queryParams, "'queryParams' must not be null");
this.queryParams = queryParams;
return this;
}
@Override
public Builder pathVariable(String key, String value) {
Assert.notNull(key, "'key' must not be null");
Assert.notNull(value, "'value' must not be null");
this.pathVariables.put(key, value);
return this;
}
@Override
public Builder pathVariables(Map<String, String> pathVariables) {
Assert.notNull(pathVariables, "'pathVariables' must not be null");
this.pathVariables = pathVariables;
return this;
}
@Override
public <T> MockRequest body(Flux<T> flux) {
MockBody body = new MockBody() {
@SuppressWarnings("unchecked")
@Override
public <S> Flux<S> convertTo(Class<? extends S> aClass) {
return (Flux<S>) flux;
}
};
return build(body);
}
@Override
public <T> MockRequest body(Mono<T> mono) {
MockBody body = new MockBody() {
@SuppressWarnings("unchecked")
@Override
public <S> Mono<S> convertToMono(Class<? extends S> aClass) {
return (Mono<S>) mono;
}
};
return build(body);
}
@Override
public MockRequest build() {
return build(new MockBody());
}
private MockRequest build(MockBody body) {
return new MockRequest(this.method, this.uri, this.headers, body, this.attributes,
this.queryParams, this.pathVariables);
}
}
private static class MockHeaders implements Headers {
private final HttpHeaders headers;
public MockHeaders(HttpHeaders headers) {
this.headers = headers;
}
private HttpHeaders delegate() {
return this.headers;
}
public void header(String key, String value) {
this.headers.add(key, value);
}
@Override
public List<MediaType> accept() {
return delegate().getAccept();
}
@Override
public List<Charset> acceptCharset() {
return delegate().getAcceptCharset();
}
@Override
public OptionalLong contentLength() {
return toOptionalLong(delegate().getContentLength());
}
@Override
public Optional<MediaType> contentType() {
return Optional.ofNullable(delegate().getContentType());
}
@Override
public InetSocketAddress host() {
return delegate().getHost();
}
@Override
public List<HttpRange> range() {
return delegate().getRange();
}
@Override
public List<String> header(String headerName) {
List<String> headerValues = delegate().get(headerName);
return headerValues != null ? headerValues : Collections.emptyList();
}
@Override
public HttpHeaders asHttpHeaders() {
return HttpHeaders.readOnlyHttpHeaders(delegate());
}
private OptionalLong toOptionalLong(long value) {
return value != -1 ? OptionalLong.of(value) : OptionalLong.empty();
}
private Optional<ZonedDateTime> toZonedDateTime(long date) {
if (date != -1) {
Instant instant = Instant.ofEpochMilli(date);
return Optional.of(ZonedDateTime.ofInstant(instant, ZoneId.of("GMT")));
}
else {
return Optional.empty();
}
}
}
private static class MockBody implements Body {
@Override
public Flux<DataBuffer> stream() {
return Flux.empty();
}
@Override
public <T> Flux<T> convertTo(Class<? extends T> aClass) {
return Flux.empty();
}
@Override
public <T> Mono<T> convertToMono(Class<? extends T> aClass) {
return Mono.empty();
}
}
}

View File

@@ -0,0 +1,141 @@
/*
* 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.function;
import java.util.List;
import org.junit.Before;
import org.junit.Test;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.http.HttpMethod;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.client.RestTemplate;
import static org.junit.Assert.assertEquals;
import static org.springframework.web.reactive.function.Router.route;
/**
* @author Arjen Poutsma
*/
public class PublisherHandlerFunctionIntegrationTests
extends AbstractRoutingFunctionIntegrationTests {
private RestTemplate restTemplate;
@Before
public void createRestTemplate() {
this.restTemplate = new RestTemplate();
}
@Override
protected RoutingFunction<?> routingFunction() {
PersonHandler personHandler = new PersonHandler();
return route(RequestPredicates.GET("/mono"), personHandler::mono)
.and(route(RequestPredicates.GET("/flux"), personHandler::flux));
}
@Test
public void mono() throws Exception {
ResponseEntity<Person> result =
restTemplate.getForEntity("http://localhost:" + port + "/mono", Person.class);
assertEquals(HttpStatus.OK, result.getStatusCode());
assertEquals("John", result.getBody().getName());
}
@Test
public void flux() throws Exception {
ParameterizedTypeReference<List<Person>> reference = new ParameterizedTypeReference<List<Person>>() {};
ResponseEntity<List<Person>> result =
restTemplate.exchange("http://localhost:" + port + "/flux", HttpMethod.GET, null, reference);
assertEquals(HttpStatus.OK, result.getStatusCode());
List<Person> body = result.getBody();
assertEquals(2, body.size());
assertEquals("John", body.get(0).getName());
assertEquals("Jane", body.get(1).getName());
}
private static class PersonHandler {
public Response<Publisher<Person>> mono(Request request) {
Person person = new Person("John");
return Response.ok().stream(Mono.just(person), Person.class);
}
public Response<Publisher<Person>> flux(Request request) {
Person person1 = new Person("John");
Person person2 = new Person("Jane");
return Response.ok().stream(Flux.just(person1, person2), Person.class);
}
}
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 + '\'' +
'}';
}
}
}

View File

@@ -0,0 +1,67 @@
/*
* 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.function;
import java.net.URI;
import java.util.Collections;
import org.junit.Test;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
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.server.reactive.MockServerHttpRequest;
import org.springframework.http.server.reactive.MockServerHttpResponse;
import org.springframework.web.server.ServerWebExchange;
import org.springframework.web.server.adapter.DefaultServerWebExchange;
import org.springframework.web.server.session.MockWebSessionManager;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
/**
* @author Arjen Poutsma
*/
public class PublisherResponseTests {
private final Publisher<String> publisher = Flux.just("foo", "bar");
private final PublisherResponse<String> publisherResponse =
new PublisherResponse<>(200, new HttpHeaders(), publisher, String.class);
@Test
public void body() throws Exception {
assertEquals(publisher, publisherResponse.body());
}
@Test
public void writeTo() throws Exception {
MockServerHttpRequest request = new MockServerHttpRequest(HttpMethod.GET, URI.create("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<CharSequence>(new CharSequenceEncoder())).stream());
publisherResponse.writeTo(exchange).block();
assertNotNull(response.getBody());
}
}

View File

@@ -0,0 +1,66 @@
/*
* 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.function;
import org.junit.Test;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
/**
* @author Arjen Poutsma
*/
public class RequestPredicateTests {
@Test
public void and() throws Exception {
RequestPredicate predicate1 = request -> true;
RequestPredicate predicate2 = request -> true;
RequestPredicate predicate3 = request -> false;
MockRequest request = MockRequest.builder().build();
assertTrue(predicate1.and(predicate2).test(request));
assertTrue(predicate2.and(predicate1).test(request));
assertFalse(predicate1.and(predicate3).test(request));
}
@Test
public void negate() throws Exception {
RequestPredicate predicate = request -> false;
RequestPredicate negated = predicate.negate();
MockRequest mockRequest = MockRequest.builder().build();
assertTrue(negated.test(mockRequest));
predicate = request -> true;
negated = predicate.negate();
assertFalse(negated.test(mockRequest));
}
@Test
public void or() throws Exception {
RequestPredicate predicate1 = request -> true;
RequestPredicate predicate2 = request -> false;
RequestPredicate predicate3 = request -> false;
MockRequest request = MockRequest.builder().build();
assertTrue(predicate1.or(predicate2).test(request));
assertTrue(predicate2.or(predicate1).test(request));
assertFalse(predicate2.or(predicate3).test(request));
}
}

View File

@@ -0,0 +1,136 @@
/*
* 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.function;
import java.net.URI;
import java.util.Collections;
import org.junit.Test;
import org.springframework.http.HttpMethod;
import org.springframework.http.MediaType;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
/**
* @author Arjen Poutsma
*/
public class RequestPredicatesTests {
@Test
public void all() throws Exception {
RequestPredicate predicate = RequestPredicates.all();
MockRequest request = MockRequest.builder().build();
assertTrue(predicate.test(request));
}
@Test
public void method() throws Exception {
HttpMethod httpMethod = HttpMethod.GET;
RequestPredicate predicate = RequestPredicates.method(httpMethod);
MockRequest request = MockRequest.builder().method(httpMethod).build();
assertTrue(predicate.test(request));
request = MockRequest.builder().method(HttpMethod.POST).build();
assertFalse(predicate.test(request));
}
@Test
public void methods() throws Exception {
URI uri = URI.create("http://localhost/path");
RequestPredicate predicate = RequestPredicates.GET("/p*");
MockRequest request = MockRequest.builder().method(HttpMethod.GET).uri(uri).build();
assertTrue(predicate.test(request));
predicate = RequestPredicates.HEAD("/p*");
request = MockRequest.builder().method(HttpMethod.HEAD).uri(uri).build();
assertTrue(predicate.test(request));
predicate = RequestPredicates.POST("/p*");
request = MockRequest.builder().method(HttpMethod.POST).uri(uri).build();
assertTrue(predicate.test(request));
predicate = RequestPredicates.PUT("/p*");
request = MockRequest.builder().method(HttpMethod.PUT).uri(uri).build();
assertTrue(predicate.test(request));
predicate = RequestPredicates.PATCH("/p*");
request = MockRequest.builder().method(HttpMethod.PATCH).uri(uri).build();
assertTrue(predicate.test(request));
predicate = RequestPredicates.DELETE("/p*");
request = MockRequest.builder().method(HttpMethod.DELETE).uri(uri).build();
assertTrue(predicate.test(request));
predicate = RequestPredicates.OPTIONS("/p*");
request = MockRequest.builder().method(HttpMethod.OPTIONS).uri(uri).build();
assertTrue(predicate.test(request));
}
@Test
public void path() throws Exception {
URI uri = URI.create("http://localhost/path");
RequestPredicate predicate = RequestPredicates.path("/p*");
MockRequest request = MockRequest.builder().uri(uri).build();
assertTrue(predicate.test(request));
request = MockRequest.builder().build();
assertFalse(predicate.test(request));
}
@Test
public void headers() throws Exception {
String name = "MyHeader";
String value = "MyValue";
RequestPredicate predicate =
RequestPredicates.headers(headers -> {
return headers.header(name).equals(
Collections.singletonList(value));
});
MockRequest request = MockRequest.builder().header(name, value).build();
assertTrue(predicate.test(request));
request = MockRequest.builder().build();
assertFalse(predicate.test(request));
}
@Test
public void contentType() throws Exception {
MediaType json = MediaType.APPLICATION_JSON;
RequestPredicate predicate = RequestPredicates.contentType(json);
MockRequest request = MockRequest.builder().header("Content-Type", json.toString()).build();
assertTrue(predicate.test(request));
request = MockRequest.builder().build();
assertFalse(predicate.test(request));
}
@Test
public void accept() throws Exception {
MediaType json = MediaType.APPLICATION_JSON;
RequestPredicate predicate = RequestPredicates.accept(json);
MockRequest request = MockRequest.builder().header("Accept", json.toString()).build();
assertTrue(predicate.test(request));
request = MockRequest.builder().build();
assertFalse(predicate.test(request));
}
}

View File

@@ -0,0 +1,148 @@
/*
* 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.function;
import java.net.URI;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import org.junit.Before;
import org.junit.Test;
import org.springframework.http.HttpMethod;
import org.springframework.web.reactive.function.support.RequestWrapper;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertSame;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
/**
* @author Arjen Poutsma
*/
public class RequestWrapperTests {
private Request mockRequest;
private RequestWrapper wrapper;
@Before
public void createWrapper() {
mockRequest = mock(Request.class);
wrapper = new RequestWrapper(mockRequest);
}
@Test
public void request() throws Exception {
assertSame(mockRequest, wrapper.request());
}
@Test
public void method() throws Exception {
HttpMethod method = HttpMethod.POST;
when(mockRequest.method()).thenReturn(method);
assertSame(method, wrapper.method());
}
@Test
public void uri() throws Exception {
URI uri = URI.create("https://example.com");
when(mockRequest.uri()).thenReturn(uri);
assertSame(uri, wrapper.uri());
}
@Test
public void path() throws Exception {
String path = "/foo/bar";
when(mockRequest.path()).thenReturn(path);
assertSame(path, wrapper.path());
}
@Test
public void headers() throws Exception {
Request.Headers headers = mock(Request.Headers.class);
when(mockRequest.headers()).thenReturn(headers);
assertSame(headers, wrapper.headers());
}
@Test
public void body() throws Exception {
Request.Body body = mock(Request.Body.class);
when(mockRequest.body()).thenReturn(body);
assertEquals(body, wrapper.body());
}
@Test
public void attribute() throws Exception {
String name = "foo";
String value = "bar";
when(mockRequest.attribute(name)).thenReturn(Optional.of(value));
assertEquals(Optional.of(value), wrapper.attribute(name));
}
@Test
public void attributes() throws Exception {
Map<String, Object> attributes = Collections.singletonMap("foo", "bar");
when(mockRequest.attributes()).thenReturn(attributes);
assertSame(attributes, wrapper.attributes());
}
@Test
public void queryParam() throws Exception {
String name = "foo";
String value = "bar";
when(mockRequest.queryParam(name)).thenReturn(Optional.of(value));
assertEquals(Optional.of(value), wrapper.queryParam(name));
}
@Test
public void queryParams() throws Exception {
String name = "foo";
List<String> value = Collections.singletonList("bar");
when(mockRequest.queryParams(name)).thenReturn(value);
assertSame(value, wrapper.queryParams(name));
}
@Test
public void pathVariable() throws Exception {
String name = "foo";
String value = "bar";
when(mockRequest.pathVariable(name)).thenReturn(Optional.of(value));
assertEquals(Optional.of(value), wrapper.pathVariable(name));
}
@Test
public void pathVariables() throws Exception {
Map<String, String> pathVariables = Collections.singletonMap("foo", "bar");
when(mockRequest.pathVariables()).thenReturn(pathVariables);
assertSame(pathVariables, wrapper.pathVariables());
}
}

View File

@@ -0,0 +1,58 @@
/*
* 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.function;
import org.junit.Test;
import org.springframework.core.io.ClassPathResource;
import org.springframework.core.io.Resource;
import org.springframework.http.HttpHeaders;
import org.springframework.http.server.reactive.MockServerHttpResponse;
import org.springframework.web.server.ServerWebExchange;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
/**
* @author Arjen Poutsma
*/
public class ResourceResponseTests {
private final Resource resource = new ClassPathResource("response.txt", ResourceResponseTests.class);
private final ResourceResponse resourceResponse =
new ResourceResponse(200, new HttpHeaders(), resource);
@Test
public void body() throws Exception {
assertEquals(resource, resourceResponse.body());
}
@Test
public void writeTo() throws Exception {
ServerWebExchange exchange = mock(ServerWebExchange.class);
MockServerHttpResponse response = new MockServerHttpResponse();
when(exchange.getResponse()).thenReturn(response);
resourceResponse.writeTo(exchange).block();
assertNotNull(response.getBody());
}
}

View File

@@ -0,0 +1,147 @@
/*
* 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.function;
import java.net.URI;
import java.time.ZonedDateTime;
import java.util.Collections;
import org.junit.Test;
import org.springframework.http.CacheControl;
import org.springframework.http.HttpMethod;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import static org.junit.Assert.assertEquals;
/**
* @author Arjen Poutsma
*/
public class ResponseTests {
@Test
public void from() throws Exception {
Response<Void> other = Response.ok().header("foo", "bar").build();
Response<Void> result = Response.from(other).build();
assertEquals(HttpStatus.OK, result.statusCode());
assertEquals("bar", result.headers().getFirst("foo"));
}
@Test
public void status() throws Exception {
Response<Void> result = Response.status(HttpStatus.CREATED).build();
assertEquals(HttpStatus.CREATED, result.statusCode());
}
@Test
public void statusInt() throws Exception {
Response<Void> result = Response.status(201).build();
assertEquals(HttpStatus.CREATED, result.statusCode());
}
@Test
public void ok() throws Exception {
Response<Void> result = Response.ok().build();
assertEquals(HttpStatus.OK, result.statusCode());
}
@Test
public void created() throws Exception {
URI location = URI.create("http://example.com");
Response<Void> result = Response.created(location).build();
assertEquals(HttpStatus.CREATED, result.statusCode());
assertEquals(location, result.headers().getLocation());
}
@Test
public void accepted() throws Exception {
Response<Void> result = Response.accepted().build();
assertEquals(HttpStatus.ACCEPTED, result.statusCode());
}
@Test
public void noContent() throws Exception {
Response<Void> result = Response.noContent().build();
assertEquals(HttpStatus.NO_CONTENT, result.statusCode());
}
@Test
public void badRequest() throws Exception {
Response<Void> result = Response.badRequest().build();
assertEquals(HttpStatus.BAD_REQUEST, result.statusCode());
}
@Test
public void notFound() throws Exception {
Response<Void> result = Response.notFound().build();
assertEquals(HttpStatus.NOT_FOUND, result.statusCode());
}
@Test
public void unprocessableEntity() throws Exception {
Response<Void> result = Response.unprocessableEntity().build();
assertEquals(HttpStatus.UNPROCESSABLE_ENTITY, result.statusCode());
}
@Test
public void allow() throws Exception {
Response<Void> result = Response.ok().allow(HttpMethod.GET).build();
assertEquals(Collections.singleton(HttpMethod.GET), result.headers().getAllow());
}
@Test
public void contentLength() throws Exception {
Response<Void> result = Response.ok().contentLength(42).build();
assertEquals(42, result.headers().getContentLength());
}
@Test
public void contentType() throws Exception {
Response<Void> result = Response.ok().contentType(MediaType.APPLICATION_JSON).build();
assertEquals(MediaType.APPLICATION_JSON, result.headers().getContentType());
}
@Test
public void eTag() throws Exception {
Response<Void> result = Response.ok().eTag("foo").build();
assertEquals("\"foo\"", result.headers().getETag());
}
@Test
public void lastModified() throws Exception {
ZonedDateTime now = ZonedDateTime.now();
Response<Void> result = Response.ok().lastModified(now).build();
assertEquals(now.toInstant().toEpochMilli()/1000, result.headers().getLastModified()/1000);
}
@Test
public void cacheControlTag() throws Exception {
Response<Void> result = Response.ok().cacheControl(CacheControl.noCache()).build();
assertEquals("no-cache", result.headers().getCacheControl());
}
@Test
public void varyBy() throws Exception {
Response<Void> result = Response.ok().varyBy("foo").build();
assertEquals(Collections.singletonList("foo"), result.headers().getVary());
}
@Test
public void writeTo() throws Exception {
}
}

View File

@@ -0,0 +1,202 @@
/*
* 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.function;
import java.net.URI;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
import org.junit.Test;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.context.support.StaticApplicationContext;
import org.springframework.core.ResolvableType;
import org.springframework.http.HttpMethod;
import org.springframework.http.MediaType;
import org.springframework.http.ReactiveHttpInputMessage;
import org.springframework.http.ReactiveHttpOutputMessage;
import org.springframework.http.codec.HttpMessageReader;
import org.springframework.http.codec.HttpMessageWriter;
import org.springframework.http.server.reactive.HttpHandler;
import org.springframework.http.server.reactive.MockServerHttpRequest;
import org.springframework.http.server.reactive.MockServerHttpResponse;
import org.springframework.web.server.ServerWebExchange;
import static org.junit.Assert.*;
import static org.mockito.Mockito.*;
/**
* @author Arjen Poutsma
*/
@SuppressWarnings("unchecked")
public class RouterTests {
@Test
public void routeMatch() throws Exception {
HandlerFunction<Void> handlerFunction = request -> Response.ok().build();
MockRequest request = MockRequest.builder().build();
RequestPredicate requestPredicate = mock(RequestPredicate.class);
when(requestPredicate.test(request)).thenReturn(true);
RoutingFunction<Void> result = Router.route(requestPredicate, handlerFunction);
assertNotNull(result);
Optional<HandlerFunction<Void>> resultHandlerFunction = result.route(request);
assertTrue(resultHandlerFunction.isPresent());
assertEquals(handlerFunction, resultHandlerFunction.get());
}
@Test
public void routeNoMatch() throws Exception {
HandlerFunction<Void> handlerFunction = request -> Response.ok().build();
MockRequest request = MockRequest.builder().build();
RequestPredicate requestPredicate = mock(RequestPredicate.class);
when(requestPredicate.test(request)).thenReturn(false);
RoutingFunction<Void> result = Router.route(requestPredicate, handlerFunction);
assertNotNull(result);
Optional<HandlerFunction<Void>> resultHandlerFunction = result.route(request);
assertFalse(resultHandlerFunction.isPresent());
}
@Test
public void subrouteMatch() throws Exception {
HandlerFunction<Void> handlerFunction = request -> Response.ok().build();
RoutingFunction<Void> routingFunction = request -> Optional.of(handlerFunction);
MockRequest request = MockRequest.builder().build();
RequestPredicate requestPredicate = mock(RequestPredicate.class);
when(requestPredicate.test(request)).thenReturn(true);
RoutingFunction<Void> result = Router.subroute(requestPredicate, routingFunction);
assertNotNull(result);
Optional<HandlerFunction<Void>> resultHandlerFunction = result.route(request);
assertTrue(resultHandlerFunction.isPresent());
assertEquals(handlerFunction, resultHandlerFunction.get());
}
@Test
public void subrouteNoMatch() throws Exception {
HandlerFunction<Void> handlerFunction = request -> Response.ok().build();
RoutingFunction<Void> routingFunction = request -> Optional.of(handlerFunction);
MockRequest request = MockRequest.builder().build();
RequestPredicate requestPredicate = mock(RequestPredicate.class);
when(requestPredicate.test(request)).thenReturn(false);
RoutingFunction<Void> result = Router.subroute(requestPredicate, routingFunction);
assertNotNull(result);
Optional<HandlerFunction<Void>> resultHandlerFunction = result.route(request);
assertFalse(resultHandlerFunction.isPresent());
}
@Test
public void toHttpHandler() throws Exception {
Request request = mock(Request.class);
Response response = mock(Response.class);
when(response.writeTo(any(ServerWebExchange.class))).thenReturn(Mono.empty());
HandlerFunction handlerFunction = mock(HandlerFunction.class);
when(handlerFunction.handle(any(Request.class))).thenReturn(response);
RoutingFunction routingFunction = mock(RoutingFunction.class);
when(routingFunction.route(any(Request.class))).thenReturn(Optional.of(handlerFunction));
RequestPredicate requestPredicate = mock(RequestPredicate.class);
when(requestPredicate.test(request)).thenReturn(false);
Router.Configuration configuration = mock(Router.Configuration.class);
when(configuration.messageReaders()).thenReturn(
() -> Collections.<HttpMessageReader<?>>emptyList().stream());
when(configuration.messageWriters()).thenReturn(
() -> Collections.<HttpMessageWriter<?>>emptyList().stream());
HttpHandler result = Router.toHttpHandler(routingFunction, configuration);
assertNotNull(result);
MockServerHttpRequest serverHttpRequest = new MockServerHttpRequest(HttpMethod.GET,
URI.create("http://localhost"));
MockServerHttpResponse serverHttpResponse = new MockServerHttpResponse();
result.handle(serverHttpRequest, serverHttpResponse);
}
@Test
public void toConfiguration() throws Exception {
StaticApplicationContext applicationContext = new StaticApplicationContext();
applicationContext.registerSingleton("messageWriter", DummyMessageWriter.class);
applicationContext.registerSingleton("messageReader", DummyMessageReader.class);
applicationContext.refresh();
Router.Configuration configuration = Router.toConfiguration(applicationContext);
assertTrue(configuration.messageReaders().get()
.allMatch(r -> r instanceof DummyMessageReader));
assertTrue(configuration.messageWriters().get()
.allMatch(r -> r instanceof DummyMessageWriter));
}
private static class DummyMessageWriter implements HttpMessageWriter<Object> {
@Override
public boolean canWrite(ResolvableType type, MediaType mediaType) {
return false;
}
@Override
public List<MediaType> getWritableMediaTypes() {
return Collections.emptyList();
}
@Override
public Mono<Void> write(Publisher<?> inputStream, ResolvableType type,
MediaType contentType,
ReactiveHttpOutputMessage outputMessage) {
return Mono.empty();
}
}
private static class DummyMessageReader implements HttpMessageReader<Object> {
@Override
public boolean canRead(ResolvableType type, MediaType mediaType) {
return false;
}
@Override
public List<MediaType> getReadableMediaTypes() {
return Collections.emptyList();
}
@Override
public Flux<Object> read(ResolvableType type, ReactiveHttpInputMessage inputMessage) {
return Flux.empty();
}
@Override
public Mono<Object> readMono(ResolvableType type, ReactiveHttpInputMessage inputMessage) {
return Mono.empty();
}
}
}

View File

@@ -0,0 +1,81 @@
/*
* 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.function;
import java.util.Optional;
import org.junit.Test;
import static org.junit.Assert.*;
/**
* @author Arjen Poutsma
*/
@SuppressWarnings("unchecked")
public class RoutingFunctionTests {
@Test
public void and() throws Exception {
HandlerFunction<Void> handlerFunction = request -> Response.ok().build();
RoutingFunction<Void> routingFunction1 = request -> Optional.empty();
RoutingFunction<Void> routingFunction2 = request -> Optional.of(handlerFunction);
RoutingFunction<Void> result = routingFunction1.and(routingFunction2);
assertNotNull(result);
MockRequest request = MockRequest.builder().build();
Optional<HandlerFunction<Void>> resultHandlerFunction = result.route(request);
assertTrue(resultHandlerFunction.isPresent());
assertEquals(handlerFunction, resultHandlerFunction.get());
}
@Test
public void andOther() throws Exception {
HandlerFunction<String> handlerFunction = request -> Response.ok().body("42");
RoutingFunction<Void> routingFunction1 = request -> Optional.empty();
RoutingFunction<String> routingFunction2 = request -> Optional.of(handlerFunction);
RoutingFunction<?> result = routingFunction1.andOther(routingFunction2);
assertNotNull(result);
MockRequest request = MockRequest.builder().build();
Optional<? extends HandlerFunction<?>> resultHandlerFunction = result.route(request);
assertTrue(resultHandlerFunction.isPresent());
assertEquals(handlerFunction, resultHandlerFunction.get());
}
@Test
public void filter() throws Exception {
HandlerFunction<String> handlerFunction = request -> Response.ok().body("42");
RoutingFunction<String> routingFunction = request -> Optional.of(handlerFunction);
FilterFunction<String, Integer> filterFunction = (request, next) -> {
Response<String> response = next.handle(request);
int i = Integer.parseInt(response.body());
return Response.ok().body(i);
};
RoutingFunction<Integer> result = routingFunction.filter(filterFunction);
assertNotNull(result);
MockRequest request = MockRequest.builder().build();
Optional<? extends HandlerFunction<?>> resultHandlerFunction = result.route(request);
assertTrue(resultHandlerFunction.isPresent());
Response<?> resultResponse = resultHandlerFunction.get().handle(request);
assertEquals(42, resultResponse.body());
}
}

View File

@@ -0,0 +1,72 @@
/*
* 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.function;
import java.net.URI;
import org.junit.Test;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Mono;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.http.server.reactive.MockServerHttpRequest;
import org.springframework.http.server.reactive.MockServerHttpResponse;
import org.springframework.tests.TestSubscriber;
import org.springframework.web.server.ServerWebExchange;
import org.springframework.web.server.adapter.DefaultServerWebExchange;
import org.springframework.web.server.session.MockWebSessionManager;
import static org.junit.Assert.assertEquals;
/**
* @author Arjen Poutsma
*/
public class ServerSentEventResponseTests {
private final ServerSentEvent<String> sse =
ServerSentEvent.<String>builder().data("42").build();
private final Publisher<ServerSentEvent<String>> body = Mono.just(sse);
private final ServerSentEventResponse<ServerSentEvent<String>> sseResponse =
ServerSentEventResponse.fromSseEvents(200, new HttpHeaders(), body);
@Test
public void body() throws Exception {
assertEquals(body, sseResponse.body());
}
@Test
public void writeTo() throws Exception {
MockServerHttpRequest request = new MockServerHttpRequest(HttpMethod.GET, URI.create("http://localhost"));
MockServerHttpResponse response = new MockServerHttpResponse();
ServerWebExchange exchange = new DefaultServerWebExchange(request, response, new MockWebSessionManager());
sseResponse.writeTo(exchange);
Publisher<Publisher<DataBuffer>> result = response.getBodyWithFlush();
TestSubscriber.subscribe(result).
assertNoError().
assertValuesWith(publisher -> {
TestSubscriber.subscribe(publisher).assertNoError();
});
}
}

View File

@@ -0,0 +1,180 @@
/*
* 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.function;
import java.time.Duration;
import org.junit.Before;
import org.junit.Test;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.http.MediaType;
import org.springframework.http.client.reactive.ReactorClientHttpConnector;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.tests.TestSubscriber;
import org.springframework.web.client.reactive.WebClient;
import static org.springframework.web.client.reactive.ClientWebRequestBuilders.get;
import static org.springframework.web.client.reactive.ResponseExtractors.bodyStream;
import static org.springframework.web.reactive.function.Router.route;
/**
* @author Arjen Poutsma
*/
public class SseHandlerFunctionIntegrationTests
extends AbstractRoutingFunctionIntegrationTests {
private WebClient webClient;
@Before
public void createWebClient() {
this.webClient = new WebClient(new ReactorClientHttpConnector());
}
@Override
protected RoutingFunction<?> routingFunction() {
SseHandler sseHandler = new SseHandler();
return route(RequestPredicates.GET("/string"), sseHandler::string)
.andOther(route(RequestPredicates.GET("/person"), sseHandler::person))
.andOther(route(RequestPredicates.GET("/event"), sseHandler::sse));
}
@Test
public void sseAsString() throws Exception {
Flux<String> result = this.webClient
.perform(get("http://localhost:" + port + "/string")
.accept(new MediaType("text", "event-stream")))
.extract(bodyStream(String.class))
.filter(s -> !s.equals("\n"))
.map(s -> (s.replace("\n", "")))
.take(2);
TestSubscriber
.subscribe(result)
.await(Duration.ofSeconds(5))
.assertValues("data:foo 0", "data:foo 1");
}
@Test
public void sseAsPerson() throws Exception {
Mono<String> result = this.webClient
.perform(get("http://localhost:" + port + "/person")
.accept(new MediaType("text", "event-stream")))
.extract(bodyStream(String.class))
.filter(s -> !s.equals("\n"))
.map(s -> s.replace("\n", ""))
.takeUntil(s -> s.endsWith("foo 1\"}"))
.reduce((s1, s2) -> s1 + s2);
TestSubscriber
.subscribe(result)
.await(Duration.ofSeconds(5))
.assertValues("data:{\"name\":\"foo 0\"}data:{\"name\":\"foo 1\"}");
}
@Test
public void sseAsEvent() throws Exception {
Flux<String> result = this.webClient
.perform(get("http://localhost:" + port + "/event")
.accept(new MediaType("text", "event-stream")))
.extract(bodyStream(String.class))
.filter(s -> !s.equals("\n"))
.map(s -> s.replace("\n", ""))
.take(2);
TestSubscriber
.subscribe(result)
.await(Duration.ofSeconds(5))
.assertValues(
"id:0:bardata:foo",
"id:1:bardata:foo"
);
}
private static class SseHandler {
public Response<Publisher<String>> string(Request request) {
Flux<String> flux = Flux.interval(Duration.ofMillis(100)).map(l -> "foo " + l).take(2);
return Response.ok().sse(flux, String.class);
}
public Response<Publisher<Person>> person(Request request) {
Flux<Person> flux = Flux.interval(Duration.ofMillis(100))
.map(l -> new Person("foo " + l)).take(2);
return Response.ok().sse(flux, Person.class);
}
public Response<Publisher<ServerSentEvent<String>>> sse(Request request) {
Flux<ServerSentEvent<String>> flux = Flux.interval(Duration.ofMillis(100))
.map(l -> ServerSentEvent.<String>builder().data("foo")
.id(Long.toString(l))
.comment("bar")
.build()).take(2);
return Response.ok().sse(flux);
}
}
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 + '\'' +
'}';
}
}
}

View File

@@ -0,0 +1,2 @@
Hello World
This is a sample response text file.