Filter out empty PartEvents in PartEventHttpMessageWriter

This commit makes sure that PartEvents with empty data buffer are
filtered out before written. Empty buffers caused issues with the
JdkClientHttpConnector.

Closes gh-29400

Author: poutsma

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

@@ -28,6 +28,7 @@
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferFactory;
 import org.springframework.core.io.buffer.DataBufferUtils;
+import org.springframework.http.HttpHeaders;
 import org.springframework.http.MediaType;
 import org.springframework.http.ReactiveHttpOutputMessage;
 import org.springframework.http.codec.HttpMessageWriter;
@@ -83,7 +84,9 @@ public Mono<Void> write(Publisher<? extends PartEvent> partDataStream, Resolvabl
 					if (signal.hasValue()) {
 						PartEvent value = signal.get();
 						Assert.state(value != null, "Null value");
-						return encodePartData(boundary, outputMessage.bufferFactory(), value, flux);
+						Flux<DataBuffer> dataBuffers = flux.map(PartEvent::content)
+								.filter(buffer -> buffer.readableByteCount() > 0);
+						return encodePartData(boundary, outputMessage.bufferFactory(), value.headers(), dataBuffers);
 					}
 					else {
 						return flux.cast(DataBuffer.class);
@@ -99,11 +102,11 @@ public Mono<Void> write(Publisher<? extends PartEvent> partDataStream, Resolvabl
 		return outputMessage.writeWith(body);
 	}
 
-	private Flux<DataBuffer> encodePartData(byte[] boundary, DataBufferFactory bufferFactory, PartEvent first, Flux<? extends PartEvent> flux) {
+	private Flux<DataBuffer> encodePartData(byte[] boundary, DataBufferFactory bufferFactory, HttpHeaders headers, Flux<DataBuffer> body) {
 		return Flux.concat(
 				generateBoundaryLine(boundary, bufferFactory),
-				generatePartHeaders(first.headers(), bufferFactory),
-				flux.map(PartEvent::content),
+				generatePartHeaders(headers, bufferFactory),
+				body,
 				generateNewLine(bufferFactory));
 	}
 

spring-web/src/test/java/org/springframework/http/codec/multipart/PartEventHttpMessageWriterTests.java

@@ -0,0 +1,103 @@
+/*
+ * Copyright 2002-2020 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
+ *
+ *      https://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.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.Map;
+
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.core.ResolvableType;
+import org.springframework.core.codec.StringDecoder;
+import org.springframework.core.testfixture.io.buffer.AbstractLeakCheckingTests;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.MediaType;
+import org.springframework.util.MultiValueMap;
+import org.springframework.web.testfixture.http.server.reactive.MockServerHttpResponse;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.springframework.http.codec.multipart.MultipartHttpMessageWriterTests.parse;
+
+/**
+ * Unit tests for {@link PartHttpMessageWriter}.
+ *
+ * @author Arjen Poutsma
+ */
+public class PartEventHttpMessageWriterTests extends AbstractLeakCheckingTests {
+
+	private final PartEventHttpMessageWriter writer = new PartEventHttpMessageWriter();
+
+	private final MockServerHttpResponse response = new MockServerHttpResponse(this.bufferFactory);
+
+
+	@Test
+	public void canWrite() {
+		assertThat(this.writer.canWrite(ResolvableType.forClass(PartEvent.class), MediaType.MULTIPART_FORM_DATA)).isTrue();
+		assertThat(this.writer.canWrite(ResolvableType.forClass(FilePartEvent.class), MediaType.MULTIPART_FORM_DATA)).isTrue();
+		assertThat(this.writer.canWrite(ResolvableType.forClass(FormPartEvent.class), MediaType.MULTIPART_FORM_DATA)).isTrue();
+	}
+
+	@Test
+	void write() {
+		HttpHeaders headers = new HttpHeaders();
+		headers.setContentType(MediaType.TEXT_PLAIN);
+		Mono<FormPartEvent> formPartEvent = FormPartEvent.create("text part", "text");
+
+		Flux<FilePartEvent> filePartEvents =
+				FilePartEvent.create("file part", "file.txt", MediaType.APPLICATION_OCTET_STREAM,
+						Flux.just(
+								this.bufferFactory.wrap("Aa".getBytes(StandardCharsets.UTF_8)),
+								this.bufferFactory.wrap("Bb".getBytes(StandardCharsets.UTF_8)),
+								this.bufferFactory.wrap("Cc".getBytes(StandardCharsets.UTF_8))
+						));
+
+		Flux<PartEvent> partEvents = Flux.concat(
+				formPartEvent,
+				filePartEvents
+		);
+
+		Map<String, Object> hints = Collections.emptyMap();
+		this.writer.write(partEvents, null, MediaType.MULTIPART_FORM_DATA, this.response, hints)
+				.block(Duration.ofSeconds(5));
+
+		MultiValueMap<String, Part> requestParts = parse(this.response, hints);
+		assertThat(requestParts.size()).isEqualTo(2);
+
+		Part part = requestParts.getFirst("text part");
+		assertThat(part.name()).isEqualTo("text part");
+		assertThat(part.headers().getContentType().isCompatibleWith(MediaType.TEXT_PLAIN)).isTrue();
+		String value = decodeToString(part);
+		assertThat(value).isEqualTo("text");
+
+		part = requestParts.getFirst("file part");
+		assertThat(part.name()).isEqualTo("file part");
+		assertThat(((FilePart) part).filename()).isEqualTo("file.txt");
+		assertThat(decodeToString(part)).isEqualTo("AaBbCc");
+	}
+
+	@SuppressWarnings("ConstantConditions")
+	private String decodeToString(Part part) {
+		return StringDecoder.textPlainOnly().decodeToMono(part.content(),
+				ResolvableType.forClass(String.class), MediaType.TEXT_PLAIN,
+				Collections.emptyMap()).block(Duration.ZERO);
+	}
+
+}