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

WebFlux multpart support polish + minor refactoring

Browse Source
This commit is contained in:
Rossen Stoyanchev
2017-05-01 18:00:08 -04:00
parent ea85431ac5
commit 2390748fd7
11 changed files with 353 additions and 399 deletions
Download Patch File Download Diff File Expand all files Collapse all files

58
spring-web/src/main/java/org/springframework/http/codec/multipart/MultipartHttpMessageReader.java
View File

@@ -1,58 +0,0 @@
/*
* Copyright 2002-2017 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.codec.multipart;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import reactor.core.publisher.Flux;
import org.springframework.core.ResolvableType;
import org.springframework.http.MediaType;
import org.springframework.http.ReactiveHttpInputMessage;
import org.springframework.http.codec.HttpMessageReader;
import org.springframework.util.MultiValueMap;
/**
* Interface for reading multipart HTML forms with {@code "multipart/form-data"} media
* type in accordance with <a href="https://tools.ietf.org/html/rfc7578">RFC 7578</a>.
*
* @author Sebastien Deleuze
* @since 5.0
*/
public interface MultipartHttpMessageReader extends HttpMessageReader<MultiValueMap<String, Part>> {
ResolvableType MULTIPART_VALUE_TYPE =
ResolvableType.forClassWithGenerics(MultiValueMap.class, String.class, Part.class);
@Override
default List<MediaType> getReadableMediaTypes() {
return Collections.singletonList(MediaType.MULTIPART_FORM_DATA);
}
@Override
default boolean canRead(ResolvableType elementType, MediaType mediaType) {
return MULTIPART_VALUE_TYPE.isAssignableFrom(elementType) &&
(mediaType == null || MediaType.MULTIPART_FORM_DATA.isCompatibleWith(mediaType));
}
@Override
default Flux<MultiValueMap<String, Part>> read(ResolvableType elementType, ReactiveHttpInputMessage message, Map<String, Object> hints) {
return Flux.from(readMono(elementType, message, hints));
}
}

161
spring-web/src/main/java/org/springframework/http/codec/multipart/MultipartHttpMessageWriter.java
View File

