diff --git a/spring-web-reactive/src/main/java/org/springframework/http/ZeroCopyHttpOutputMessage.java b/spring-web-reactive/src/main/java/org/springframework/http/ZeroCopyHttpOutputMessage.java new file mode 100644 index 0000000000..9499c8d81f --- /dev/null +++ b/spring-web-reactive/src/main/java/org/springframework/http/ZeroCopyHttpOutputMessage.java @@ -0,0 +1,42 @@ +/* + * 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.http; + +import java.io.File; + +import reactor.core.publisher.Mono; + +/** + * Sub-interface of {@code ReactiveOutputMessage} that has support for "zero-copy" + * file transfers. + * + * @author Arjen Poutsma + * @see Zero-copy + */ +public interface ZeroCopyHttpOutputMessage extends ReactiveHttpOutputMessage { + + /** + * Set the body of the message to the given {@link File} which will be + * used to write to the underlying HTTP layer. + * @param file the file to transfer + * @param position the position within the file from which the transfer is to begin + * @param count the number of bytes to be transferred + * @return a publisher that indicates completion or error. + */ + Mono setBody(File file, long position, long count); + +} diff --git a/spring-web-reactive/src/main/java/org/springframework/http/converter/reactive/CodecHttpMessageConverter.java b/spring-web-reactive/src/main/java/org/springframework/http/converter/reactive/CodecHttpMessageConverter.java new file mode 100644 index 0000000000..b55ad9c4cb --- /dev/null +++ b/spring-web-reactive/src/main/java/org/springframework/http/converter/reactive/CodecHttpMessageConverter.java @@ -0,0 +1,102 @@ +/* + * 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.http.converter.reactive; + +import java.util.Collections; +import java.util.List; +import java.util.stream.Collectors; + +import org.reactivestreams.Publisher; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import org.springframework.core.ResolvableType; +import org.springframework.core.codec.Decoder; +import org.springframework.core.codec.Encoder; +import org.springframework.core.io.buffer.DataBuffer; +import org.springframework.core.io.buffer.DataBufferAllocator; +import org.springframework.http.MediaType; +import org.springframework.http.ReactiveHttpInputMessage; +import org.springframework.http.ReactiveHttpOutputMessage; +import org.springframework.http.support.MediaTypeUtils; + +/** + * @author Arjen Poutsma + */ +public class CodecHttpMessageConverter implements HttpMessageConverter { + + private final Encoder encoder; + + private final Decoder decoder; + + public CodecHttpMessageConverter(Encoder encoder, Decoder decoder) { + this.encoder = encoder; + this.decoder = decoder; + } + + @Override + public boolean canRead(ResolvableType type, MediaType mediaType) { + return this.decoder != null && this.decoder.canDecode(type, mediaType); + } + + @Override + public boolean canWrite(ResolvableType type, MediaType mediaType) { + return this.encoder != null && this.encoder.canEncode(type, mediaType); + } + + @Override + public List getReadableMediaTypes() { + return this.decoder != null ? this.decoder.getSupportedMimeTypes().stream(). + map(MediaTypeUtils::toMediaType). + collect(Collectors.toList()) : Collections.emptyList(); + } + + @Override + public List getWritableMediaTypes() { + return this.encoder != null ? this.encoder.getSupportedMimeTypes().stream(). + map(MediaTypeUtils::toMediaType). + collect(Collectors.toList()) : Collections.emptyList(); + } + + @Override + public Flux read(ResolvableType type, ReactiveHttpInputMessage inputMessage) { + if (this.decoder == null) { + return Flux.error(new IllegalStateException("No decoder set")); + } + MediaType contentType = inputMessage.getHeaders().getContentType(); + if (contentType == null) { + contentType = MediaType.APPLICATION_OCTET_STREAM; + } + + Flux body = inputMessage.getBody(); + + return this.decoder.decode(body, type, contentType); + } + + @Override + public Mono write(Publisher inputStream, ResolvableType type, + MediaType contentType, + ReactiveHttpOutputMessage outputMessage) { + if (this.encoder == null) { + return Mono.error(new IllegalStateException("No decoder set")); + } + outputMessage.getHeaders().setContentType(contentType); + DataBufferAllocator allocator = outputMessage.allocator(); + Flux body = encoder.encode(inputStream, allocator, type, contentType); + return outputMessage.setBody(body); + } +} diff --git a/spring-web-reactive/src/main/java/org/springframework/http/converter/reactive/HttpMessageConverter.java b/spring-web-reactive/src/main/java/org/springframework/http/converter/reactive/HttpMessageConverter.java new file mode 100644 index 0000000000..ce001c5d08 --- /dev/null +++ b/spring-web-reactive/src/main/java/org/springframework/http/converter/reactive/HttpMessageConverter.java @@ -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.http.converter.reactive; + +import java.util.List; + +import org.reactivestreams.Publisher; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import org.springframework.core.ResolvableType; +import org.springframework.http.MediaType; +import org.springframework.http.ReactiveHttpInputMessage; +import org.springframework.http.ReactiveHttpOutputMessage; + +/** + * Strategy interface that specifies a converter that can convert from and to HTTP + * requests and responses. + * @author Arjen Poutsma + */ +public interface HttpMessageConverter { + + /** + * Indicates whether the given class can be read by this converter. + * @param type the type to test for readability + * @param mediaType the media type to read, can be {@code null} if not specified. + * Typically the value of a {@code Content-Type} header. + * @return {@code true} if readable; {@code false} otherwise + */ + boolean canRead(ResolvableType type, MediaType mediaType); + + /** + * Return the list of {@link MediaType} objects that can be read by this converter. + * @return the list of supported readable media types + */ + List getReadableMediaTypes(); + + /** + * Read an object of the given type form the given input message, and returns it. + * @param type the type of object to return. This type must have previously been + * passed to the + * {@link #canRead canRead} method of this interface, which must have returned {@code + * true}. + * @param inputMessage the HTTP input message to read from + * @return the converted object + */ + Flux read(ResolvableType type, ReactiveHttpInputMessage inputMessage); + + /** + * Indicates whether the given class can be written by this converter. + * @param type the class to test for writability + * @param mediaType the media type to write, can be {@code null} if not specified. + * Typically the value of an {@code Accept} header. + * @return {@code true} if writable; {@code false} otherwise + */ + boolean canWrite(ResolvableType type, MediaType mediaType); + + /** + * Return the list of {@link MediaType} objects that can be written by this + * converter. + * @return the list of supported readable media types + */ + List getWritableMediaTypes(); + + /** + * Write an given object to the given output message. + * @param inputStream the input stream to write + * @param type the stream element type to process. + * @param contentType the content type to use when writing. May be {@code null} to + * indicate that the default content type of the converter must be used. + * @param outputMessage the message to write to + * @return + */ + Mono write(Publisher inputStream, + ResolvableType type, MediaType contentType, + ReactiveHttpOutputMessage outputMessage); +} \ No newline at end of file diff --git a/spring-web-reactive/src/main/java/org/springframework/http/converter/reactive/ResourceHttpMessageConverter.java b/spring-web-reactive/src/main/java/org/springframework/http/converter/reactive/ResourceHttpMessageConverter.java new file mode 100644 index 0000000000..5091fc6fd2 --- /dev/null +++ b/spring-web-reactive/src/main/java/org/springframework/http/converter/reactive/ResourceHttpMessageConverter.java @@ -0,0 +1,231 @@ +/* + * 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.http.converter.reactive; + +import java.io.File; +import java.io.IOException; +import java.io.InputStream; +import java.net.URI; +import java.util.Collections; +import java.util.List; +import java.util.Optional; + +import org.reactivestreams.Publisher; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import org.springframework.core.ResolvableType; +import org.springframework.core.io.ByteArrayResource; +import org.springframework.core.io.DescriptiveResource; +import org.springframework.core.io.InputStreamResource; +import org.springframework.core.io.Resource; +import org.springframework.core.io.buffer.DataBuffer; +import org.springframework.core.io.buffer.support.DataBufferUtils; +import org.springframework.http.HttpHeaders; +import org.springframework.http.HttpRangeResource; +import org.springframework.http.MediaType; +import org.springframework.http.ReactiveHttpInputMessage; +import org.springframework.http.ReactiveHttpOutputMessage; +import org.springframework.http.ZeroCopyHttpOutputMessage; +import org.springframework.http.support.MediaTypeUtils; +import org.springframework.util.MimeTypeUtils2; +import org.springframework.util.ResourceUtils; +import org.springframework.util.StreamUtils; + +/** + * @author Arjen Poutsma + */ +public class ResourceHttpMessageConverter implements HttpMessageConverter { + + private static final int BUFFER_SIZE = StreamUtils.BUFFER_SIZE; + + private static final List SUPPORTED_MEDIA_TYPES = + Collections.singletonList(MediaType.ALL); + + @Override + public boolean canRead(ResolvableType type, MediaType mediaType) { + return Resource.class.isAssignableFrom(type.getRawClass()); + } + + @Override + public boolean canWrite(ResolvableType type, MediaType mediaType) { + return Resource.class.isAssignableFrom(type.getRawClass()); + } + + @Override + public List getReadableMediaTypes() { + return SUPPORTED_MEDIA_TYPES; + } + + @Override + public List getWritableMediaTypes() { + return SUPPORTED_MEDIA_TYPES; + } + + @Override + public Flux read(ResolvableType type, + ReactiveHttpInputMessage inputMessage) { + Class clazz = type.getRawClass(); + + Flux body = inputMessage.getBody(); + + if (InputStreamResource.class.equals(clazz)) { + InputStream is = DataBufferUtils.toInputStream(body); + return Flux.just(new InputStreamResource(is)); + } + else if (clazz.isAssignableFrom(ByteArrayResource.class)) { + Mono singleBuffer = body.reduce(DataBuffer::write); + return Flux.from(singleBuffer.map(buffer -> { + byte[] bytes = new byte[buffer.readableByteCount()]; + buffer.read(bytes); + return new ByteArrayResource(bytes); + })); + } + else { + return Flux.error(new IllegalStateException( + "Unsupported resource class: " + clazz)); + } + } + + @Override + public Mono write(Publisher inputStream, + ResolvableType type, MediaType contentType, + ReactiveHttpOutputMessage outputMessage) { + + if (inputStream instanceof Mono) { + // single resource + return Mono.from(Flux.from(inputStream). + flatMap(resource -> { + HttpHeaders headers = outputMessage.getHeaders(); + addHeaders(headers, resource, contentType); + + if (resource instanceof HttpRangeResource) { + return writePartialContent((HttpRangeResource) resource, + outputMessage); + } + else { + return writeContent(resource, outputMessage, 0, -1); + } + + + })); + } + else { + // multiple resources, not supported! + return Mono.error(new IllegalArgumentException( + "Multiple resources not yet supported")); + } + } + + protected void addHeaders(HttpHeaders headers, Resource resource, + MediaType contentType) { + if (headers.getContentType() == null) { + if (contentType == null || + !contentType.isConcrete() || + MediaType.APPLICATION_OCTET_STREAM.equals(contentType)) { + contentType = MimeTypeUtils2.getMimeType(resource.getFilename()). + map(MediaTypeUtils::toMediaType). + orElse(MediaType.APPLICATION_OCTET_STREAM); + } + headers.setContentType(contentType); + } + if (headers.getContentLength() < 0) { + contentLength(resource).ifPresent(headers::setContentLength); + } + headers.add(HttpHeaders.ACCEPT_RANGES, "bytes"); + } + + private Mono writeContent(Resource resource, + ReactiveHttpOutputMessage outputMessage, long position, long count) { + if (outputMessage instanceof ZeroCopyHttpOutputMessage) { + Optional file = getFile(resource); + if (file.isPresent()) { + ZeroCopyHttpOutputMessage zeroCopyResponse = + (ZeroCopyHttpOutputMessage) outputMessage; + + if (count < 0) { + count = file.get().length(); + } + + return zeroCopyResponse.setBody(file.get(), position, count); + } + } + + // non-zero copy fallback + try { + InputStream is = resource.getInputStream(); + long skipped = is.skip(position); + if (skipped < position) { + return Mono.error(new IOException( + "Skipped only " + skipped + " bytes out of " + count + + " required.")); + } + + Flux responseBody = + DataBufferUtils.read(is, outputMessage.allocator(), BUFFER_SIZE); + if (count > 0) { + responseBody = DataBufferUtils.takeUntilByteCount(responseBody, count); + } + + return outputMessage.setBody(responseBody); + } + catch (IOException ex) { + return Mono.error(ex); + } + } + + protected Mono writePartialContent(HttpRangeResource resource, + ReactiveHttpOutputMessage outputMessage) { + + // TODO: implement + + return Mono.empty(); + } + + private static Optional contentLength(Resource resource) { + // Don't try to determine contentLength on InputStreamResource - cannot be read afterwards... + // Note: custom InputStreamResource subclasses could provide a pre-calculated content length! + if (InputStreamResource.class != resource.getClass()) { + try { + return Optional.of(resource.contentLength()); + } + catch (IOException ignored) { + } + } + return Optional.empty(); + } + + private static Optional getFile(Resource resource) { + // TODO: introduce Resource.hasFile() property to bypass the potential IOException thrown in Resource.getFile() + // the following Resource implementations do not support getURI/getFile + if (!(resource instanceof ByteArrayResource || + resource instanceof DescriptiveResource || + resource instanceof InputStreamResource)) { + try { + URI resourceUri = resource.getURI(); + if (ResourceUtils.URL_PROTOCOL_FILE.equals(resourceUri.getScheme())) { + return Optional.of(ResourceUtils.getFile(resourceUri)); + } + } + catch (IOException ignored) { + } + } + return Optional.empty(); + } + + +} diff --git a/spring-web-reactive/src/main/java/org/springframework/http/server/reactive/AbstractServerHttpResponse.java b/spring-web-reactive/src/main/java/org/springframework/http/server/reactive/AbstractServerHttpResponse.java index edf2a51a23..22041600e8 100644 --- a/spring-web-reactive/src/main/java/org/springframework/http/server/reactive/AbstractServerHttpResponse.java +++ b/spring-web-reactive/src/main/java/org/springframework/http/server/reactive/AbstractServerHttpResponse.java @@ -94,7 +94,7 @@ public abstract class AbstractServerHttpResponse implements ServerHttpResponse { applyBeforeCommit().after(() -> setBodyInternal(writePublisher))); } - private Mono applyBeforeCommit() { + protected Mono applyBeforeCommit() { Mono mono = Mono.empty(); if (this.state.compareAndSet(STATE_NEW, STATE_COMMITTING)) { for (Supplier> action : this.beforeCommitActions) { diff --git a/spring-web-reactive/src/main/java/org/springframework/http/server/reactive/ReactorServerHttpResponse.java b/spring-web-reactive/src/main/java/org/springframework/http/server/reactive/ReactorServerHttpResponse.java index 8eabb39314..98dbe3f7aa 100644 --- a/spring-web-reactive/src/main/java/org/springframework/http/server/reactive/ReactorServerHttpResponse.java +++ b/spring-web-reactive/src/main/java/org/springframework/http/server/reactive/ReactorServerHttpResponse.java @@ -16,6 +16,8 @@ package org.springframework.http.server.reactive; +import java.io.File; + import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.handler.codec.http.HttpResponseStatus; @@ -31,6 +33,7 @@ import org.springframework.core.io.buffer.DataBufferAllocator; import org.springframework.core.io.buffer.NettyDataBuffer; import org.springframework.http.HttpStatus; import org.springframework.http.ResponseCookie; +import org.springframework.http.ZeroCopyHttpOutputMessage; import org.springframework.util.Assert; /** @@ -39,7 +42,8 @@ import org.springframework.util.Assert; * @author Stephane Maldini * @author Rossen Stoyanchev */ -public class ReactorServerHttpResponse extends AbstractServerHttpResponse { +public class ReactorServerHttpResponse extends AbstractServerHttpResponse + implements ZeroCopyHttpOutputMessage { private final HttpChannel channel; @@ -99,4 +103,11 @@ public class ReactorServerHttpResponse extends AbstractServerHttpResponse { return Unpooled.wrappedBuffer(buffer.asByteBuffer()); } } + + @Override + public Mono setBody(File file, long position, long count) { + return applyBeforeCommit().after(() -> { + return this.channel.sendFile(file, position, count); + }); + } } diff --git a/spring-web-reactive/src/main/java/org/springframework/http/server/reactive/RxNettyServerHttpResponse.java b/spring-web-reactive/src/main/java/org/springframework/http/server/reactive/RxNettyServerHttpResponse.java index a45c2b1b17..f22b794433 100644 --- a/spring-web-reactive/src/main/java/org/springframework/http/server/reactive/RxNettyServerHttpResponse.java +++ b/spring-web-reactive/src/main/java/org/springframework/http/server/reactive/RxNettyServerHttpResponse.java @@ -104,4 +104,35 @@ public class RxNettyServerHttpResponse extends AbstractServerHttpResponse { } } +/* + While the underlying implementation of {@link ZeroCopyHttpOutputMessage} seems to + work; it does bypass {@link #applyBeforeCommit} and more importantly it doesn't change + its {@linkplain #state()). Therefore it's commented out, for now. + + We should revisit this code once + https://github.com/ReactiveX/RxNetty/issues/194 has been fixed. + + + @Override + public Mono setBody(File file, long position, long count) { + Channel channel = this.response.unsafeNettyChannel(); + + HttpResponse httpResponse = + new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK); + io.netty.handler.codec.http.HttpHeaders headers = httpResponse.headers(); + + for (Map.Entry> header : getHeaders().entrySet()) { + String headerName = header.getKey(); + for (String headerValue : header.getValue()) { + headers.add(headerName, headerValue); + } + } + Mono responseWrite = MonoChannelFuture.from(channel.write(httpResponse)); + + FileRegion fileRegion = new DefaultFileRegion(file, position, count); + Mono fileWrite = MonoChannelFuture.from(channel.writeAndFlush(fileRegion)); + + return Flux.concat(applyBeforeCommit(), responseWrite, fileWrite).after(); + } +*/ } \ No newline at end of file diff --git a/spring-web-reactive/src/main/java/org/springframework/http/server/reactive/UndertowHttpHandlerAdapter.java b/spring-web-reactive/src/main/java/org/springframework/http/server/reactive/UndertowHttpHandlerAdapter.java index 8acffe45bb..05cf650a5e 100644 --- a/spring-web-reactive/src/main/java/org/springframework/http/server/reactive/UndertowHttpHandlerAdapter.java +++ b/spring-web-reactive/src/main/java/org/springframework/http/server/reactive/UndertowHttpHandlerAdapter.java @@ -67,9 +67,12 @@ public class UndertowHttpHandlerAdapter implements io.undertow.server.HttpHandle requestBody.registerListener(); ServerHttpRequest request = new UndertowServerHttpRequest(exchange, requestBody); - ResponseBodySubscriber responseBody = new ResponseBodySubscriber(exchange); + StreamSinkChannel responseChannel = exchange.getResponseChannel(); + ResponseBodySubscriber responseBody = + new ResponseBodySubscriber(exchange, responseChannel); responseBody.registerListener(); - ServerHttpResponse response = new UndertowServerHttpResponse(exchange, + ServerHttpResponse response = + new UndertowServerHttpResponse(exchange, responseChannel, publisher -> Mono.from(subscriber -> publisher.subscribe(responseBody)), allocator); @@ -202,9 +205,10 @@ public class UndertowHttpHandlerAdapter implements io.undertow.server.HttpHandle private Subscription subscription; - public ResponseBodySubscriber(HttpServerExchange exchange) { + public ResponseBodySubscriber(HttpServerExchange exchange, + StreamSinkChannel responseChannel) { this.exchange = exchange; - this.responseChannel = exchange.getResponseChannel(); + this.responseChannel = responseChannel; } public void registerListener() { diff --git a/spring-web-reactive/src/main/java/org/springframework/http/server/reactive/UndertowServerHttpResponse.java b/spring-web-reactive/src/main/java/org/springframework/http/server/reactive/UndertowServerHttpResponse.java index 1d3d1598da..67dd0a893c 100644 --- a/spring-web-reactive/src/main/java/org/springframework/http/server/reactive/UndertowServerHttpResponse.java +++ b/spring-web-reactive/src/main/java/org/springframework/http/server/reactive/UndertowServerHttpResponse.java @@ -16,6 +16,10 @@ package org.springframework.http.server.reactive; +import java.io.File; +import java.io.FileInputStream; +import java.io.IOException; +import java.nio.channels.FileChannel; import java.util.List; import java.util.Map; import java.util.function.Function; @@ -25,12 +29,14 @@ import io.undertow.server.handlers.Cookie; import io.undertow.server.handlers.CookieImpl; import io.undertow.util.HttpString; import org.reactivestreams.Publisher; +import org.xnio.channels.StreamSinkChannel; import reactor.core.publisher.Mono; import org.springframework.core.io.buffer.DataBuffer; import org.springframework.core.io.buffer.DataBufferAllocator; import org.springframework.http.HttpStatus; import org.springframework.http.ResponseCookie; +import org.springframework.http.ZeroCopyHttpOutputMessage; import org.springframework.util.Assert; /** @@ -39,19 +45,25 @@ import org.springframework.util.Assert; * @author Marek Hawrylczak * @author Rossen Stoyanchev */ -public class UndertowServerHttpResponse extends AbstractServerHttpResponse { +public class UndertowServerHttpResponse extends AbstractServerHttpResponse + implements ZeroCopyHttpOutputMessage { private final HttpServerExchange exchange; + private final StreamSinkChannel responseChannel; + private final Function, Mono> responseBodyWriter; public UndertowServerHttpResponse(HttpServerExchange exchange, + StreamSinkChannel responseChannel, Function, Mono> responseBodyWriter, DataBufferAllocator allocator) { super(allocator); Assert.notNull(exchange, "'exchange' is required."); + Assert.notNull(responseChannel, "'responseChannel' must not be null"); Assert.notNull(responseBodyWriter, "'responseBodyWriter' must not be null"); this.exchange = exchange; + this.responseChannel = responseChannel; this.responseBodyWriter = responseBodyWriter; } @@ -71,6 +83,26 @@ public class UndertowServerHttpResponse extends AbstractServerHttpResponse { return this.responseBodyWriter.apply(publisher); } + @Override + public Mono setBody(File file, long position, long count) { + writeHeaders(); + writeCookies(); + try { + FileChannel in = new FileInputStream(file).getChannel(); + long result = this.responseChannel.transferFrom(in, position, count); + if (result < count) { + return Mono.error(new IOException("Could only write " + result + + " out of " + count + " bytes")); + } + else { + return Mono.empty(); + } + } + catch (IOException ex) { + return Mono.error(ex); + } + } + @Override protected void writeHeaders() { for (Map.Entry> entry : getHeaders().entrySet()) { diff --git a/spring-web-reactive/src/main/java/org/springframework/web/reactive/accept/PathExtensionContentTypeResolver.java b/spring-web-reactive/src/main/java/org/springframework/web/reactive/accept/PathExtensionContentTypeResolver.java index 57b98d666f..2326797943 100644 --- a/spring-web-reactive/src/main/java/org/springframework/web/reactive/accept/PathExtensionContentTypeResolver.java +++ b/spring-web-reactive/src/main/java/org/springframework/web/reactive/accept/PathExtensionContentTypeResolver.java @@ -13,23 +13,19 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.web.reactive.accept; -import java.io.IOException; -import java.io.InputStream; import java.util.Locale; import java.util.Map; -import javax.activation.FileTypeMap; -import javax.activation.MimetypesFileTypeMap; +import java.util.Optional; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - -import org.springframework.core.io.ClassPathResource; import org.springframework.core.io.Resource; import org.springframework.http.MediaType; +import org.springframework.http.support.MediaTypeUtils; import org.springframework.util.Assert; -import org.springframework.util.ClassUtils; +import org.springframework.util.MimeType; +import org.springframework.util.MimeTypeUtils2; import org.springframework.util.StringUtils; import org.springframework.web.server.NotAcceptableStatusException; import org.springframework.web.server.ServerWebExchange; @@ -48,12 +44,6 @@ import org.springframework.web.util.WebUtils; */ public class PathExtensionContentTypeResolver extends AbstractMappingContentTypeResolver { - private static final Log logger = LogFactory.getLog(PathExtensionContentTypeResolver.class); - - private static final boolean JAF_PRESENT = ClassUtils.isPresent("javax.activation.FileTypeMap", - PathExtensionContentTypeResolver.class.getClassLoader()); - - private boolean useJaf = true; private boolean ignoreUnknownExtensions = true; @@ -103,8 +93,9 @@ public class PathExtensionContentTypeResolver extends AbstractMappingContentType @Override protected MediaType handleNoMatch(String key) throws NotAcceptableStatusException { - if (this.useJaf && JAF_PRESENT) { - MediaType mediaType = JafMediaTypeFactory.getMediaType("file." + key); + if (this.useJaf) { + Optional mimeType = MimeTypeUtils2.getMimeType("file." + key); + MediaType mediaType = mimeType.map(MediaTypeUtils::toMediaType).orElse(null); if (mediaType != null && !MediaType.APPLICATION_OCTET_STREAM.equals(mediaType)) { return mediaType; } @@ -130,8 +121,10 @@ public class PathExtensionContentTypeResolver extends AbstractMappingContentType if (extension != null) { mediaType = getMediaType(extension); } - if (mediaType == null && JAF_PRESENT) { - mediaType = JafMediaTypeFactory.getMediaType(filename); + if (mediaType == null) { + mediaType = + MimeTypeUtils2.getMimeType(filename).map(MediaTypeUtils::toMediaType) + .orElse(null); } if (MediaType.APPLICATION_OCTET_STREAM.equals(mediaType)) { mediaType = null; @@ -139,56 +132,4 @@ public class PathExtensionContentTypeResolver extends AbstractMappingContentType return mediaType; } - - /** - * Inner class to avoid hard-coded dependency on JAF. - */ - private static class JafMediaTypeFactory { - - private static final FileTypeMap fileTypeMap; - - static { - fileTypeMap = initFileTypeMap(); - } - - /** - * Find extended mime.types from the spring-context-support module. - */ - private static FileTypeMap initFileTypeMap() { - Resource resource = new ClassPathResource("org/springframework/mail/javamail/mime.types"); - if (resource.exists()) { - if (logger.isTraceEnabled()) { - logger.trace("Loading JAF FileTypeMap from " + resource); - } - InputStream inputStream = null; - try { - inputStream = resource.getInputStream(); - return new MimetypesFileTypeMap(inputStream); - } - catch (IOException ex) { - // ignore - } - finally { - if (inputStream != null) { - try { - inputStream.close(); - } - catch (IOException ex) { - // ignore - } - } - } - } - if (logger.isTraceEnabled()) { - logger.trace("Loading default Java Activation Framework FileTypeMap"); - } - return FileTypeMap.getDefaultFileTypeMap(); - } - - public static MediaType getMediaType(String filename) { - String mediaType = fileTypeMap.getContentType(filename); - return (StringUtils.hasText(mediaType) ? MediaType.parseMediaType(mediaType) : null); - } - } - } diff --git a/spring-web-reactive/src/main/java/org/springframework/web/reactive/result/method/annotation/RequestBodyArgumentResolver.java b/spring-web-reactive/src/main/java/org/springframework/web/reactive/result/method/annotation/RequestBodyArgumentResolver.java index 216a62cc4c..fca79df098 100644 --- a/spring-web-reactive/src/main/java/org/springframework/web/reactive/result/method/annotation/RequestBodyArgumentResolver.java +++ b/spring-web-reactive/src/main/java/org/springframework/web/reactive/result/method/annotation/RequestBodyArgumentResolver.java @@ -24,10 +24,10 @@ import reactor.core.publisher.Mono; import org.springframework.core.MethodParameter; import org.springframework.core.ResolvableType; -import org.springframework.core.codec.Decoder; import org.springframework.core.convert.ConversionService; import org.springframework.core.io.buffer.DataBuffer; import org.springframework.http.MediaType; +import org.springframework.http.converter.reactive.HttpMessageConverter; import org.springframework.ui.ModelMap; import org.springframework.util.Assert; import org.springframework.web.bind.annotation.RequestBody; @@ -40,15 +40,15 @@ import org.springframework.web.server.ServerWebExchange; */ public class RequestBodyArgumentResolver implements HandlerMethodArgumentResolver { - private final List> decoders; + private final List> messageConverters; private final ConversionService conversionService; - - public RequestBodyArgumentResolver(List> decoders, ConversionService service) { - Assert.notEmpty(decoders, "At least one decoder is required."); + public RequestBodyArgumentResolver(List> messageConverters, + ConversionService service) { + Assert.notEmpty(messageConverters, "At least one message converter is required."); Assert.notNull(service, "'conversionService' is required."); - this.decoders = decoders; + this.messageConverters = messageConverters; this.conversionService = service; } @@ -62,22 +62,29 @@ public class RequestBodyArgumentResolver implements HandlerMethodArgumentResolve public Mono resolveArgument(MethodParameter parameter, ModelMap model, ServerWebExchange exchange) { + ResolvableType type = ResolvableType.forMethodParameter(parameter); + ResolvableType elementType = type.hasGenerics() ? type.getGeneric(0) : type; + MediaType mediaType = exchange.getRequest().getHeaders().getContentType(); if (mediaType == null) { mediaType = MediaType.APPLICATION_OCTET_STREAM; } - ResolvableType type = ResolvableType.forMethodParameter(parameter); - Flux body = exchange.getRequest().getBody(); - Flux elementFlux = body; - ResolvableType elementType = type.hasGenerics() ? type.getGeneric(0) : type; - Decoder decoder = resolveDecoder(elementType, mediaType); - if (decoder != null) { - elementFlux = decoder.decode(body, elementType, mediaType); + Flux body = exchange.getRequest().getBody(); + Flux elementFlux; + + HttpMessageConverter messageConverter = + resolveMessageConverter(elementType, mediaType); + if (messageConverter != null) { + elementFlux = messageConverter.read(elementType, exchange.getRequest()); + } + else { + elementFlux = body; } if (this.conversionService.canConvert(Publisher.class, type.getRawClass())) { - return Mono.just(this.conversionService.convert(elementFlux, type.getRawClass())); + return Mono.just(this.conversionService + .convert(elementFlux, type.getRawClass())); } else if (type.getRawClass() == Flux.class) { return Mono.just(elementFlux); @@ -90,10 +97,11 @@ public class RequestBodyArgumentResolver implements HandlerMethodArgumentResolve return elementFlux.next().map(o -> o); } - private Decoder resolveDecoder(ResolvableType type, MediaType mediaType, Object... hints) { - for (Decoder decoder : this.decoders) { - if (decoder.canDecode(type, mediaType, hints)) { - return decoder; + private HttpMessageConverter resolveMessageConverter(ResolvableType type, + MediaType mediaType) { + for (HttpMessageConverter messageConverter : this.messageConverters) { + if (messageConverter.canRead(type, mediaType)) { + return messageConverter; } } return null; diff --git a/spring-web-reactive/src/main/java/org/springframework/web/reactive/result/method/annotation/RequestMappingHandlerAdapter.java b/spring-web-reactive/src/main/java/org/springframework/web/reactive/result/method/annotation/RequestMappingHandlerAdapter.java index e906bf7a83..d9d2527e62 100644 --- a/spring-web-reactive/src/main/java/org/springframework/web/reactive/result/method/annotation/RequestMappingHandlerAdapter.java +++ b/spring-web-reactive/src/main/java/org/springframework/web/reactive/result/method/annotation/RequestMappingHandlerAdapter.java @@ -17,6 +17,7 @@ package org.springframework.web.reactive.result.method.annotation; import java.lang.reflect.Method; +import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Arrays; import java.util.List; @@ -28,16 +29,19 @@ import org.apache.commons.logging.LogFactory; import reactor.core.publisher.Mono; import org.springframework.beans.factory.InitializingBean; -import org.springframework.core.codec.Decoder; import org.springframework.core.codec.support.ByteBufferDecoder; +import org.springframework.core.codec.support.ByteBufferEncoder; import org.springframework.core.codec.support.JacksonJsonDecoder; +import org.springframework.core.codec.support.JacksonJsonEncoder; import org.springframework.core.codec.support.Jaxb2Decoder; +import org.springframework.core.codec.support.Jaxb2Encoder; import org.springframework.core.codec.support.JsonObjectDecoder; import org.springframework.core.codec.support.StringDecoder; +import org.springframework.core.codec.support.StringEncoder; import org.springframework.core.convert.ConversionService; import org.springframework.core.convert.support.DefaultConversionService; -import org.springframework.core.io.buffer.DataBufferAllocator; -import org.springframework.core.io.buffer.DefaultDataBufferAllocator; +import org.springframework.http.converter.reactive.CodecHttpMessageConverter; +import org.springframework.http.converter.reactive.HttpMessageConverter; import org.springframework.ui.ExtendedModelMap; import org.springframework.ui.ModelMap; import org.springframework.util.ObjectUtils; @@ -62,8 +66,6 @@ public class RequestMappingHandlerAdapter implements HandlerAdapter, Initializin private ConversionService conversionService = new DefaultConversionService(); - private DataBufferAllocator allocator = new DefaultDataBufferAllocator(); - private final Map, ExceptionHandlerMethodResolver> exceptionHandlerCache = new ConcurrentHashMap<>(64); @@ -92,20 +94,23 @@ public class RequestMappingHandlerAdapter implements HandlerAdapter, Initializin return this.conversionService; } - public void setAllocator(DataBufferAllocator allocator) { - this.allocator = allocator; - } - @Override public void afterPropertiesSet() throws Exception { if (ObjectUtils.isEmpty(this.argumentResolvers)) { + List> messageConverters = Arrays.asList( + new CodecHttpMessageConverter(new ByteBufferEncoder(), + new ByteBufferDecoder()), + new CodecHttpMessageConverter(new StringEncoder(), + new StringDecoder()), + new CodecHttpMessageConverter(new Jaxb2Encoder(), + new Jaxb2Decoder()), + new CodecHttpMessageConverter(new JacksonJsonEncoder(), + new JacksonJsonDecoder(new JsonObjectDecoder()))); - List> decoders = Arrays.asList(new ByteBufferDecoder(), - new StringDecoder(), new Jaxb2Decoder(), - new JacksonJsonDecoder(new JsonObjectDecoder())); this.argumentResolvers.add(new RequestParamArgumentResolver()); - this.argumentResolvers.add(new RequestBodyArgumentResolver(decoders, this.conversionService)); + this.argumentResolvers.add(new RequestBodyArgumentResolver(messageConverters, + this.conversionService)); this.argumentResolvers.add(new ModelArgumentResolver()); } } diff --git a/spring-web-reactive/src/main/java/org/springframework/web/reactive/result/method/annotation/ResponseBodyResultHandler.java b/spring-web-reactive/src/main/java/org/springframework/web/reactive/result/method/annotation/ResponseBodyResultHandler.java index c69a315401..5e5c61a4e9 100644 --- a/spring-web-reactive/src/main/java/org/springframework/web/reactive/result/method/annotation/ResponseBodyResultHandler.java +++ b/spring-web-reactive/src/main/java/org/springframework/web/reactive/result/method/annotation/ResponseBodyResultHandler.java @@ -34,10 +34,9 @@ import org.springframework.core.MethodParameter; import org.springframework.core.Ordered; import org.springframework.core.ResolvableType; import org.springframework.core.annotation.AnnotationUtils; -import org.springframework.core.codec.Encoder; import org.springframework.core.convert.ConversionService; -import org.springframework.core.io.buffer.DataBufferAllocator; import org.springframework.http.MediaType; +import org.springframework.http.converter.reactive.HttpMessageConverter; import org.springframework.http.server.reactive.ServerHttpRequest; import org.springframework.http.server.reactive.ServerHttpResponse; import org.springframework.util.Assert; @@ -54,60 +53,51 @@ import org.springframework.web.server.ServerWebExchange; * @author Rossen Stoyanchev * @author Stephane Maldini * @author Sebastien Deleuze + * @author Arjen Poutsma */ public class ResponseBodyResultHandler implements HandlerResultHandler, Ordered { private static final MediaType MEDIA_TYPE_APPLICATION = new MediaType("application"); - private final List> encoders; + private final List> messageConverters; private final ConversionService conversionService; private final List allMediaTypes; - private final Map, List> mediaTypesByEncoder; + private final Map, List> mediaTypesByEncoder; private int order = 0; // TODO: should be MAX_VALUE - - public ResponseBodyResultHandler(List> encoders, ConversionService service) { - Assert.notEmpty(encoders, "At least one encoders is required."); + public ResponseBodyResultHandler(List> messageConverters, + ConversionService service) { + Assert.notEmpty(messageConverters, "At least one message converter is required."); Assert.notNull(service, "'conversionService' is required."); - this.encoders = encoders; + this.messageConverters = messageConverters; this.conversionService = service; - this.allMediaTypes = getAllMediaTypes(encoders); - this.mediaTypesByEncoder = getMediaTypesByEncoder(encoders); + this.allMediaTypes = getAllMediaTypes(messageConverters); + this.mediaTypesByEncoder = getMediaTypesByConverter(messageConverters); } - private static List getAllMediaTypes(List> encoders) { + private static List getAllMediaTypes( + List> messageConverters) { Set set = new LinkedHashSet<>(); - encoders.forEach(encoder -> set.addAll(toMediaTypes(encoder.getSupportedMimeTypes()))); + messageConverters.forEach( + converter -> set.addAll(converter.getWritableMediaTypes())); List result = new ArrayList<>(set); MediaType.sortBySpecificity(result); return Collections.unmodifiableList(result); } - private static Map, List> getMediaTypesByEncoder(List> encoders) { - Map, List> result = new HashMap<>(encoders.size()); - encoders.forEach(encoder -> result.put(encoder, toMediaTypes(encoder.getSupportedMimeTypes()))); + private static Map, List> getMediaTypesByConverter( + List> converters) { + Map, List> result = + new HashMap<>(converters.size()); + converters.forEach(converter -> result + .put(converter, converter.getWritableMediaTypes())); return Collections.unmodifiableMap(result); } - /** - * TODO: MediaType static method - */ - private static List toMediaTypes(List mimeTypes) { - return mimeTypes.stream().map(ResponseBodyResultHandler::toMediaType).collect(Collectors.toList()); - } - - /** - * TODO: MediaType constructor - */ - private static MediaType toMediaType(MimeType mimeType) { - return new MediaType(mimeType.getType(), mimeType.getSubtype(), mimeType.getParameters()); - } - - public void setOrder(int order) { this.order = order; } @@ -154,65 +144,77 @@ public class ResponseBodyResultHandler implements HandlerResultHandler, Ordered elementType = returnType; } - List requestedMediaTypes = getAcceptableMediaTypes(exchange.getRequest()); - List producibleMediaTypes = getProducibleMediaTypes(elementType); - - if (producibleMediaTypes.isEmpty()) { - producibleMediaTypes.add(MediaType.ALL); - } - - Set compatibleMediaTypes = new LinkedHashSet<>(); - for (MediaType requestedType : requestedMediaTypes) { - for (MediaType producibleType : producibleMediaTypes) { - if (requestedType.isCompatibleWith(producibleType)) { - compatibleMediaTypes.add(getMostSpecificMediaType(requestedType, producibleType)); - } - } - } + List compatibleMediaTypes = + getCompatibleMediaTypes(exchange.getRequest(), elementType); if (compatibleMediaTypes.isEmpty()) { - return Mono.error(new NotAcceptableStatusException(producibleMediaTypes)); + return Mono.error(new NotAcceptableStatusException( + getProducibleMediaTypes(elementType))); } - List mediaTypes = new ArrayList<>(compatibleMediaTypes); - MediaType.sortBySpecificityAndQuality(mediaTypes); + Optional selectedMediaType = selectBestMediaType(compatibleMediaTypes); - MediaType selectedMediaType = null; - for (MediaType mediaType : mediaTypes) { - if (mediaType.isConcrete()) { - selectedMediaType = mediaType; - break; - } - else if (mediaType.equals(MediaType.ALL) || mediaType.equals(MEDIA_TYPE_APPLICATION)) { - selectedMediaType = MediaType.APPLICATION_OCTET_STREAM; - break; - } - } - - if (selectedMediaType != null) { - Encoder encoder = resolveEncoder(elementType, selectedMediaType); - if (encoder != null) { + if (selectedMediaType.isPresent()) { + HttpMessageConverter converter = + resolveEncoder(elementType, selectedMediaType.get()); + if (converter != null) { ServerHttpResponse response = exchange.getResponse(); - response.getHeaders().setContentType(selectedMediaType); - DataBufferAllocator allocator = response.allocator(); - return response.setBody( - encoder.encode((Publisher) publisher, allocator, elementType, - selectedMediaType)); + return converter.write((Publisher) publisher, elementType, + selectedMediaType.get(), + response); } } return Mono.error(new NotAcceptableStatusException(this.allMediaTypes)); } + private List getCompatibleMediaTypes(ServerHttpRequest request, + ResolvableType elementType) { + + List acceptableMediaTypes = getAcceptableMediaTypes(request); + List producibleMediaTypes = getProducibleMediaTypes(elementType); + + Set compatibleMediaTypes = new LinkedHashSet<>(); + for (MediaType acceptableMediaType : acceptableMediaTypes) { + compatibleMediaTypes.addAll(producibleMediaTypes.stream(). + filter(acceptableMediaType::isCompatibleWith). + map(producibleType -> getMostSpecificMediaType(acceptableMediaType, + producibleType)).collect(Collectors.toList())); + } + + List result = new ArrayList<>(compatibleMediaTypes); + MediaType.sortBySpecificityAndQuality(result); + return result; + } + private List getAcceptableMediaTypes(ServerHttpRequest request) { List mediaTypes = request.getHeaders().getAccept(); return (mediaTypes.isEmpty() ? Collections.singletonList(MediaType.ALL) : mediaTypes); } + private Optional selectBestMediaType( + List compatibleMediaTypes) { + for (MediaType mediaType : compatibleMediaTypes) { + if (mediaType.isConcrete()) { + return Optional.of(mediaType); + } + else if (mediaType.equals(MediaType.ALL) || + mediaType.equals(MEDIA_TYPE_APPLICATION)) { + return Optional.of(MediaType.APPLICATION_OCTET_STREAM); + } + } + return Optional.empty(); + } + private List getProducibleMediaTypes(ResolvableType type) { - return this.encoders.stream() - .filter(encoder -> encoder.canEncode(type, null)) + List result = this.messageConverters.stream() + .filter(converter -> converter.canWrite(type, null)) .flatMap(encoder -> this.mediaTypesByEncoder.get(encoder).stream()) .collect(Collectors.toList()); + if (result.isEmpty()) { + result.add(MediaType.ALL); + } + + return result; } /** @@ -225,10 +227,11 @@ public class ResponseBodyResultHandler implements HandlerResultHandler, Ordered return (comparator.compare(acceptType, produceType) <= 0 ? acceptType : produceType); } - private Encoder resolveEncoder(ResolvableType type, MediaType mediaType, Object... hints) { - for (Encoder encoder : this.encoders) { - if (encoder.canEncode(type, mediaType, hints)) { - return encoder; + private HttpMessageConverter resolveEncoder(ResolvableType type, + MediaType mediaType) { + for (HttpMessageConverter converter : this.messageConverters) { + if (converter.canWrite(type, mediaType)) { + return converter; } } return null; diff --git a/spring-web-reactive/src/test/java/org/springframework/http/server/reactive/ZeroCopyIntegrationTests.java b/spring-web-reactive/src/test/java/org/springframework/http/server/reactive/ZeroCopyIntegrationTests.java new file mode 100644 index 0000000000..bee5462c1a --- /dev/null +++ b/spring-web-reactive/src/test/java/org/springframework/http/server/reactive/ZeroCopyIntegrationTests.java @@ -0,0 +1,98 @@ +/* + * 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.http.server.reactive; + +import java.io.File; +import java.net.URI; + +import org.junit.Test; +import reactor.core.publisher.Mono; + +import org.springframework.core.io.ClassPathResource; +import org.springframework.core.io.Resource; +import org.springframework.http.MediaType; +import org.springframework.http.RequestEntity; +import org.springframework.http.ResponseEntity; +import org.springframework.http.ZeroCopyHttpOutputMessage; +import org.springframework.http.server.reactive.boot.ReactorHttpServer; +import org.springframework.http.server.reactive.boot.UndertowHttpServer; +import org.springframework.web.client.RestTemplate; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.junit.Assume.assumeTrue; + +/** + * @author Arjen Poutsma + */ +public class ZeroCopyIntegrationTests extends AbstractHttpHandlerIntegrationTests { + + private final ZeroCopyHandler handler = new ZeroCopyHandler(); + + @Override + protected HttpHandler createHttpHandler() { + return handler; + } + + @Test + public void zeroCopy() throws Exception { + // Zero-copy only does not support servlet + assumeTrue(server instanceof ReactorHttpServer || + server instanceof UndertowHttpServer); + + RestTemplate restTemplate = new RestTemplate(); + + RequestEntity request = + RequestEntity.get(new URI("http://localhost:" + port)).build(); + + ResponseEntity response = restTemplate.exchange(request, byte[].class); + + Resource logo = + new ClassPathResource("spring.png", ZeroCopyIntegrationTests.class); + + assertTrue(response.hasBody()); + assertEquals(logo.contentLength(), response.getHeaders().getContentLength()); + assertEquals(logo.contentLength(), response.getBody().length); + assertEquals(MediaType.IMAGE_PNG, response.getHeaders().getContentType()); + + } + + private static class ZeroCopyHandler implements HttpHandler { + + @Override + public Mono handle(ServerHttpRequest request, ServerHttpResponse response) { + try { + ZeroCopyHttpOutputMessage zeroCopyResponse = + (ZeroCopyHttpOutputMessage) response; + + Resource logo = new ClassPathResource("spring.png", + ZeroCopyIntegrationTests.class); + File logoFile = logo.getFile(); + zeroCopyResponse.getHeaders().setContentType(MediaType.IMAGE_PNG); + zeroCopyResponse.getHeaders().setContentLength(logoFile.length()); + return zeroCopyResponse.setBody(logoFile, 0, logoFile.length()); + + } + catch (Throwable ex) { + return Mono.error(ex); + } + + + } + } + +} \ No newline at end of file diff --git a/spring-web-reactive/src/test/java/org/springframework/web/reactive/DispatcherHandlerErrorTests.java b/spring-web-reactive/src/test/java/org/springframework/web/reactive/DispatcherHandlerErrorTests.java index 91c91bc9aa..c7f84a8028 100644 --- a/spring-web-reactive/src/test/java/org/springframework/web/reactive/DispatcherHandlerErrorTests.java +++ b/spring-web-reactive/src/test/java/org/springframework/web/reactive/DispatcherHandlerErrorTests.java @@ -30,7 +30,7 @@ import reactor.core.util.SignalKind; import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; -import org.springframework.core.codec.Encoder; +import org.springframework.core.codec.support.StringDecoder; import org.springframework.core.codec.support.StringEncoder; import org.springframework.core.convert.support.DefaultConversionService; import org.springframework.core.io.buffer.DataBuffer; @@ -38,6 +38,8 @@ import org.springframework.core.io.buffer.DefaultDataBufferAllocator; import org.springframework.http.HttpMethod; import org.springframework.http.HttpStatus; import org.springframework.http.MediaType; +import org.springframework.http.converter.reactive.CodecHttpMessageConverter; +import org.springframework.http.converter.reactive.HttpMessageConverter; import org.springframework.http.server.reactive.MockServerHttpRequest; import org.springframework.http.server.reactive.MockServerHttpResponse; import org.springframework.stereotype.Controller; @@ -230,8 +232,11 @@ public class DispatcherHandlerErrorTests { @Bean public ResponseBodyResultHandler resultHandler() { - List> encoders = Collections.singletonList(new StringEncoder()); - return new ResponseBodyResultHandler(encoders, new DefaultConversionService()); + List> converters = Collections.singletonList( + new CodecHttpMessageConverter<>(new StringEncoder(), + new StringDecoder())); + return new ResponseBodyResultHandler(converters, + new DefaultConversionService()); } @Bean diff --git a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/RequestMappingIntegrationTests.java b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/RequestMappingIntegrationTests.java index 66d595e495..579de341e6 100644 --- a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/RequestMappingIntegrationTests.java +++ b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/RequestMappingIntegrationTests.java @@ -40,14 +40,18 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.core.ParameterizedTypeReference; import org.springframework.core.ResolvableType; -import org.springframework.core.codec.Encoder; +import org.springframework.core.codec.support.ByteBufferDecoder; import org.springframework.core.codec.support.ByteBufferEncoder; +import org.springframework.core.codec.support.JacksonJsonDecoder; import org.springframework.core.codec.support.JacksonJsonEncoder; +import org.springframework.core.codec.support.StringDecoder; import org.springframework.core.codec.support.StringEncoder; import org.springframework.core.convert.ConversionService; import org.springframework.core.convert.support.GenericConversionService; import org.springframework.core.convert.support.ReactiveStreamsToCompletableFutureConverter; import org.springframework.core.convert.support.ReactiveStreamsToRxJava1Converter; +import org.springframework.core.io.ClassPathResource; +import org.springframework.core.io.Resource; import org.springframework.core.io.buffer.DataBuffer; import org.springframework.core.io.buffer.DataBufferAllocator; import org.springframework.core.io.buffer.DefaultDataBufferAllocator; @@ -55,14 +59,19 @@ import org.springframework.http.HttpStatus; import org.springframework.http.MediaType; import org.springframework.http.RequestEntity; import org.springframework.http.ResponseEntity; +import org.springframework.http.converter.reactive.CodecHttpMessageConverter; +import org.springframework.http.converter.reactive.HttpMessageConverter; +import org.springframework.http.converter.reactive.ResourceHttpMessageConverter; import org.springframework.http.server.reactive.AbstractHttpHandlerIntegrationTests; import org.springframework.http.server.reactive.HttpHandler; +import org.springframework.http.server.reactive.ZeroCopyIntegrationTests; import org.springframework.stereotype.Controller; import org.springframework.ui.Model; import org.springframework.web.bind.annotation.ExceptionHandler; import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestParam; +import org.springframework.web.bind.annotation.ResponseBody; import org.springframework.web.bind.annotation.RestController; import org.springframework.web.client.RestTemplate; import org.springframework.web.reactive.DispatcherHandler; @@ -73,8 +82,7 @@ import org.springframework.web.reactive.view.freemarker.FreeMarkerConfigurer; import org.springframework.web.reactive.view.freemarker.FreeMarkerViewResolver; import org.springframework.web.server.adapter.WebHttpHandlerBuilder; -import static org.junit.Assert.assertArrayEquals; -import static org.junit.Assert.assertEquals; +import static org.junit.Assert.*; /** * @author Rossen Stoyanchev @@ -85,6 +93,8 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati private AnnotationConfigApplicationContext wac; + private RestTemplate restTemplate = new RestTemplate(); + @Override protected HttpHandler createHttpHandler() { @@ -100,9 +110,6 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati @Test public void helloWithQueryParam() throws Exception { - - RestTemplate restTemplate = new RestTemplate(); - URI url = new URI("http://localhost:" + port + "/param?name=George"); RequestEntity request = RequestEntity.get(url).build(); ResponseEntity response = restTemplate.exchange(request, String.class); @@ -112,9 +119,6 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati @Test public void rawPojoResponse() throws Exception { - - RestTemplate restTemplate = new RestTemplate(); - URI url = new URI("http://localhost:" + port + "/raw"); RequestEntity request = RequestEntity.get(url).accept(MediaType.APPLICATION_JSON).build(); @@ -125,9 +129,6 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati @Test public void rawFluxResponse() throws Exception { - - RestTemplate restTemplate = new RestTemplate(); - URI url = new URI("http://localhost:" + port + "/raw-flux"); RequestEntity request = RequestEntity.get(url).build(); ResponseEntity response = restTemplate.exchange(request, String.class); @@ -137,9 +138,6 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati @Test public void rawObservableResponse() throws Exception { - - RestTemplate restTemplate = new RestTemplate(); - URI url = new URI("http://localhost:" + port + "/raw-observable"); RequestEntity request = RequestEntity.get(url).build(); ResponseEntity response = restTemplate.exchange(request, String.class); @@ -149,9 +147,6 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati @Test public void handleWithThrownException() throws Exception { - - RestTemplate restTemplate = new RestTemplate(); - URI url = new URI("http://localhost:" + port + "/thrown-exception"); RequestEntity request = RequestEntity.get(url).build(); ResponseEntity response = restTemplate.exchange(request, String.class); @@ -161,9 +156,6 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati @Test public void handleWithErrorSignal() throws Exception { - - RestTemplate restTemplate = new RestTemplate(); - URI url = new URI("http://localhost:" + port + "/error-signal"); RequestEntity request = RequestEntity.get(url).build(); ResponseEntity response = restTemplate.exchange(request, String.class); @@ -174,8 +166,6 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati @Test @Ignore public void streamResult() throws Exception { - RestTemplate restTemplate = new RestTemplate(); - URI url = new URI("http://localhost:" + port + "/stream-result"); RequestEntity request = RequestEntity.get(url).build(); ResponseEntity response = restTemplate.exchange(request, String[].class); @@ -295,9 +285,6 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati @Test public void html() throws Exception { - - RestTemplate restTemplate = new RestTemplate(); - URI url = new URI("http://localhost:" + port + "/html?name=Jason"); RequestEntity request = RequestEntity.get(url).accept(MediaType.TEXT_HTML).build(); ResponseEntity response = restTemplate.exchange(request, String.class); @@ -305,9 +292,20 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati assertEquals("Hello: Jason!", response.getBody()); } + @Test + public void resource() throws Exception { + URI url = new URI("http://localhost:" + port + "/resource"); + RequestEntity request = RequestEntity.get(url).build(); + ResponseEntity response = restTemplate.exchange(request, byte[].class); + + assertTrue(response.hasBody()); + assertEquals(951, response.getHeaders().getContentLength()); + assertEquals(951, response.getBody().length); + assertEquals(new MediaType("image", "x-png"), + response.getHeaders().getContentType()); + } private void serializeAsPojo(String requestUrl) throws Exception { - RestTemplate restTemplate = new RestTemplate(); RequestEntity request = RequestEntity.get(new URI(requestUrl)) .accept(MediaType.APPLICATION_JSON) .build(); @@ -317,7 +315,6 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati } private void serializeAsCollection(String requestUrl) throws Exception { - RestTemplate restTemplate = new RestTemplate(); RequestEntity request = RequestEntity.get(new URI(requestUrl)) .accept(MediaType.APPLICATION_JSON) .build(); @@ -331,7 +328,6 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati private void capitalizePojo(String requestUrl) throws Exception { - RestTemplate restTemplate = new RestTemplate(); RequestEntity request = RequestEntity.post(new URI(requestUrl)) .contentType(MediaType.APPLICATION_JSON) .accept(MediaType.APPLICATION_JSON) @@ -342,7 +338,6 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati } private void capitalizeCollection(String requestUrl) throws Exception { - RestTemplate restTemplate = new RestTemplate(); RequestEntity> request = RequestEntity.post(new URI(requestUrl)) .contentType(MediaType.APPLICATION_JSON) .accept(MediaType.APPLICATION_JSON) @@ -356,7 +351,6 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati } private void createJson(String requestUrl) throws Exception { - RestTemplate restTemplate = new RestTemplate(); URI url = new URI(requestUrl); RequestEntity> request = RequestEntity.post(url) .contentType(MediaType.APPLICATION_JSON) @@ -368,7 +362,6 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati } private void createXml(String requestUrl) throws Exception { - RestTemplate restTemplate = new RestTemplate(); URI url = new URI(requestUrl); People people = new People(); people.getPerson().add(new Person("Robert")); @@ -413,9 +406,16 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati @Bean public ResponseBodyResultHandler responseBodyResultHandler() { - List> encoders = Arrays.asList(new ByteBufferEncoder(), - new StringEncoder(), new JacksonJsonEncoder()); - ResponseBodyResultHandler resultHandler = new ResponseBodyResultHandler(encoders, conversionService()); + List> converters = + Arrays.asList(new ResourceHttpMessageConverter(), + new CodecHttpMessageConverter( + new ByteBufferEncoder(), new ByteBufferDecoder()), + new CodecHttpMessageConverter(new StringEncoder(), + new StringDecoder()), + new CodecHttpMessageConverter( + new JacksonJsonEncoder(), new JacksonJsonDecoder())); + ResponseBodyResultHandler resultHandler = + new ResponseBodyResultHandler(converters, conversionService()); resultHandler.setOrder(1); return resultHandler; } @@ -626,6 +626,12 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati return Mono.just("Recovered from error: " + ex.getMessage()); } + @RequestMapping("/resource") + @ResponseBody + public Resource resource() { + return new ClassPathResource("spring.png", ZeroCopyIntegrationTests.class); + } + //TODO add mixed and T request mappings tests } diff --git a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/ResponseBodyResultHandlerTests.java b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/ResponseBodyResultHandlerTests.java index f290593a8d..38106e7465 100644 --- a/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/ResponseBodyResultHandlerTests.java +++ b/spring-web-reactive/src/test/java/org/springframework/web/reactive/result/method/annotation/ResponseBodyResultHandlerTests.java @@ -24,6 +24,7 @@ import org.reactivestreams.Publisher; import org.springframework.core.ResolvableType; import org.springframework.core.codec.support.StringEncoder; import org.springframework.core.convert.support.DefaultConversionService; +import org.springframework.http.converter.reactive.CodecHttpMessageConverter; import org.springframework.ui.ExtendedModelMap; import org.springframework.web.bind.annotation.ResponseBody; import org.springframework.web.method.HandlerMethod; @@ -41,7 +42,7 @@ public class ResponseBodyResultHandlerTests { @Test public void supports() throws NoSuchMethodException { ResponseBodyResultHandler handler = new ResponseBodyResultHandler(Collections.singletonList( - new StringEncoder()), + new CodecHttpMessageConverter(new StringEncoder(), null)), new DefaultConversionService()); TestController controller = new TestController(); diff --git a/spring-web-reactive/src/test/resources/org/springframework/http/server/reactive/spring.png b/spring-web-reactive/src/test/resources/org/springframework/http/server/reactive/spring.png new file mode 100644 index 0000000000..2fec781a5e Binary files /dev/null and b/spring-web-reactive/src/test/resources/org/springframework/http/server/reactive/spring.png differ