Support data binding for multipart requests in WebFlux

Issue: SPR-14546

Author: rstoyanchev

spring-web/src/main/java/org/springframework/http/codec/multipart/FilePart.java

@@ -0,0 +1,45 @@
+/*
+ * 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.io.File;
+
+import reactor.core.publisher.Mono;
+
+/**
+ * Specialization of {@link Part} for a file upload.
+ *
+ * @author Rossen Stoyanchev
+ * @since 5.0
+ */
+public interface FilePart extends Part {
+
+	/**
+	 * Return the name of the file selected by the user in a browser form.
+	 */
+	String getFilename();
+
+
+	/**
+	 * 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);
+
+}

spring-web/src/main/java/org/springframework/http/codec/multipart/FormFieldPart.java

@@ -0,0 +1,32 @@
+/*
+ * 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;
+
+/**
+ * Specialization of {@link Part} for a form field.
+ *
+ * @author Rossen Stoyanchev
+ * @since 5.0
+ */
+public interface FormFieldPart extends Part {
+
+	/**
+	 * Return the form field value.
+	 */
+	String getValue();
+
+}

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

@@ -16,11 +16,7 @@
 
 package org.springframework.http.codec.multipart;
 
-import java.io.File;
-import java.util.Optional;
-
 import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
 
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.http.HttpHeaders;
@@ -29,9 +25,10 @@
  * 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.
+ * part is either a {@link FormFieldPart} or a {@link FilePart}.
+ *
+ * <p>Multipart requests may also be used outside of a browser for data of any
+ * content type (e.g. JSON, PDF, etc).
  *
  * @author Sebastien Deleuze
  * @author Rossen Stoyanchev
@@ -53,30 +50,9 @@ public interface 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 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 part raw content as a stream of DataBuffer's.
 	 */
 	Flux<DataBuffer> getContent();
 
-	/**
-	 * Transfer the file in this part to the given file destination.
-	 * @param destination 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 destination);
-
 }

spring-web/src/main/java/org/springframework/http/codec/multipart/SynchronossPartHttpMessageReader.java

@@ -52,8 +52,6 @@
 import org.springframework.http.ReactiveHttpInputMessage;
 import org.springframework.http.codec.HttpMessageReader;
 import org.springframework.util.Assert;
-import org.springframework.util.MimeType;
-import org.springframework.util.StreamUtils;
 
 /**
  * {@code HttpMessageReader} for parsing {@code "multipart/form-data"} requests
@@ -71,6 +69,8 @@
  */
 public class SynchronossPartHttpMessageReader implements HttpMessageReader<Part> {
 
+	private final DataBufferFactory bufferFactory = new DefaultDataBufferFactory();
+
 
 	@Override
 	public List<MediaType> getReadableMediaTypes() {
@@ -88,7 +88,7 @@ public boolean canRead(ResolvableType elementType, MediaType mediaType) {
 	public Flux<Part> read(ResolvableType elementType, ReactiveHttpInputMessage message,
 			Map<String, Object> hints) {
 
-		return Flux.create(new SynchronossPartGenerator(message));
+		return Flux.create(new SynchronossPartGenerator(message, this.bufferFactory));
 	}
 
 
@@ -109,17 +109,20 @@ private static class SynchronossPartGenerator implements Consumer<FluxSink<Part>
 
 		private final ReactiveHttpInputMessage inputMessage;
 
+		private final DataBufferFactory bufferFactory;
+
 
-		SynchronossPartGenerator(ReactiveHttpInputMessage inputMessage) {
+		SynchronossPartGenerator(ReactiveHttpInputMessage inputMessage, DataBufferFactory factory) {
 			this.inputMessage = inputMessage;
+			this.bufferFactory = factory;
 		}
 
 
 		@Override
 		public void accept(FluxSink<Part> emitter) {
 
 			MultipartContext context = createMultipartContext();
-			NioMultipartParserListener listener = new FluxSinkAdapterListener(emitter);
+			NioMultipartParserListener listener = new FluxSinkAdapterListener(emitter, this.bufferFactory);
 			NioMultipartParser parser = Multipart.multipart(context).forNIO(listener);
 
 			this.inputMessage.getBody().subscribe(buffer -> {
@@ -167,26 +170,32 @@ private static class FluxSinkAdapterListener implements NioMultipartParserListen
 
 		private final FluxSink<Part> sink;
 
+		private final DataBufferFactory bufferFactory;
+
 		private final AtomicInteger terminated = new AtomicInteger(0);
 
 
-		FluxSinkAdapterListener(FluxSink<Part> sink) {
+		FluxSinkAdapterListener(FluxSink<Part> sink, DataBufferFactory bufferFactory) {
 			this.sink = sink;
+			this.bufferFactory = bufferFactory;
 		}
 
 
 		@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));
+			Part part = MultipartUtils.getFileName(httpHeaders) != null ?
+					new SynchronossFilePart(httpHeaders, storage, this.bufferFactory) :
+					new DefaultSynchronossPart(httpHeaders, storage, this.bufferFactory);
+			this.sink.next(part);
 		}
 
 		@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));
+			this.sink.next(new SynchronossFormFieldPart(httpHeaders, this.bufferFactory, value));
 		}
 
 		@Override
@@ -213,31 +222,18 @@ public void onNestedPartFinished() {
 	}
 
 
-	private static class SynchronossPart implements Part {
+	private static abstract class AbstractSynchronossPart implements Part {
 
 		private final HttpHeaders headers;
 
-		private final StreamStorage storage;
-
-		private final String content;
-
-		private final DataBufferFactory bufferFactory = new DefaultDataBufferFactory();
-
+		private final DataBufferFactory bufferFactory;
 
-		SynchronossPart(HttpHeaders headers, StreamStorage storage) {
-			Assert.notNull(headers, "HttpHeaders is required");
-			Assert.notNull(storage, "'storage' is required");
-			this.headers = headers;
-			this.storage = storage;
-			this.content = null;
-		}
 
-		SynchronossPart(HttpHeaders headers, String content) {
+		AbstractSynchronossPart(HttpHeaders headers, DataBufferFactory bufferFactory) {
 			Assert.notNull(headers, "HttpHeaders is required");
-			Assert.notNull(content, "'content' is required");
+			Assert.notNull(bufferFactory, "'bufferFactory' is required");
 			this.headers = headers;
-			this.storage = null;
-			this.content = content;
+			this.bufferFactory = bufferFactory;
 		}
 
 
@@ -251,52 +247,53 @@ public HttpHeaders getHeaders() {
 			return this.headers;
 		}
 
-		@Override
-		public Optional<String> getFilename() {
-			return Optional.ofNullable(MultipartUtils.getFileName(this.headers));
+		protected DataBufferFactory getBufferFactory() {
+			return this.bufferFactory;
 		}
+	}
 
-		@Override
-		public Mono<String> getContentAsString() {
-			if (this.content != null) {
-				return Mono.just(this.content);
-			}
-			try {
-				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));
-			}
-		}
+	private static class DefaultSynchronossPart extends AbstractSynchronossPart {
 
-		private Charset getCharset() {
-			return Optional.ofNullable(this.headers.getContentType())
-					.map(MimeType::getCharset).orElse(StandardCharsets.UTF_8);
+		private final StreamStorage storage;
+
+
+		DefaultSynchronossPart(HttpHeaders headers, StreamStorage storage, DataBufferFactory factory) {
+			super(headers, factory);
+			Assert.notNull(storage, "'storage' is required");
+			this.storage = storage;
 		}
 
+
 		@Override
 		public Flux<DataBuffer> getContent() {
-			if (this.content != null) {
-				DataBuffer buffer = this.bufferFactory.allocateBuffer(this.content.length());
-				buffer.write(this.content.getBytes());
-				return Flux.just(buffer);
-			}
 			InputStream inputStream = this.storage.getInputStream();
-			return DataBufferUtils.read(inputStream, this.bufferFactory, 4096);
+			return DataBufferUtils.read(inputStream, getBufferFactory(), 4096);
+		}
+
+		protected StreamStorage getStorage() {
+			return this.storage;
+		}
+	}
+
+	private static class SynchronossFilePart extends DefaultSynchronossPart implements FilePart {
+
+
+		public SynchronossFilePart(HttpHeaders headers, StreamStorage storage, DataBufferFactory factory) {
+			super(headers, storage, factory);
+		}
+
+
+		@Override
+		public String getFilename() {
+			return MultipartUtils.getFileName(getHeaders());
 		}
 
 		@Override
 		public Mono<Void> transferTo(File destination) {
-			if (this.storage == null || !getFilename().isPresent()) {
-				return Mono.error(new IllegalStateException("The part does not represent a file."));
-			}
 			ReadableByteChannel input = null;
 			FileChannel output = null;
 			try {
-				input = Channels.newChannel(this.storage.getInputStream());
+				input = Channels.newChannel(getStorage().getInputStream());
 				output = new FileOutputStream(destination).getChannel();
 
 				long size = (input instanceof FileChannel ? ((FileChannel) input).size() : Long.MAX_VALUE);
@@ -332,4 +329,34 @@ public Mono<Void> transferTo(File destination) {
 		}
 	}
 
+	private static class SynchronossFormFieldPart extends AbstractSynchronossPart implements FormFieldPart {
+
+		private final String content;
+
+
+		SynchronossFormFieldPart(HttpHeaders headers, DataBufferFactory bufferFactory, String content) {
+			super(headers, bufferFactory);
+			this.content = content;
+		}
+
+
+		@Override
+		public String getValue() {
+			return this.content;
+		}
+
+		@Override
+		public Flux<DataBuffer> getContent() {
+			byte[] bytes = this.content.getBytes(getCharset());
+			DataBuffer buffer = getBufferFactory().allocateBuffer(bytes.length);
+			buffer.write(bytes);
+			return Flux.just(buffer);
+		}
+
+		private Charset getCharset() {
+			return Optional.ofNullable(MultipartUtils.getCharEncoding(getHeaders()))
+					.map(Charset::forName).orElse(StandardCharsets.UTF_8);
+		}
+	}
+
 }