Use DataBufferUtils.write in DefaultFilePart.transferTo

This commit makes sure that in DefaultMultipartMessageReader's
DefaultFilePart, the file is not closed before all bytes are written,
by using DataBufferUtils.write (see c1b6885191d6a50347aeaa14da994f0db88f26fe).

The commit also improves on the logging of the
DefaultMultipartMessageReader.

Closes gh-23130

Author: poutsma

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

@@ -16,14 +16,9 @@
 
 package org.springframework.http.codec.multipart;
 
-import java.io.IOException;
-import java.nio.channels.AsynchronousFileChannel;
-import java.nio.channels.Channel;
 import java.nio.charset.Charset;
 import java.nio.charset.StandardCharsets;
-import java.nio.file.OpenOption;
 import java.nio.file.Path;
-import java.nio.file.StandardOpenOption;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
@@ -100,10 +95,6 @@ public Flux<Part> read(ResolvableType elementType, ReactiveHttpInputMessage mess
 			return Flux.error(new CodecException("No multipart boundary found in Content-Type: \"" +
 					message.getHeaders().getContentType() + "\""));
 		}
-		if (logger.isTraceEnabled()) {
-			logger.trace("Boundary: " + toString(boundary));
-		}
-
 		byte[] boundaryNeedle = concat(BOUNDARY_PREFIX, boundary);
 		Flux<DataBuffer> body = skipUntilFirstBoundary(message.getBody(), boundary);
 
@@ -148,8 +139,10 @@ private static Flux<DataBuffer> skipUntilFirstBoundary(Flux<DataBuffer> dataBuff
 					DataBuffer slice = dataBuffer.retainedSlice(endIdx + 1, length);
 					DataBufferUtils.release(dataBuffer);
 					if (logger.isTraceEnabled()) {
-						logger.trace("Found first boundary at " + endIdx + " in " + toString(dataBuffer));
-					}
+							logger.trace(
+									"Found last byte of first boundary (" + toString(boundary)
+											+ ") at " + endIdx);
+						}
 					return Mono.just(slice);
 				}
 				else {
@@ -188,14 +181,14 @@ private static Part toPart(DataBuffer dataBuffer) {
 			}
 		}
 
-		if (logger.isTraceEnabled()) {
-			logger.trace("Part data: " + toString(dataBuffer));
-		}
 		int endIdx = HEADER_MATCHER.match(dataBuffer);
 
 		HttpHeaders headers;
 		DataBuffer body;
 		if (endIdx > 0) {
+			if (logger.isTraceEnabled()) {
+				logger.trace("Found last byte of part header at " + endIdx );
+			}
 			readPosition = dataBuffer.readPosition();
 			int headersLength = endIdx + 1 - (readPosition + HEADER_BODY_SEPARATOR.length);
 			DataBuffer headersBuffer = dataBuffer.retainedSlice(readPosition, headersLength);
@@ -204,6 +197,9 @@ private static Part toPart(DataBuffer dataBuffer) {
 			headers = toHeaders(headersBuffer);
 		}
 		else {
+			if (logger.isTraceEnabled()) {
+				logger.trace("No header found");
+			}
 			headers = new HttpHeaders();
 			body = DataBufferUtils.retain(dataBuffer);
 		}
@@ -252,16 +248,6 @@ private static HttpHeaders toHeaders(DataBuffer dataBuffer) {
 		return result;
 	}
 
-
-	private static String toString(DataBuffer dataBuffer) {
-		byte[] bytes = new byte[dataBuffer.readableByteCount()];
-		int j = 0;
-		for (int i = dataBuffer.readPosition(); i < dataBuffer.writePosition(); i++) {
-			bytes[j++] = dataBuffer.getByte(i);
-		}
-		return toString(bytes);
-	}
-
 	private static String toString(byte[] bytes) {
 		StringBuilder builder = new StringBuilder();
 		for (byte b : bytes) {
@@ -368,10 +354,6 @@ public String value() {
 
 	private static class DefaultFilePart extends DefaultPart implements FilePart {
 
-		private static final OpenOption[] FILE_CHANNEL_OPTIONS =
-				{StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING, StandardOpenOption.WRITE};
-
-
 		public DefaultFilePart(HttpHeaders headers, DataBuffer body) {
 			super(headers, body);
 		}
@@ -385,23 +367,9 @@ public String filename() {
 
 		@Override
 		public Mono<Void> transferTo(Path dest) {
-			return Mono.using(() -> AsynchronousFileChannel.open(dest, FILE_CHANNEL_OPTIONS),
-					this::writeBody, this::close);
-		}
-
-		private Mono<Void> writeBody(AsynchronousFileChannel channel) {
-			return DataBufferUtils.write(content(), channel)
-					.map(DataBufferUtils::release)
-					.then();
+			return DataBufferUtils.write(content(), dest);
 		}
 
-		private void close(Channel channel) {
-			try {
-				channel.close();
-			}
-			catch (IOException ignore) {
-			}
-		}
 	}
 
 }

spring-webflux/src/test/java/org/springframework/web/reactive/result/method/annotation/MultipartIntegrationTests.java

@@ -16,6 +16,10 @@
 
 package org.springframework.web.reactive.result.method.annotation;
 
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
 import java.util.List;
 import java.util.stream.Collectors;
 
@@ -31,6 +35,7 @@
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.core.io.ClassPathResource;
+import org.springframework.core.io.Resource;
 import org.springframework.http.HttpEntity;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.client.MultipartBodyBuilder;
@@ -145,6 +150,34 @@ public void filePartsMono() {
 				.verifyComplete();
 	}
 
+	@Test
+	public void transferTo() {
+		Flux<String> result = webClient
+				.post()
+				.uri("/transferTo")
+				.syncBody(generateBody())
+				.retrieve()
+				.bodyToFlux(String.class);
+
+		StepVerifier.create(result)
+				.consumeNextWith(filename -> verifyContents(Paths.get(filename), new ClassPathResource("foo.txt", MultipartHttpMessageReader.class)))
+				.consumeNextWith(filename -> verifyContents(Paths.get(filename), new ClassPathResource("logo.png", getClass())))
+				.verifyComplete();
+
+	}
+
+	private static void verifyContents(Path tempFile, Resource resource) {
+		try {
+			byte[] tempBytes = Files.readAllBytes(tempFile);
+			byte[] resourceBytes = Files.readAllBytes(resource.getFile().toPath());
+			assertThat(tempBytes).isEqualTo(resourceBytes);
+		}
+		catch (IOException ex) {
+			throw new AssertionError(ex);
+		}
+	}
+
+
 	@Test
 	public void modelAttribute() {
 		Mono<String> result = webClient
@@ -217,6 +250,21 @@ Mono<String> filePartsFlux(@RequestPart("fileParts") Mono<FilePart> parts) {
 			return partFluxDescription(Flux.from(parts));
 		}
 
+		@PostMapping("/transferTo")
+		Flux<String> transferTo(@RequestPart("fileParts") Flux<FilePart> parts) {
+			return parts.flatMap(filePart -> {
+				try {
+					Path tempFile = Files.createTempFile("MultipartIntegrationTests", filePart.filename());
+					return filePart.transferTo(tempFile)
+							.then(Mono.just(tempFile.toString() + "\n"));
+
+				}
+				catch (IOException e) {
+					return Mono.error(e);
+				}
+			});
+		}
+
 		@PostMapping("/modelAttribute")
 		String modelAttribute(@ModelAttribute FormBean formBean) {
 			return formBean.toString();