Add Flux<Part> ServerRequest.parts()

This commit introduces Flux<Part> ServerRequest.parts() that delegates
to ServerWebExchange.getParts() and offers an alternative,
streaming way of accessing multipart data.

Closes gh-23131

spring-webflux/src/test/java/org/springframework/web/reactive/function/MultipartIntegrationTests.java

@@ -16,6 +16,10 @@
 
 package org.springframework.web.reactive.function;
 
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
 import java.util.Map;
 
 import org.junit.Test;
@@ -29,6 +33,7 @@ import org.springframework.http.client.MultipartBodyBuilder;
 import org.springframework.http.codec.multipart.FilePart;
 import org.springframework.http.codec.multipart.FormFieldPart;
 import org.springframework.http.codec.multipart.Part;
+import org.springframework.util.FileCopyUtils;
 import org.springframework.util.MultiValueMap;
 import org.springframework.web.reactive.function.client.ClientResponse;
 import org.springframework.web.reactive.function.client.WebClient;
@@ -38,7 +43,7 @@ import org.springframework.web.reactive.function.server.ServerRequest;
 import org.springframework.web.reactive.function.server.ServerResponse;
 
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.springframework.web.reactive.function.server.RequestPredicates.POST;
+import static org.assertj.core.api.Assertions.fail;
 import static org.springframework.web.reactive.function.server.RouterFunctions.route;
 
 /**
@@ -48,6 +53,8 @@ public class MultipartIntegrationTests extends AbstractRouterFunctionIntegration
 
 	private final WebClient webClient = WebClient.create();
 
+	private ClassPathResource resource = new ClassPathResource("org/springframework/http/codec/multipart/foo.txt");
+
 
 	@Test
 	public void multipartData() {
@@ -77,9 +84,33 @@ public class MultipartIntegrationTests extends AbstractRouterFunctionIntegration
 				.verifyComplete();
 	}
 
+	@Test
+	public void transferTo() {
+		Mono<String> result = webClient
+				.post()
+				.uri("http://localhost:" + this.port + "/transferTo")
+				.syncBody(generateBody())
+				.retrieve()
+				.bodyToMono(String.class);
+
+		StepVerifier
+				.create(result)
+				.consumeNextWith(location -> {
+					try {
+						byte[] actualBytes = Files.readAllBytes(Paths.get(location));
+						byte[] expectedBytes = FileCopyUtils.copyToByteArray(this.resource.getInputStream());
+						assertThat(actualBytes).isEqualTo(expectedBytes);
+					}
+					catch (IOException ex) {
+						fail("IOException", ex);
+					}
+				})
+				.verifyComplete();
+	}
+
 	private MultiValueMap<String, HttpEntity<?>> generateBody() {
 		MultipartBodyBuilder builder = new MultipartBodyBuilder();
-		builder.part("fooPart", new ClassPathResource("org/springframework/http/codec/multipart/foo.txt"));
+		builder.part("fooPart", resource);
 		builder.part("barPart", "bar");
 		return builder.build();
 	}
@@ -87,44 +118,65 @@ public class MultipartIntegrationTests extends AbstractRouterFunctionIntegration
 	@Override
 	protected RouterFunction<ServerResponse> routerFunction() {
 		MultipartHandler multipartHandler = new MultipartHandler();
-		return route(POST("/multipartData"), multipartHandler::multipartData)
-				.andRoute(POST("/parts"), multipartHandler::parts);
+		return route()
+				.POST("/multipartData", multipartHandler::multipartData)
+				.POST("/parts", multipartHandler::parts)
+				.POST("/transferTo", multipartHandler::transferTo)
+				.build();
 	}
 
 
 	private static class MultipartHandler {
 
 		public Mono<ServerResponse> multipartData(ServerRequest request) {
-			return request
-					.body(BodyExtractors.toMultipartData())
+			return request.multipartData()
 					.flatMap(map -> {
 						Map<String, Part> parts = map.toSingleValueMap();
 						try {
 							assertThat(parts.size()).isEqualTo(2);
 							assertThat(((FilePart) parts.get("fooPart")).filename()).isEqualTo("foo.txt");
 							assertThat(((FormFieldPart) parts.get("barPart")).value()).isEqualTo("bar");
+							return ServerResponse.ok().build();
 						}
 						catch(Exception e) {
 							return Mono.error(e);
 						}
-						return ServerResponse.ok().build();
 					});
 		}
 
 		public Mono<ServerResponse> parts(ServerRequest request) {
-			return request.body(BodyExtractors.toParts()).collectList()
+			return request.parts().collectList()
 					.flatMap(parts -> {
 						try {
 							assertThat(parts.size()).isEqualTo(2);
 							assertThat(((FilePart) parts.get(0)).filename()).isEqualTo("foo.txt");
 							assertThat(((FormFieldPart) parts.get(1)).value()).isEqualTo("bar");
+							return ServerResponse.ok().build();
 						}
 						catch(Exception e) {
 							return Mono.error(e);
 						}
-						return ServerResponse.ok().build();
 					});
 		}
+
+		public Mono<ServerResponse> transferTo(ServerRequest request) {
+			return request.parts()
+					.filter(part -> part instanceof FilePart)
+					.next()
+					.cast(FilePart.class)
+					.flatMap(part -> {
+						try {
+							Path tempFile = Files.createTempFile("MultipartIntegrationTests", null);
+							return part.transferTo(tempFile)
+									.then(ServerResponse.ok()
+											.syncBody(tempFile.toString()));
+						}
+						catch (Exception e) {
+							return Mono.error(e);
+						}
+					});
+		}
+
 	}
 
 }

spring-webflux/src/test/java/org/springframework/web/reactive/function/server/MockServerRequest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2018 the original author or authors.
+ * Copyright 2002-2019 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.
@@ -247,6 +247,13 @@ public class MockServerRequest implements ServerRequest {
 		return (Mono<MultiValueMap<String, Part>>) this.body;
 	}
 
+	@Override
+	@SuppressWarnings("unchecked")
+	public Flux<Part> parts() {
+		Assert.state(this.body != null, "No body");
+		return (Flux<Part>) this.body;
+	}
+
 	@Override
 	public ServerWebExchange exchange() {
 		Assert.state(this.exchange != null, "No exchange");