Do not concatenate Publisher.empty() message body (#1497)

Motivation:

For `HeaderUtils#setContentLength(...)`, when the `messageBody` is exactly
`Publisher.empty()` there are no operators applied for the `messageBody`.
In this case we do not need concatenation.

Modifications:

- Skip concatenation when `messageBody == empty()`;
- Add tests when payloadBody is not set at all;

Result:

Less operators applied during request/response processing when not
necessary.

servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/HeaderUtils.java

@@ -177,7 +177,11 @@ private static Publisher<Object> setContentLength(final HttpMetaData metadata,
                                                       final BiIntConsumer<HttpHeaders> contentLengthUpdater) {
         if (messageBody == empty() || (isPayloadEmpty(metadata) && !mayHaveTrailers(metadata))) {
             contentLengthUpdater.apply(0, metadata.headers());
-            return from(metadata, EmptyHttpHeaders.INSTANCE).concat(messageBody.ignoreElements());
+            return messageBody == empty() ?
+                    from(metadata, EmptyHttpHeaders.INSTANCE) :
+                    // Subscribe to the messageBody publisher to trigger any applied transformations, but ignore its
+                    // content because the PayloadInfo indicated it's effectively empty and does not contain trailers
+                    from(metadata, EmptyHttpHeaders.INSTANCE).concat(messageBody.ignoreElements());
         }
         return messageBody.collect(() -> null, (reduction, item) -> {
             if (reduction == null) {

servicetalk-http-netty/src/test/java/io/servicetalk/http/netty/ContentLengthTest.java

@@ -83,17 +83,27 @@ private static void shouldNotCalculateRequestContentLengthFromEmptyPublisher(Htt
         setRequestContentLengthAndVerify(request, nullValue(CharSequence.class));
     }
 
+    @Test
+    public void shouldCalculateRequestContentLengthWhenNoPayloadBodySet() throws Exception {
+        shouldCalculateRequestContentLengthFromEmptyPublisher(POST, false);
+        shouldCalculateRequestContentLengthFromEmptyPublisher(PUT, false);
+        shouldCalculateRequestContentLengthFromEmptyPublisher(PATCH, false);
+    }
+
     @Test
     public void shouldCalculateRequestContentLengthFromEmptyPublisher() throws Exception {
-        shouldCalculateRequestContentLengthFromEmptyPublisher(POST);
-        shouldCalculateRequestContentLengthFromEmptyPublisher(PUT);
-        shouldCalculateRequestContentLengthFromEmptyPublisher(PATCH);
+        shouldCalculateRequestContentLengthFromEmptyPublisher(POST, true);
+        shouldCalculateRequestContentLengthFromEmptyPublisher(PUT, true);
+        shouldCalculateRequestContentLengthFromEmptyPublisher(PATCH, true);
     }
 
-    private static void shouldCalculateRequestContentLengthFromEmptyPublisher(HttpRequestMethod method)
+    private static void shouldCalculateRequestContentLengthFromEmptyPublisher(HttpRequestMethod method,
+                                                                              boolean emptyPublisher)
             throws Exception {
-        StreamingHttpRequest request = newAggregatedRequest(method).toStreamingRequest()
-                .payloadBody(Publisher.empty());
+        StreamingHttpRequest request = newAggregatedRequest(method).toStreamingRequest();
+        if (emptyPublisher) {
+            request = request.payloadBody(Publisher.empty());
+        }
         setRequestContentLengthAndVerify(request, contentEqualTo("0"));
     }
 
@@ -134,6 +144,12 @@ public void shouldCalculateRequestContentLengthFromTransformedRawMultipleItemPub
         setRequestContentLengthAndVerify(request, contentEqualTo("12"));
     }
 
+    @Test
+    public void shouldCalculateResponseContentLengthWhenNoPayloadBodySet() throws Exception {
+        StreamingHttpResponse response = newAggregatedResponse().toStreamingResponse();
+        setResponseContentLengthAndVerify(response, contentEqualTo("0"));
+    }
+
     @Test
     public void shouldCalculateResponseContentLengthFromEmptyPublisher() throws Exception {
         StreamingHttpResponse response = newAggregatedResponse().toStreamingResponse()
@@ -187,28 +203,31 @@ private static HttpResponse newAggregatedResponse() throws Exception {
 
     private static void setRequestContentLengthAndVerify(final StreamingHttpRequest request,
                                                          final Matcher<CharSequence> matcher) throws Exception {
+        final AtomicBoolean messageBodySubscribed = new AtomicBoolean(false);
+        request.transformMessageBody(publisher -> publisher.afterOnSubscribe(__ -> messageBodySubscribed.set(true)));
         Collection<Object> flattened = setRequestContentLength(request).toFuture().get();
-        assertThat("Unexpected items in the flattened request.", flattened, hasSize(greaterThanOrEqualTo(2)));
-        Iterator<Object> iterator = flattened.iterator();
-        Object firstItem = iterator.next();
-        assertThat("Unexpected items in the flattened request.", firstItem, instanceOf(HttpMetaData.class));
-        assertThat(((HttpMetaData) firstItem).headers().get(CONTENT_LENGTH), matcher);
-        assertLastItems(iterator);
+        assertFlattened(flattened, matcher, messageBodySubscribed);
     }
 
     private static void setResponseContentLengthAndVerify(final StreamingHttpResponse response,
                                                           final Matcher<CharSequence> matcher) throws Exception {
 
         final AtomicBoolean messageBodySubscribed = new AtomicBoolean(false);
-        response.transformMessageBody(publisher -> publisher.afterOnSubscribe((__) -> messageBodySubscribed.set(true)));
+        response.transformMessageBody(publisher -> publisher.afterOnSubscribe(__ -> messageBodySubscribed.set(true)));
         Collection<Object> flattened = setResponseContentLength(response).toFuture().get();
-        assertThat("Unexpected items in the flattened response.", flattened, hasSize(greaterThanOrEqualTo(2)));
+        assertFlattened(flattened, matcher, messageBodySubscribed);
+    }
+
+    private static void assertFlattened(final Collection<Object> flattened,
+                                        final Matcher<CharSequence> matcher,
+                                        final AtomicBoolean messageBodySubscribed) {
+        assertThat("Unexpected number of items in the flattened stream.", flattened, hasSize(greaterThanOrEqualTo(2)));
         Iterator<Object> iterator = flattened.iterator();
         Object firstItem = iterator.next();
-        assertThat("Unexpected items in the flattened response.", firstItem, instanceOf(HttpMetaData.class));
+        assertThat("Unexpected first item in the flattened stream.", firstItem, instanceOf(HttpMetaData.class));
         assertThat(((HttpMetaData) firstItem).headers().get(CONTENT_LENGTH), matcher);
         assertLastItems(iterator);
-        assertThat(true, is(messageBodySubscribed.get()));
+        assertThat("No subscribe for message body", messageBodySubscribed.get(), is(true));
     }
 
     private static void assertLastItems(Iterator<Object> iterator) {