@@ -32,7 +32,6 @@ import javax.mail.internet.MimeUtility;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.util.function.Tuples;
import org.springframework.core.ResolvableType;
import org.springframework.core.codec.CharSequenceEncoder;
@@ -53,16 +52,14 @@ import org.springframework.util.MimeTypeUtils;
import org.springframework.util.MultiValueMap;
/**
* Implementation of {@link HttpMessageWriter} to write multipart HTML
* forms with {@code "multipart/form-data"} media type.
* {@code HttpMessageWriter} for {@code "multipart/form-data"} requests.
*
* <p>When writing multipart data, this writer uses other
* {@link HttpMessageWriter HttpMessageWriters} to write the respective
* MIME parts. By default, basic writers are registered (for {@code Strings}
* and {@code Resources}). These can be overridden through the provided
* constructors.
* <p>This writer delegates to other message writers to write the respective
* parts. By default basic writers are registered for {@code String}, and
* {@code Resources}. These can be overridden through the provided constructors.
*
* @author Sebastien Deleuze
* @author Rossen Stoyanchev
* @since 5.0
*/
public class MultipartHttpMessageWriter implements HttpMessageWriter<MultiValueMap<String, ?>> {
@@ -70,7 +67,7 @@ public class MultipartHttpMessageWriter implements HttpMessageWriter<MultiValueM
public static final Charset DEFAULT_CHARSET = StandardCharsets.UTF_8;
private List<HttpMessageWriter<?>> partWriters;
private final List<HttpMessageWriter<?>> partWriters;
private Charset filenameCharset = DEFAULT_CHARSET;
@@ -93,9 +90,9 @@ public class MultipartHttpMessageWriter implements HttpMessageWriter<MultiValueM
this(partWriters, new DefaultDataBufferFactory());
}
public MultipartHttpMessageWriter(List<HttpMessageWriter<?>> partWriters, DataBufferFactory bufferFactory) {
public MultipartHttpMessageWriter(List<HttpMessageWriter<?>> partWriters, DataBufferFactory factory) {
this.partWriters = partWriters;
this.bufferFactory = bufferFactory;
this.bufferFactory = factory;
}
/**
@@ -115,10 +112,15 @@ public class MultipartHttpMessageWriter implements HttpMessageWriter<MultiValueM
}
@Override
public List<MediaType> getWritableMediaTypes() {
return Collections.singletonList(MediaType.MULTIPART_FORM_DATA);
}
@Override
public boolean canWrite(ResolvableType elementType, MediaType mediaType) {
return (mediaType == null || MediaType.MULTIPART_FORM_DATA.isCompatibleWith(mediaType)) &&
(MultiValueMap.class.isAssignableFrom(elementType.getRawClass()) && String.class.isAssignableFrom(elementType.resolveGeneric(0)));
return MultiValueMap.class.isAssignableFrom(elementType.getRawClass()) &&
(mediaType == null || MediaType.MULTIPART_FORM_DATA.isCompatibleWith(mediaType));
}
@Override
@@ -126,45 +128,61 @@ public class MultipartHttpMessageWriter implements HttpMessageWriter<MultiValueM
ResolvableType elementType, MediaType mediaType, ReactiveHttpOutputMessage outputMessage,
Map<String, Object> hints) {
final byte[] boundary = generateMultipartBoundary();
Map<String, String> parameters = Collections.singletonMap("boundary", new String(boundary, StandardCharsets.US_ASCII));
byte[] boundary = generateMultipartBoundary();
MediaType contentType = new MediaType(MediaType.MULTIPART_FORM_DATA, parameters);
HttpHeaders headers = outputMessage.getHeaders();
headers.setContentType(contentType);
headers.setContentType(new MediaType(MediaType.MULTIPART_FORM_DATA,
Collections.singletonMap("boundary", new String(boundary, StandardCharsets.US_ASCII))));
return Flux
.from(inputStream)
.single()
.flatMap(form -> {
Flux<DataBuffer> body = Flux.fromIterable(form.entrySet())
.concatMap(entry -> Flux.fromIterable(entry.getValue()).map(value -> Tuples.of(entry.getKey(), value)))
.concatMap(part -> generatePart(part.getT1(), getHttpEntity(part.getT2()), boundary))
.concatWith(Mono.just(generateLastLine(boundary)));
return outputMessage.writeWith(body);
});
return Mono.from(inputStream).flatMap(multiValueMap ->
outputMessage.writeWith(generateParts(multiValueMap, boundary)));
}
/**
* Generate a multipart boundary.
* <p>By default delegates to {@link MimeTypeUtils#generateMultipartBoundary()}.
*/
protected byte[] generateMultipartBoundary() {
return MimeTypeUtils.generateMultipartBoundary();
}
private Flux<DataBuffer> generateParts(MultiValueMap<String, ?> map, byte[] boundary) {
return Flux.fromIterable(map.entrySet())
.concatMap(entry -> Flux
.fromIterable(entry.getValue())
.concatMap(value -> generatePart(entry.getKey(), value, boundary)))
.concatWith(Mono.just(generateLastLine(boundary)));
}
@SuppressWarnings("unchecked")
private Flux<DataBuffer> generatePart(String name, HttpEntity<?> partEntity, byte[] boundary) {
Object partBody = partEntity.getBody();
ResolvableType partType = ResolvableType.forClass(partBody.getClass());
MultipartHttpOutputMessage outputMessage = new MultipartHttpOutputMessage(this.bufferFactory);
HttpHeaders partHeaders = outputMessage.getHeaders();
outputMessage.getHeaders().putAll(partHeaders);
MediaType partContentType = partHeaders.getContentType();
partHeaders.setContentDispositionFormData(name, getFilename(partBody));
private <T> Flux<DataBuffer> generatePart(String name, T value, byte[] boundary) {
Optional<HttpMessageWriter<?>> writer = this.partWriters
.stream()
.filter(e -> e.canWrite(partType, partContentType))
MultipartHttpOutputMessage outputMessage = new MultipartHttpOutputMessage(this.bufferFactory);
T body;
if (value instanceof HttpEntity) {
outputMessage.getHeaders().putAll(((HttpEntity<T>) value).getHeaders());
body = ((HttpEntity<T>) value).getBody();
}
else {
body = value;
}
ResolvableType bodyType = ResolvableType.forClass(body.getClass());
outputMessage.getHeaders().setContentDispositionFormData(name, getFilename(body));
MediaType contentType = outputMessage.getHeaders().getContentType();
Optional<HttpMessageWriter<?>> writer = this.partWriters.stream()
.filter(partWriter -> partWriter.canWrite(bodyType, contentType))
.findFirst();
if(!writer.isPresent()) {
return Flux.error(new CodecException("No suitable writer found!"));
return Flux.error(new CodecException("No suitable writer found for part: " + name));
}
Mono<Void> partWritten = ((HttpMessageWriter<Object>)writer.get())
.write(Mono.just(partBody), partType, partContentType, outputMessage, Collections.emptyMap());
Mono<Void> partWritten = ((HttpMessageWriter<T>) writer.get())
.write(Mono.just(body), bodyType, contentType, outputMessage, Collections.emptyMap());
// partWritten.subscribe() is required in order to make sure MultipartHttpOutputMessage#getBody()
// returns a non-null value (occurs with ResourceHttpMessageWriter that invokes
@@ -180,36 +198,12 @@ public class MultipartHttpMessageWriter implements HttpMessageWriter<MultiValueM
}
/**
* Generate a multipart boundary.
* <p>This implementation delegates to
* {@link MimeTypeUtils#generateMultipartBoundary()}.
*/
protected byte[] generateMultipartBoundary() {
return MimeTypeUtils.generateMultipartBoundary();
}
/**
* Return an {@link HttpEntity} for the given part Object.
* @param part the part to return an {@link HttpEntity} for
* @return the part Object itself it is an {@link HttpEntity},
* or a newly built {@link HttpEntity} wrapper for that part
*/
protected HttpEntity<?> getHttpEntity(Object part) {
if (part instanceof HttpEntity) {
return (HttpEntity<?>) part;
}
else {
return new HttpEntity<>(part);
}
}
/**
* Return the filename of the given multipart part. This value will be used for the
* {@code Content-Disposition} header.
* <p>The default implementation returns {@link Resource#getFilename()} if the part is a
* {@code Resource}, and {@code null} in other cases. Can be overridden in subclasses.
* @param part the part to determine the file name for
* @return the filename, or {@code null} if not known
* Return the filename of the given multipart part. This value will be used
* for the {@code Content-Disposition} header.
* <p>The default implementation returns {@link Resource#getFilename()} if
* the part is a {@code Resource}, and {@code null} in other cases.
* @param part the part for which return a file name
* @return the filename or {@code null}
*/
protected String getFilename(Object part) {
if (part instanceof Resource) {
@@ -223,7 +217,6 @@ public class MultipartHttpMessageWriter implements HttpMessageWriter<MultiValueM
}
}
private DataBuffer generateBoundaryLine(byte[] boundary) {
DataBuffer buffer = this.bufferFactory.allocateBuffer(boundary.length + 4);
buffer.write((byte)'-');
@@ -253,11 +246,6 @@ public class MultipartHttpMessageWriter implements HttpMessageWriter<MultiValueM
return buffer;
}
@Override
public List<MediaType> getWritableMediaTypes() {
return Collections.singletonList(MediaType.MULTIPART_FORM_DATA);
}
private static class MultipartHttpOutputMessage implements ReactiveHttpOutputMessage {
@@ -304,11 +292,6 @@ public class MultipartHttpMessageWriter implements HttpMessageWriter<MultiValueM
return this.body.then();
}
@Override
public Mono<Void> writeAndFlushWith(Publisher<? extends Publisher<? extends DataBuffer>> body) {
return Mono.error(new UnsupportedOperationException());
}
private DataBuffer generateHeaders() {
DataBuffer buffer = this.bufferFactory.allocateBuffer();
for (Map.Entry<String, List<String>> entry : headers.entrySet()) {
@@ -329,13 +312,21 @@ public class MultipartHttpMessageWriter implements HttpMessageWriter<MultiValueM
}
@Override
public Mono<Void> setComplete() {
return (this.body != null ? this.body.then() : Mono.error(new IllegalStateException("Body has not been written yet")));
public Mono<Void> writeAndFlushWith(Publisher<? extends Publisher<? extends DataBuffer>> body) {
return Mono.error(new UnsupportedOperationException());
}
public Flux<DataBuffer> getBody() {
return (this.body != null ? this.body : Flux.error(new IllegalStateException("Body has not been written yet")));
return (this.body != null ? this.body :
Flux.error(new IllegalStateException("Body has not been written yet")));
}
@Override
public Mono<Void> setComplete() {
return (this.body != null ? this.body.then() :
Mono.error(new IllegalStateException("Body has not been written yet")));
}
}
/**

43
spring-web/src/main/java/org/springframework/http/codec/multipart/Part.java
View File

@@ -26,45 +26,56 @@ import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.http.HttpHeaders;
/**
* A representation of a part received in a multipart request. Could contain a file, the
* string or json value of a parameter.
* Representation for a part in a "multipart/form-data" request.
*
* <p>The origin of a multipart request may a browser form in which case each
* part represents a text-based form field or a file upload. Multipart requests
* may also be used outside of browsers to transfer data with any content type
* such as JSON, PDF, etc.
*
* @author Sebastien Deleuze
* @author Rossen Stoyanchev
* @since 5.0
* @see <a href="https://tools.ietf.org/html/rfc7578">RFC 7578 (multipart/form-data)</a>
* @see <a href="https://tools.ietf.org/html/rfc2183">RFC 2183 (Content-Disposition)</a>
* @see <a href="https://www.w3.org/TR/html5/forms.html#multipart-form-data">HTML5 (multipart forms)</a>
*/
public interface Part {
/**
* @return the headers of this part
*/
HttpHeaders getHeaders();
/**
* @return the name of the parameter in the multipart form
* Return the name of the part in the multipart form.
* @return the name of the part, never {@code null} or empty
*/
String getName();
/**
* @return optionally the filename if the part contains a file
* Return the headers associated with the part.
*/
HttpHeaders getHeaders();
/**
*
* Return the name of the file selected by the user in a browser form.
* @return the filename if defined and available
*/
Optional<String> getFilename();
/**
* @return the content of the part as a String using the charset specified in the
* {@code Content-Type} header if any, or else using {@code UTF-8} by default.
* Return the part content converted to a String with the charset from the
* {@code Content-Type} header or {@code UTF-8} by default.
*/
Mono<String> getContentAsString();
/**
* @return the content of the part as a stream of bytes
* Return the part raw content as a stream of DataBuffer's.
*/
Flux<DataBuffer> getContent();
/**
* Transfer the file contained in this part to the specified destination.
* @param dest the destination file
* @return a {@link Mono} that indicates completion of the file transfer or an error,
* for example an {@link IllegalStateException} if the part does not contain a file
* Transfer the file in this part to the given file destination.
* @param dest the target file
* @return completion {@code Mono} with the result of the file transfer,
* possibly {@link IllegalStateException} if the part isn't a file
*/
Mono<Void> transferTo(File dest);

261
spring-web/src/main/java/org/springframework/http/codec/multipart/SynchronossMultipartHttpMessageReader.java
View File

@@ -26,6 +26,8 @@ import java.nio.channels.ReadableByteChannel;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -52,74 +54,102 @@ import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.ReactiveHttpInputMessage;
import org.springframework.http.codec.HttpMessageReader;
import org.springframework.util.Assert;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MimeType;
import org.springframework.util.MultiValueMap;
import org.springframework.util.StreamUtils;
/**
* Implementation of {@link HttpMessageReader} to read multipart HTML
* forms with {@code "multipart/form-data"} media type in accordance
* with <a href="https://tools.ietf.org/html/rfc7578">RFC 7578</a> based
* {@code HttpMessageReader} for {@code "multipart/form-data"} requests based
* on the Synchronoss NIO Multipart library.
*
* @author Sebastien Deleuze
* @author Rossen Stoyanchev
* @since 5.0
* @see <a href="https://github.com/synchronoss/nio-multipart">Synchronoss NIO Multipart</a>
*/
public class SynchronossMultipartHttpMessageReader implements MultipartHttpMessageReader {
public class SynchronossMultipartHttpMessageReader implements HttpMessageReader<MultiValueMap<String, Part>> {
private static final ResolvableType MULTIPART_VALUE_TYPE = ResolvableType.forClassWithGenerics(
MultiValueMap.class, String.class, Part.class);
@Override
public Mono<MultiValueMap<String, Part>> readMono(ResolvableType elementType, ReactiveHttpInputMessage inputMessage, Map<String, Object> hints) {
public List<MediaType> getReadableMediaTypes() {
return Collections.singletonList(MediaType.MULTIPART_FORM_DATA);
}
return Flux.create(new NioMultipartConsumer(inputMessage))
.collectMultimap(part -> part.getName())
.map(partsMap -> new LinkedMultiValueMap<>(partsMap
.entrySet()
.stream()
.collect(Collectors.toMap(
entry -> entry.getKey(),
entry -> new ArrayList<>(entry.getValue()))
)));
@Override
public boolean canRead(ResolvableType elementType, MediaType mediaType) {
return MULTIPART_VALUE_TYPE.isAssignableFrom(elementType) &&
(mediaType == null || MediaType.MULTIPART_FORM_DATA.isCompatibleWith(mediaType));
}
private static class NioMultipartConsumer implements Consumer<FluxSink<Part>> {
@Override
public Flux<MultiValueMap<String, Part>> read(ResolvableType elementType,
ReactiveHttpInputMessage message, Map<String, Object> hints) {
return Flux.from(readMono(elementType, message, hints));
}
@Override
public Mono<MultiValueMap<String, Part>> readMono(ResolvableType elementType,
ReactiveHttpInputMessage inputMessage, Map<String, Object> hints) {
return Flux.create(new SynchronossPartGenerator(inputMessage))
.collectMultimap(Part::getName).map(this::toMultiValueMap);
}
private LinkedMultiValueMap<String, Part> toMultiValueMap(Map<String, Collection<Part>> map) {
return new LinkedMultiValueMap<>(map.entrySet().stream()
.collect(Collectors.toMap(Map.Entry::getKey, e -> toList(e.getValue()))));
}
private List<Part> toList(Collection<Part> collection) {
return collection instanceof List ? (List<Part>) collection : new ArrayList<>(collection);
}
/**
* Consume and feed input to the Synchronoss parser, then adapt parser
* output events to {@code Flux<Sink<Part>>}.
*/
private static class SynchronossPartGenerator implements Consumer<FluxSink<Part>> {
private final ReactiveHttpInputMessage inputMessage;
public NioMultipartConsumer(ReactiveHttpInputMessage inputMessage) {
SynchronossPartGenerator(ReactiveHttpInputMessage inputMessage) {
this.inputMessage = inputMessage;
}
@Override
public void accept(FluxSink<Part> emitter) {
HttpHeaders headers = inputMessage.getHeaders();
MultipartContext context = new MultipartContext(
headers.getContentType().toString(),
Math.toIntExact(headers.getContentLength()),
headers.getFirst(HttpHeaders.ACCEPT_CHARSET));
NioMultipartParserListener listener = new ReactiveNioMultipartParserListener(emitter);
MultipartContext context = createMultipartContext();
NioMultipartParserListener listener = new FluxSinkAdapterListener(emitter);
NioMultipartParser parser = Multipart.multipart(context).forNIO(listener);
inputMessage.getBody().subscribe(buffer -> {
this.inputMessage.getBody().subscribe(buffer -> {
byte[] resultBytes = new byte[buffer.readableByteCount()];
buffer.read(resultBytes);
try {
parser.write(resultBytes);
}
catch (IOException ex) {
listener.onError("Exception thrown while closing the parser", ex);
listener.onError("Exception thrown providing input to the parser", ex);
}
}, (e) -> {
}, (ex) -> {
try {
listener.onError("Exception thrown while reading the request body", e);
listener.onError("Request body input error", ex);
parser.close();
}
catch (IOException ex) {
listener.onError("Exception thrown while closing the parser", ex);
catch (IOException ex2) {
listener.onError("Exception thrown while closing the parser", ex2);
}
}, () -> {
try {
@@ -132,85 +162,100 @@ public class SynchronossMultipartHttpMessageReader implements MultipartHttpMessa
}
private static class ReactiveNioMultipartParserListener implements NioMultipartParserListener {
private FluxSink<Part> emitter;
private final AtomicInteger errorCount = new AtomicInteger(0);
private MultipartContext createMultipartContext() {
HttpHeaders headers = this.inputMessage.getHeaders();
String contentType = headers.getContentType().toString();
int contentLength = Math.toIntExact(headers.getContentLength());
String charset = headers.getFirst(HttpHeaders.ACCEPT_CHARSET);
return new MultipartContext(contentType, contentLength, charset);
}
public ReactiveNioMultipartParserListener(FluxSink<Part> emitter) {
this.emitter = emitter;
}
/**
* Listen for parser output and adapt to {@code Flux<Sink<Part>>}.
*/
private static class FluxSinkAdapterListener implements NioMultipartParserListener {
private final FluxSink<Part> sink;
private final AtomicInteger terminated = new AtomicInteger(0);
FluxSinkAdapterListener(FluxSink<Part> sink) {
this.sink = sink;
}
@Override
public void onPartFinished(StreamStorage storage, Map<String, List<String>> headers) {
HttpHeaders httpHeaders = new HttpHeaders();
httpHeaders.putAll(headers);
this.sink.next(new SynchronossPart(httpHeaders, storage));
}
@Override
public void onFormFieldPartFinished(String name, String value, Map<String, List<String>> headers) {
HttpHeaders httpHeaders = new HttpHeaders();
httpHeaders.putAll(headers);
this.sink.next(new SynchronossPart(httpHeaders, value));
}
@Override
public void onError(String message, Throwable cause) {
if (this.terminated.getAndIncrement() == 0) {
this.sink.error(new RuntimeException(message, cause));
}
}
@Override
public void onPartFinished(StreamStorage streamStorage, Map<String, List<String>> headersFromPart) {
HttpHeaders headers = new HttpHeaders();
headers.putAll(headersFromPart);
emitter.next(new NioPart(headers, streamStorage));
@Override
public void onAllPartsFinished() {
if (this.terminated.getAndIncrement() == 0) {
this.sink.complete();
}
}
@Override
public void onFormFieldPartFinished(String fieldName, String fieldValue, Map<String, List<String>> headersFromPart) {
HttpHeaders headers = new HttpHeaders();
headers.putAll(headersFromPart);
emitter.next(new NioPart(headers, fieldValue));
}
@Override
public void onAllPartsFinished() {
emitter.complete();
}
@Override
public void onNestedPartStarted(Map<String, List<String>> headersFromParentPart) {
}
@Override
public void onNestedPartFinished() {
}
@Override
public void onError(String message, Throwable cause) {
if (errorCount.getAndIncrement() == 1) {
emitter.error(new RuntimeException(message, cause));
}
}
@Override
public void onNestedPartStarted(Map<String, List<String>> headersFromParentPart) {
}
@Override
public void onNestedPartFinished() {
}
}
/**
* {@link Part} implementation based on the NIO Multipart library.
*/
private static class NioPart implements Part {
private static class SynchronossPart implements Part {
private final HttpHeaders headers;
private final StreamStorage streamStorage;
private final StreamStorage storage;
private final String content;
private final DataBufferFactory bufferFactory = new DefaultDataBufferFactory();
public NioPart(HttpHeaders headers, StreamStorage streamStorage) {
SynchronossPart(HttpHeaders headers, StreamStorage storage) {
Assert.notNull(headers, "HttpHeaders is required");
Assert.notNull(storage, "'storage' is required");
this.headers = headers;
this.streamStorage = streamStorage;
this.storage = storage;
this.content = null;
}
public NioPart(HttpHeaders headers, String content) {
SynchronossPart(HttpHeaders headers, String content) {
Assert.notNull(headers, "HttpHeaders is required");
Assert.notNull(content, "'content' is required");
this.headers = headers;
this.streamStorage = null;
this.storage = null;
this.content = content;
}
@Override
public String getName() {
return MultipartUtils.getFieldName(headers);
return MultipartUtils.getFieldName(this.headers);
}
@Override
@@ -223,45 +268,27 @@ public class SynchronossMultipartHttpMessageReader implements MultipartHttpMessa
return Optional.ofNullable(MultipartUtils.getFileName(this.headers));
}
@Override
public Mono<Void> transferTo(File dest) {
if (!getFilename().isPresent()) {
return Mono.error(new IllegalStateException("The part does not contain a file."));
}
try {
InputStream inputStream = this.streamStorage.getInputStream();
// Get a FileChannel when possible in order to use zero copy mechanism
ReadableByteChannel inChannel = Channels.newChannel(inputStream);
FileChannel outChannel = new FileOutputStream(dest).getChannel();
// NIO Multipart has previously limited the size of the content
long count = (inChannel instanceof FileChannel ? ((FileChannel)inChannel).size() : Long.MAX_VALUE);
long result = outChannel.transferFrom(inChannel, 0, count);
if (result < count) {
return Mono.error(new IOException(
"Could only write " + result + " out of " + count + " bytes"));
}
}
catch (IOException ex) {
return Mono.error(ex);
}
return Mono.empty();
}
@Override
public Mono<String> getContentAsString() {
if (this.content != null) {
return Mono.just(this.content);
}
MediaType contentType = this.headers.getContentType();
Charset charset = (contentType.getCharset() == null ? StandardCharsets.UTF_8 : contentType.getCharset());
try {
return Mono.just(StreamUtils.copyToString(this.streamStorage.getInputStream(), charset));
InputStream inputStream = this.storage.getInputStream();
Charset charset = getCharset();
return Mono.just(StreamUtils.copyToString(inputStream, charset));
}
catch (IOException e) {
return Mono.error(new IllegalStateException("Error while reading part content as a string", e));
return Mono.error(new IllegalStateException(
"Error while reading part content as a string", e));
}
}
private Charset getCharset() {
return Optional.ofNullable(this.headers.getContentType())
.map(MimeType::getCharset).orElse(StandardCharsets.UTF_8);
}
@Override
public Flux<DataBuffer> getContent() {
if (this.content != null) {
@@ -269,9 +296,29 @@ public class SynchronossMultipartHttpMessageReader implements MultipartHttpMessa
buffer.write(this.content.getBytes());
return Flux.just(buffer);
}
InputStream inputStream = this.streamStorage.getInputStream();
InputStream inputStream = this.storage.getInputStream();
return DataBufferUtils.read(inputStream, this.bufferFactory, 4096);
}
@Override
public Mono<Void> transferTo(File dest) {
if (this.storage == null || !getFilename().isPresent()) {
return Mono.error(new IllegalStateException("The part does not represent a file."));
}
try {
ReadableByteChannel ch = Channels.newChannel(this.storage.getInputStream());
long expected = (ch instanceof FileChannel ? ((FileChannel) ch).size() : Long.MAX_VALUE);
long actual = new FileOutputStream(dest).getChannel().transferFrom(ch, 0, expected);
if (actual < expected) {
return Mono.error(new IOException(
"Could only write " + actual + " out of " + expected + " bytes"));
}
}
catch (IOException ex) {
return Mono.error(ex);
}
return Mono.empty();
}
}
}

8
spring-web/src/main/java/org/springframework/web/server/ServerWebExchange.java
View File

@@ -79,13 +79,15 @@ public interface ServerWebExchange {
/**
* Return the form data from the body of the request if the Content-Type is
* {@code "application/x-www-form-urlencoded"} or an empty map.
* {@code "application/x-www-form-urlencoded"} or an empty map otherwise.
* This method may be called multiple times.
*/
Mono<MultiValueMap<String, String>> getFormData();
/**
* Return the form parts from the body of the request or an empty {@code Mono}
* if the Content-Type is not "multipart/form-data".
* Return the parts of a multipart request if the Content-Type is
* {@code "multipart/form-data"} or an empty map otherwise.
* This method may be called multiple times.
*/
Mono<MultiValueMap<String, Part>> getMultipartData();

19
spring-web/src/main/java/org/springframework/web/server/adapter/DefaultServerWebExchange.java
View File

@@ -48,8 +48,8 @@ import org.springframework.web.server.ServerWebExchange;
import org.springframework.web.server.WebSession;
import org.springframework.web.server.session.WebSessionManager;
import static org.springframework.http.MediaType.*;
import static org.springframework.http.codec.multipart.MultipartHttpMessageReader.*;
import static org.springframework.http.MediaType.APPLICATION_FORM_URLENCODED;
import static org.springframework.http.MediaType.MULTIPART_FORM_DATA;
/**
* Default implementation of {@link ServerWebExchange}.
@@ -64,6 +64,9 @@ public class DefaultServerWebExchange implements ServerWebExchange {
private static final ResolvableType FORM_DATA_VALUE_TYPE =
ResolvableType.forClassWithGenerics(MultiValueMap.class, String.class, String.class);
private static final ResolvableType MULTIPART_VALUE_TYPE = ResolvableType.forClassWithGenerics(
MultiValueMap.class, String.class, Part.class);
private static final Mono<MultiValueMap<String, String>> EMPTY_FORM_DATA =
Mono.just(CollectionUtils.unmodifiableMultiValueMap(new LinkedMultiValueMap<String, String>(0)))
.cache();
@@ -121,9 +124,10 @@ public class DefaultServerWebExchange implements ServerWebExchange {
return ((HttpMessageReader<MultiValueMap<String, String>>)codecConfigurer
.getReaders()
.stream()
.filter(messageReader -> messageReader.canRead(FORM_DATA_VALUE_TYPE, APPLICATION_FORM_URLENCODED))
.filter(reader -> reader.canRead(FORM_DATA_VALUE_TYPE, APPLICATION_FORM_URLENCODED))
.findFirst()
.orElseThrow(() -> new IllegalStateException("Could not find HttpMessageReader that supports " + APPLICATION_FORM_URLENCODED)))
.orElseThrow(() -> new IllegalStateException(
"Could not find HttpMessageReader that supports " + APPLICATION_FORM_URLENCODED)))
.readMono(FORM_DATA_VALUE_TYPE, request, Collections.emptyMap())
.switchIfEmpty(EMPTY_FORM_DATA)
.cache();
@@ -143,12 +147,13 @@ public class DefaultServerWebExchange implements ServerWebExchange {
try {
contentType = request.getHeaders().getContentType();
if (MULTIPART_FORM_DATA.isCompatibleWith(contentType)) {
return ((HttpMessageReader<MultiValueMap<String, Part>>)codecConfigurer
return ((HttpMessageReader<MultiValueMap<String, Part>>) codecConfigurer
.getReaders()
.stream()
.filter(messageReader -> messageReader.canRead(MULTIPART_VALUE_TYPE, MULTIPART_FORM_DATA))
.filter(reader -> reader.canRead(MULTIPART_VALUE_TYPE, MULTIPART_FORM_DATA))
.findFirst()
.orElseThrow(() -> new IllegalStateException("Could not find HttpMessageReader that supports " + MULTIPART_FORM_DATA)))
.orElseThrow(() -> new IllegalStateException(
"Could not find HttpMessageReader that supports " + MULTIPART_FORM_DATA)))
.readMono(FORM_DATA_VALUE_TYPE, request, Collections.emptyMap())
.switchIfEmpty(EMPTY_MULTIPART_DATA)
.cache();