fix #1478 Schedule the first channel.read() once the HttpHeaders are sent

reactor-netty-http/src/main/java/reactor/netty/http/HttpOperations.java

@@ -197,7 +197,8 @@ public Mono<Void> then() {
 					throw e;
 				}
 
-				return channel().writeAndFlush(msg);
+				return channel().writeAndFlush(msg)
+				                .addListener(f -> onHeadersSent());
 			}
 			else {
 				return channel().newSucceededFuture();
@@ -209,6 +210,8 @@ public Mono<Void> then() {
 
 	protected abstract void afterMarkSentHeaders();
 
+	protected abstract void onHeadersSent();
+
 	protected abstract HttpMessage newFullBodyMessage(ByteBuf body);
 
 	@Override

reactor-netty-http/src/main/java/reactor/netty/http/client/HttpClientOperations.java

@@ -524,6 +524,14 @@ protected void beforeMarkSentHeaders() {
 		}
 	}
 
+	@Override
+	protected void onHeadersSent() {
+		channel().read();
+		if (channel().parent() != null) {
+			channel().parent().read();
+		}
+	}
+
 	@Override
 	@SuppressWarnings("FutureReturnValueIgnored")
 	protected void onOutboundComplete() {

reactor-netty-http/src/main/java/reactor/netty/http/server/HttpServerOperations.java

@@ -576,6 +576,11 @@ protected void beforeMarkSentHeaders() {
 		//noop
 	}
 
+	@Override
+	protected void onHeadersSent() {
+		//noop
+	}
+
 	@Override
 	protected void onOutboundComplete() {
 		if (isWebsocket()) {

reactor-netty-http/src/test/java/reactor/netty/http/client/HttpClientTest.java

@@ -27,8 +27,10 @@
 import java.nio.channels.SocketChannel;
 import java.nio.charset.Charset;
 import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.nio.file.StandardOpenOption;
 import java.security.cert.CertificateException;
 import java.time.Duration;
 import java.util.ArrayList;
@@ -2552,4 +2554,39 @@ private void doTestEvictInBackground(int expectation, boolean evict) throws Exce
 		assertThat(m.get()).isNotNull();
 		assertThat(m.get().idleSize()).isEqualTo(expectation);
 	}
+
+	@Test
+	void testIssue1478() throws Exception {
+		disposableServer =
+				HttpServer.create()
+				          .handle((req, res) -> res.addHeader(HttpHeaderNames.CONNECTION, HttpHeaderValues.CLOSE)
+				                                   .status(HttpResponseStatus.BAD_REQUEST)
+				                                   .send(req.receive()
+				                                            .retain()
+				                                            .next()))
+				          .bindNow();
+
+		Path largeFile = Paths.get(getClass().getResource("/largeFile.txt").toURI());
+
+		Path largeFileParent = largeFile.getParent();
+		assertThat(largeFileParent).isNotNull();
+
+		Path tempFile = Files.createTempFile(largeFileParent, "temp", ".txt");
+		tempFile.toFile().deleteOnExit();
+
+		byte[] fileBytes = Files.readAllBytes(largeFile);
+		for (int i = 0; i < 1000; i++) {
+			Files.write(tempFile, fileBytes, StandardOpenOption.APPEND);
+		}
+
+		HttpClient.create()
+		          .port(disposableServer.port())
+		          .post()
+		          .send((req, out) -> out.sendFile(tempFile))
+		          .responseSingle((res, bytes) -> Mono.just(res.status().code()))
+		          .as(StepVerifier::create)
+		          .expectNext(400)
+		          .expectComplete()
+		          .verify(Duration.ofSeconds(5));
+	}
 }