HTTPCLIENT-2428: adjust reactor capacity in inflating AsyncDataConsumers

httpclient5/src/main/java/org/apache/hc/client5/http/async/methods/InflatingAsyncDataConsumer.java

@@ -73,7 +73,7 @@ public InflatingAsyncDataConsumer(
 
     @Override
     public void updateCapacity(final CapacityChannel ch) throws IOException {
-        downstream.updateCapacity(ch);
+        downstream.updateCapacity(new InflatingCapacityChannel(ch));
     }
 
     @Override

httpclient5/src/main/java/org/apache/hc/client5/http/async/methods/InflatingBrotliDataConsumer.java

@@ -67,7 +67,6 @@ public final class InflatingBrotliDataConsumer implements AsyncDataConsumer {
 
     private final AsyncDataConsumer downstream;
     private final DecoderJNI.Wrapper decoder;
-    private volatile CapacityChannel capacity;
 
 
     public InflatingBrotliDataConsumer(final AsyncDataConsumer downstream) {
@@ -81,8 +80,7 @@ public InflatingBrotliDataConsumer(final AsyncDataConsumer downstream) {
 
     @Override
     public void updateCapacity(final CapacityChannel capacityChannel) throws IOException {
-        this.capacity = capacityChannel;
-        downstream.updateCapacity(capacityChannel);
+        downstream.updateCapacity(new InflatingCapacityChannel(capacityChannel));
     }
 
     @Override
@@ -103,10 +101,6 @@ public void consume(final ByteBuffer src) throws IOException {
             decoder.push(xfer);
             pump();
         }
-        final CapacityChannel ch = this.capacity;
-        if (ch != null) {
-            ch.update(Integer.MAX_VALUE);
-        }
     }
 
     @Override

httpclient5/src/main/java/org/apache/hc/client5/http/async/methods/InflatingCapacityChannel.java

@@ -0,0 +1,74 @@
+/*
+ * ====================================================================
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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
+ *
+ *   http://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.
+ * ====================================================================
+ *
+ * This software consists of voluntary contributions made by many
+ * individuals on behalf of the Apache Software Foundation.  For more
+ * information on the Apache Software Foundation, please see
+ * <http://www.apache.org/>.
+ *
+ */
+package org.apache.hc.client5.http.async.methods;
+
+import java.io.IOException;
+
+import org.apache.hc.core5.http.nio.CapacityChannel;
+import org.apache.hc.core5.util.Args;
+
+/**
+ * {@link CapacityChannel} decorator used by the inflating {@code AsyncDataConsumer}s. A downstream
+ * consumer advertises capacity in decompressed bytes, but the increment is applied to the compressed
+ * stream read from the I/O reactor. Forwarding it unchanged lets the reactor deliver that many
+ * compressed bytes, which decompress into several times as many bytes and overshoot the downstream
+ * consumer's advertised capacity. This decorator scales the requested increment down by a fixed
+ * expansion factor so the amount of decompressed data delivered per round stays closer to what the
+ * downstream consumer asked for.
+ */
+final class InflatingCapacityChannel implements CapacityChannel {
+
+    /**
+     * Assumed decompressed-to-compressed size ratio. A conservative estimate for typical HTTP
+     * payloads; a larger value trades throughput for tighter capacity predictability.
+     */
+    static final int DEFAULT_EXPANSION_FACTOR = 4;
+
+    private final CapacityChannel delegate;
+    private final int expansionFactor;
+
+    InflatingCapacityChannel(final CapacityChannel delegate) {
+        this(delegate, DEFAULT_EXPANSION_FACTOR);
+    }
+
+    InflatingCapacityChannel(final CapacityChannel delegate, final int expansionFactor) {
+        this.delegate = Args.notNull(delegate, "Capacity channel");
+        this.expansionFactor = Args.positive(expansionFactor, "Expansion factor");
+    }
+
+    @Override
+    public void update(final int increment) throws IOException {
+        // Integer.MAX_VALUE is the idiomatic "unbounded" capacity request; forward it (and any
+        // non-positive value) unchanged so scaling does not turn "no limit" into a finite bound.
+        if (increment <= 0 || increment == Integer.MAX_VALUE) {
+            delegate.update(increment);
+            return;
+        }
+        // Request at least one byte so a small increment still lets the stream make progress.
+        delegate.update(Math.max(1, increment / expansionFactor));
+    }
+}

httpclient5/src/main/java/org/apache/hc/client5/http/async/methods/InflatingGzipDataConsumer.java

@@ -71,7 +71,7 @@ public InflatingGzipDataConsumer(final AsyncDataConsumer downstream) {
 
     @Override
     public void updateCapacity(final CapacityChannel c) throws IOException {
-        downstream.updateCapacity(c);
+        downstream.updateCapacity(new InflatingCapacityChannel(c));
     }
 
     @Override

httpclient5/src/main/java/org/apache/hc/client5/http/async/methods/InflatingZstdDataConsumer.java

@@ -81,7 +81,7 @@ public InflatingZstdDataConsumer(final AsyncDataConsumer downstream) {
 
     @Override
     public void updateCapacity(final CapacityChannel c) throws IOException {
-        downstream.updateCapacity(c);
+        downstream.updateCapacity(new InflatingCapacityChannel(c));
     }
 
     @Override

httpclient5/src/test/java/org/apache/hc/client5/http/async/methods/InflatingBrotliDataConsumerTest.java

@@ -34,6 +34,7 @@
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.LinkedHashMap;
@@ -191,6 +192,53 @@ public Set<String> getTrailerNames() {
         assertEquals(ORIGINAL, inner.getContent(), "br inflate mismatch");
     }
 
+    @Test
+    void consumeDoesNotOverrideDownstreamCapacity() throws Exception {
+        final byte[] compressed = brCompress();
+        final RecordingCapacityChannel reactor = new RecordingCapacityChannel();
+        final AsyncDataConsumer downstream = new AsyncDataConsumer() {
+            @Override
+            public void updateCapacity(final CapacityChannel ch) throws IOException {
+                ch.update(4096);
+            }
+
+            @Override
+            public void consume(final ByteBuffer src) {
+            }
+
+            @Override
+            public void streamEnd(final List<? extends Header> trailers) {
+            }
+
+            @Override
+            public void releaseResources() {
+            }
+        };
+        final InflatingBrotliDataConsumer inflating = new InflatingBrotliDataConsumer(downstream);
+
+        // The downstream advertises 4096 decompressed bytes, so the reactor must receive a
+        // scaled-down (4096 / 4 = 1024) increment, and consume() must never override it with
+        // Integer.MAX_VALUE as the previous implementation did.
+        inflating.updateCapacity(reactor);
+        for (int off = 0; off < compressed.length; off += 1024) {
+            final int n = Math.min(1024, compressed.length - off);
+            inflating.consume(ByteBuffer.wrap(compressed, off, n));
+        }
+
+        assertEquals(1, reactor.increments.size(), "consume() must not push extra capacity updates");
+        assertEquals(1024, reactor.increments.get(0).intValue());
+    }
+
+    private static final class RecordingCapacityChannel implements CapacityChannel {
+
+        final List<Integer> increments = new ArrayList<>();
+
+        @Override
+        public void update(final int increment) {
+            increments.add(increment);
+        }
+    }
+
     @Test
     void registerInExec() {
         final LinkedHashMap<String, UnaryOperator<AsyncDataConsumer>> map = new LinkedHashMap<>();

httpclient5/src/test/java/org/apache/hc/client5/http/async/methods/InflatingCapacityChannelTest.java

@@ -0,0 +1,112 @@
+/*
+ * ====================================================================
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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
+ *
+ *   http://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.
+ * ====================================================================
+ *
+ * This software consists of voluntary contributions made by many
+ * individuals on behalf of the Apache Software Foundation.  For more
+ * information on the Apache Software Foundation, please see
+ * <http://www.apache.org/>.
+ *
+ */
+package org.apache.hc.client5.http.async.methods;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.hc.core5.http.nio.CapacityChannel;
+import org.junit.jupiter.api.Test;
+
+class InflatingCapacityChannelTest {
+
+    private static final class RecordingCapacityChannel implements CapacityChannel {
+
+        final List<Integer> increments = new ArrayList<>();
+
+        @Override
+        public void update(final int increment) {
+            increments.add(increment);
+        }
+    }
+
+    @Test
+    void scalesDownByDefaultExpansionFactor() throws Exception {
+        final RecordingCapacityChannel delegate = new RecordingCapacityChannel();
+        final CapacityChannel channel = new InflatingCapacityChannel(delegate);
+
+        channel.update(400);
+
+        assertEquals(1, delegate.increments.size());
+        assertEquals(400 / InflatingCapacityChannel.DEFAULT_EXPANSION_FACTOR,
+                delegate.increments.get(0).intValue());
+    }
+
+    @Test
+    void scalesDownByExplicitExpansionFactor() throws Exception {
+        final RecordingCapacityChannel delegate = new RecordingCapacityChannel();
+        final CapacityChannel channel = new InflatingCapacityChannel(delegate, 8);
+
+        channel.update(800);
+
+        assertEquals(100, delegate.increments.get(0).intValue());
+    }
+
+    @Test
+    void smallIncrementStillRequestsAtLeastOneByte() throws Exception {
+        final RecordingCapacityChannel delegate = new RecordingCapacityChannel();
+        final CapacityChannel channel = new InflatingCapacityChannel(delegate, 4);
+
+        channel.update(2);
+
+        assertEquals(1, delegate.increments.get(0).intValue());
+    }
+
+    @Test
+    void nonPositiveIncrementIsForwardedUnchanged() throws Exception {
+        final RecordingCapacityChannel delegate = new RecordingCapacityChannel();
+        final CapacityChannel channel = new InflatingCapacityChannel(delegate, 4);
+
+        channel.update(0);
+
+        assertEquals(0, delegate.increments.get(0).intValue());
+    }
+
+    @Test
+    void unboundedCapacityIsForwardedUnchanged() throws Exception {
+        final RecordingCapacityChannel delegate = new RecordingCapacityChannel();
+        final CapacityChannel channel = new InflatingCapacityChannel(delegate, 4);
+
+        channel.update(Integer.MAX_VALUE);
+
+        assertEquals(Integer.MAX_VALUE, delegate.increments.get(0).intValue());
+    }
+
+    @Test
+    void rejectsNonPositiveExpansionFactor() {
+        assertThrows(IllegalArgumentException.class,
+                () -> new InflatingCapacityChannel(new RecordingCapacityChannel(), 0));
+    }
+
+    @Test
+    void rejectsNullDelegate() {
+        assertThrows(NullPointerException.class, () -> new InflatingCapacityChannel(null));
+    }
+}