ContentEncodingHttpRequesterFilter to remove content-length (#2365)

Motivation:
ContentEncodingHttpRequesterFilter will apply compression or
decompression. This may change the overall payload length and
if the content-length header is present it will likely be
incorrect after.

Modifications:
- Remove the content-length header if encoding/decoding is
  applied in ContentEncodingHttpRequesterFilter. Headers are
  already manipulated to add/remove content-encoding.

servicetalk-http-api/build.gradle

@@ -48,6 +48,7 @@ dependencies {
   testImplementation testFixtures(project(":servicetalk-transport-netty-internal"))
   testImplementation project(":servicetalk-buffer-netty")
   testImplementation project(":servicetalk-concurrent-test-internal")
+  testImplementation project(":servicetalk-encoding-netty")
   testImplementation project(":servicetalk-test-resources")
   testImplementation project(":servicetalk-transport-netty-internal")
   testImplementation "org.junit.jupiter:junit-jupiter-api"

servicetalk-http-api/src/main/java/io/servicetalk/http/api/ContentEncodingHttpRequesterFilter.java

@@ -27,6 +27,7 @@
 import static io.servicetalk.http.api.HeaderUtils.addContentEncoding;
 import static io.servicetalk.http.api.HttpHeaderNames.ACCEPT_ENCODING;
 import static io.servicetalk.http.api.HttpHeaderNames.CONTENT_ENCODING;
+import static io.servicetalk.http.api.HttpHeaderNames.CONTENT_LENGTH;
 import static java.util.Objects.requireNonNull;
 
 /**
@@ -94,6 +95,8 @@ private Single<StreamingHttpResponse> applyEncodingAndDecoding(final StreamingHt
             final StreamingHttpRequest encodedRequest;
             if (encoder != null && !identityEncoder().equals(encoder)) {
                 addContentEncoding(request.headers(), encoder.encodingName());
+                // After we encode the content length is unlikely to still be correct, remove it!
+                request.headers().remove(CONTENT_LENGTH);
                 encodedRequest = request.transformPayloadBody(pub -> encoder.streamingEncoder().serialize(pub,
                         delegate.executionContext().bufferAllocator()));
             } else {
@@ -115,6 +118,8 @@ private Single<StreamingHttpResponse> applyEncodingAndDecoding(final StreamingHt
                             "<null>").toString());
                 }
 
+                // After we decode the content length is unlikely to still be correct, remove it!
+                response.headers().remove(CONTENT_LENGTH);
                 return response.transformPayloadBody(pub -> decoder.streamingDecoder().deserialize(pub,
                         delegate.executionContext().bufferAllocator()));
             }) : respSingle).shareContextOnSubscribe();

servicetalk-http-api/src/test/java/io/servicetalk/http/api/ContentEncodingHttpServiceFilterTest.java

@@ -15,22 +15,36 @@
  */
 package io.servicetalk.http.api;
 
+import io.servicetalk.buffer.api.Buffer;
 import io.servicetalk.buffer.api.CharSequences;
+import io.servicetalk.concurrent.api.Publisher;
+import io.servicetalk.concurrent.api.Single;
+import io.servicetalk.encoding.api.BufferDecoderGroupBuilder;
 
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
 
 import java.util.ArrayList;
 import java.util.List;
 
 import static io.servicetalk.buffer.api.CharSequences.contentEqualsIgnoreCase;
+import static io.servicetalk.buffer.netty.BufferAllocators.DEFAULT_ALLOCATOR;
+import static io.servicetalk.concurrent.api.Publisher.from;
+import static io.servicetalk.concurrent.api.Single.failed;
+import static io.servicetalk.encoding.api.Identity.identityEncoder;
+import static io.servicetalk.encoding.netty.NettyBufferEncoders.gzipDefault;
 import static io.servicetalk.http.api.ContentEncodingHttpServiceFilter.matchAndRemoveEncoding;
 import static io.servicetalk.http.api.HttpHeaderNames.CONTENT_ENCODING;
+import static io.servicetalk.http.api.HttpHeaderNames.CONTENT_LENGTH;
 import static java.util.function.Function.identity;
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.nullValue;
 
-class ContentEncodingHttpServiceFilterTest {
+class ContentEncodingHttpServiceFilterTest extends AbstractHttpRequesterFilterTest {
+
     @Test
     void testMatchAndRemoveEncodingFirst() {
         List<CharSequence> supportedDecoders = new ArrayList<>();
@@ -73,4 +87,46 @@ void testMatchAndRemoveEncodingMiddleDoesNotMatch() {
         assertThat("unexpected header: " + contentEncoding,
                 contentEqualsIgnoreCase(contentEncoding, " deflate , foo , gzip "), is(true));
     }
+
+    @ParameterizedTest(name = "{displayName} [{index}] {0}-{1}")
+    @MethodSource("requesterTypes")
+    void contentLengthRemovedOnEncode(final RequesterType type,
+                                      final SecurityType security) throws Exception {
+        setUp(security);
+        StreamingHttpRequester filter = createFilter(type, (respFactory, request) -> {
+            try {
+                assertContentLength(request.headers(), request.payloadBody());
+            } catch (Exception e) {
+                return failed(e);
+            }
+
+            // Simulate the server compressed reply, with CONTENT_LENGTH included.
+            String responseStr = "aaaaaaaaaaaaaaaa";
+            Buffer responseBuf = DEFAULT_ALLOCATOR.fromAscii(responseStr);
+            Buffer encoded = gzipDefault().encoder().serialize(responseBuf, DEFAULT_ALLOCATOR);
+            return Single.succeeded(respFactory.ok().payloadBody(from(encoded))
+                    .addHeader(CONTENT_LENGTH, String.valueOf(encoded.readableBytes()))
+                    .addHeader(CONTENT_ENCODING, gzipDefault().encodingName()));
+        }, new ContentEncodingHttpRequesterFilter(new BufferDecoderGroupBuilder()
+                .add(gzipDefault(), true)
+                .add(identityEncoder(), false)
+                .build()));
+
+        // Simulate the user (or earlier filter) setting the content length before compression.
+        StreamingHttpRequest request = filter.post("/foo");
+        String payloadBody = "bbbbbbbbbbbbbbbbbbb";
+        request.payloadBody(from(filter.executionContext().bufferAllocator().fromAscii(payloadBody)));
+        request.headers().add(CONTENT_LENGTH, String.valueOf(payloadBody.length()));
+        request.contentEncoding(gzipDefault());
+        StreamingHttpResponse response = filter.request(request).toFuture().get();
+        assertContentLength(response.headers(), response.payloadBody());
+    }
+
+    private static void assertContentLength(HttpHeaders headers, Publisher<Buffer> publisher) throws Exception {
+        CharSequence reqLen = headers.get(CONTENT_LENGTH);
+        if (reqLen != null) {
+            final int len = publisher.collect(() -> 0, (sum, buff) -> sum + buff.readableBytes()).toFuture().get();
+            assertThat(Integer.parseInt(reqLen.toString()), equalTo(len));
+        }
+    }
 }

servicetalk-http-api/src/testFixtures/java/io/servicetalk/http/api/AbstractHttpRequesterFilterTest.java

@@ -92,6 +92,7 @@ final void setupContext() {
 
     protected void setUp(SecurityType security) {
         lenient().when(mockExecutionContext.executionStrategy()).thenReturn(defaultStrategy());
+        lenient().when(mockExecutionContext.bufferAllocator()).thenReturn(DEFAULT_ALLOCATOR);
         lenient().when(mockConnectionContext.sslSession()).thenAnswer(__ -> {
             switch (security) {
                 case Secure: