WebClient writes Content-Length for Mono bodies

In SPR-16892, the `EncoderHttpMessageWriter` has been improved to write
`"Content-Length"` HTTP response headers if the response body is of type
`Mono` (i.e. the actual content length is easily accessible without
buffering a possibly large response body). That change was relying on
the fact that the server side is using a `ChannelSendOperator` to delay
the writing of the body until the first signal is received.

This strategy is not effective on the client side, since no such channel
operator is used for `WebClient`. This commit improves
`EncoderHttpMessageWriter` and delays, for `Mono` HTTP message bodies
only, the writing of the body so that we can write the
`"Content-Length"` header information once we've got the body resolved.

Issue: SPR-16949

spring-web/src/main/java/org/springframework/http/client/reactive/AbstractClientHttpRequest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -118,12 +118,12 @@ public abstract class AbstractClientHttpRequest implements ClientHttpRequest {
 			return Mono.empty();
 		}
 
-		this.commitActions.add(() -> {
+		this.commitActions.add(() ->
-			applyHeaders();
+				Mono.fromRunnable(() -> {
-			applyCookies();
+					applyHeaders();
-			this.state.set(State.COMMITTED);
+					applyCookies();
-			return Mono.empty();
+					this.state.set(State.COMMITTED);
-		});
+				}));
 
 		if (writeAction != null) {
 			this.commitActions.add(writeAction);
@@ -132,7 +132,7 @@ public abstract class AbstractClientHttpRequest implements ClientHttpRequest {
 		List<? extends Publisher<Void>> actions = this.commitActions.stream()
 				.map(Supplier::get).collect(Collectors.toList());
 
-		return Mono.fromDirect(Flux.concat(actions));
+		return Flux.concat(actions).then();
 	}
 
 

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

@@ -103,11 +103,14 @@ public class EncoderHttpMessageWriter<T> implements HttpMessageWriter<T> {
 		Flux<DataBuffer> body = this.encoder.encode(
 				inputStream, message.bufferFactory(), elementType, contentType, hints);
 
-		// Response is not committed until the first signal...
 		if (inputStream instanceof Mono) {
 			HttpHeaders headers = message.getHeaders();
 			if (headers.getContentLength() < 0 && !headers.containsKey(HttpHeaders.TRANSFER_ENCODING)) {
-				body = body.doOnNext(data -> headers.setContentLength(data.readableByteCount()));
+				return Mono.from(body)
+						.flatMap(dataBuffer -> {
+							headers.setContentLength(dataBuffer.readableByteCount());
+							return message.writeWith(Mono.just(dataBuffer));
+						});
 			}
 		}
 

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

@@ -340,10 +340,12 @@ public class BodyInsertersTests {
 					String content = new String(resultBytes, StandardCharsets.UTF_8);
 					assertThat(content, containsString("Content-Disposition: form-data; name=\"name\"\r\n" +
 							"Content-Type: text/plain;charset=UTF-8\r\n" +
+							"Content-Length: 6\r\n" +
 							"\r\n" +
 							"value1"));
 					assertThat(content, containsString("Content-Disposition: form-data; name=\"name\"\r\n" +
 							"Content-Type: text/plain;charset=UTF-8\r\n" +
+							"Content-Length: 6\r\n" +
 							"\r\n" +
 							"value2"));
 				})

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

@@ -318,7 +318,7 @@ public class WebClientIntegrationTests {
 		expectRequest(request -> {
 			assertEquals("/pojo/capitalize", request.getPath());
 			assertEquals("{\"foo\":\"foofoo\",\"bar\":\"barbar\"}", request.getBody().readUtf8());
-			assertEquals("chunked", request.getHeader(HttpHeaders.TRANSFER_ENCODING));
+			assertEquals("31", request.getHeader(HttpHeaders.CONTENT_LENGTH));
 			assertEquals("application/json", request.getHeader(HttpHeaders.ACCEPT));
 			assertEquals("application/json", request.getHeader(HttpHeaders.CONTENT_TYPE));
 		});