Add reactive multipart request support

build.gradle

@@ -89,6 +89,7 @@ configure(allprojects) { project ->
 	ext.servletVersion         = "3.1.0"
 	ext.slf4jVersion           = "1.7.25"
 	ext.snakeyamlVersion       = "1.18"
+	ext.nioMultipartVersion    = "1.0.2"
 	ext.testngVersion          = "6.11"
 	ext.tiles3Version          = "3.0.7"
 	ext.tomcatVersion          = "8.5.14"
@@ -747,6 +748,7 @@ project("spring-web") {
 		optional("javax.xml.bind:jaxb-api:${jaxbVersion}")
 		optional("javax.xml.ws:jaxws-api:${jaxwsVersion}")
 		optional("javax.mail:javax.mail-api:${javamailVersion}")
+		optional("org.synchronoss.cloud:nio-multipart-parser:${nioMultipartVersion}")
 		optional("org.jetbrains.kotlin:kotlin-stdlib-jre8:${kotlinVersion}")
 		testCompile(project(":spring-context-support"))  // for JafMediaTypeFactory
 		testCompile("io.projectreactor.addons:reactor-test")
@@ -838,6 +840,10 @@ project("spring-webflux") {
 		testRuntime("org.jetbrains.kotlin:kotlin-compiler:${kotlinVersion}")
 		testCompile("org.jetbrains.kotlin:kotlin-script-runtime:${kotlinVersion}")
 		testRuntime("org.jetbrains.kotlin:kotlin-script-util:${kotlinVersion}")
+		testRuntime("org.synchronoss.cloud:nio-multipart-parser:${nioMultipartVersion}")
+		testRuntime("com.sun.mail:javax.mail:${javamailVersion}")
+		testRuntime("com.sun.xml.bind:jaxb-core:${jaxbVersion}")
+		testRuntime("com.sun.xml.bind:jaxb-impl:${jaxbVersion}")
 	}
 
 	if (JavaVersion.current().java9Compatible) {

spring-test/src/main/java/org/springframework/mock/http/server/reactive/MockServerWebExchange.java

@@ -15,6 +15,7 @@
  */
 package org.springframework.mock.http.server.reactive;
 
+import org.springframework.http.codec.ServerCodecConfigurer;
 import org.springframework.web.server.ServerWebExchangeDecorator;
 import org.springframework.web.server.adapter.DefaultServerWebExchange;
 import org.springframework.web.server.session.DefaultWebSessionManager;
@@ -35,7 +36,8 @@ public class MockServerWebExchange extends ServerWebExchangeDecorator {
 
 	public MockServerWebExchange(MockServerHttpRequest request) {
 		super(new DefaultServerWebExchange(
-				request, new MockServerHttpResponse(), new DefaultWebSessionManager()));
+				request, new MockServerHttpResponse(), new DefaultWebSessionManager(),
+				ServerCodecConfigurer.create()));
 	}
 
 

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

@@ -21,6 +21,7 @@
 import org.springframework.core.codec.Decoder;
 import org.springframework.core.codec.StringDecoder;
 import org.springframework.http.codec.json.Jackson2JsonDecoder;
+import org.springframework.http.codec.multipart.MultipartHttpMessageWriter;
 
 /**
  * Default implementation of {@link ClientCodecConfigurer}.
@@ -57,6 +58,7 @@ public void serverSentEventDecoder(Decoder<?> decoder) {
 		protected void addTypedWritersTo(List<HttpMessageWriter<?>> result) {
 			super.addTypedWritersTo(result);
 			addWriterTo(result, FormHttpMessageWriter::new);
+			addWriterTo(result, MultipartHttpMessageWriter::new);
 		}
 
 		@Override

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

@@ -21,6 +21,8 @@
 import org.springframework.core.codec.Encoder;
 import org.springframework.core.codec.StringDecoder;
 import org.springframework.http.codec.json.Jackson2JsonEncoder;
+import org.springframework.http.codec.multipart.SynchronossMultipartHttpMessageReader;
+import org.springframework.util.ClassUtils;
 
 /**
  * Default implementation of {@link ServerCodecConfigurer}.
@@ -30,6 +32,11 @@
  */
 class DefaultServerCodecConfigurer extends DefaultCodecConfigurer implements ServerCodecConfigurer {
 
+	static final boolean synchronossMultipartPresent =
+			ClassUtils.isPresent("org.synchronoss.cloud.nio.multipart.NioMultipartParser",
+					org.springframework.http.codec.DefaultCodecConfigurer.class.getClassLoader());
+
+
 	public DefaultServerCodecConfigurer() {
 		super(new DefaultServerDefaultCodecsConfigurer());
 	}
@@ -57,6 +64,9 @@ public void serverSentEventEncoder(Encoder<?> encoder) {
 		public void addTypedReadersTo(List<HttpMessageReader<?>> result) {
 			super.addTypedReadersTo(result);
 			addReaderTo(result, FormHttpMessageReader::new);
+			if (synchronossMultipartPresent) {
+				addReaderTo(result, SynchronossMultipartHttpMessageReader::new);
+			}
 		}
 
 		@Override

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

@@ -0,0 +1,58 @@
+/*
+ * 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));
+	}
+}