HTTPCLIENT-1822: async transparent content decompression

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

@@ -0,0 +1,239 @@
+/*
+ * ====================================================================
+ * 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 java.nio.ByteBuffer;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.zip.Deflater;
+
+import org.apache.hc.core5.http.Header;
+import org.apache.hc.core5.http.nio.AsyncEntityProducer;
+import org.apache.hc.core5.http.nio.DataStreamChannel;
+import org.apache.hc.core5.util.Args;
+
+/**
+ * {@code AsyncEntityProducer} that streams the output of another producer
+ * through the raw DEFLATE compression algorithm.
+ *
+ * <p>The delegate’s bytes are read in small chunks, compressed with
+ * {@link java.util.zip.Deflater} and written immediately to the HTTP I/O
+ * layer.  Memory use is therefore bounded even for very large request
+ * entities.</p>
+ *
+ * @since 5.6
+ */
+public final class DeflatingAsyncEntityProducer implements AsyncEntityProducer {
+
+    /**
+     * inbound copy‐buffer
+     */
+    private static final int IN_BUF = 8 * 1024;
+    /**
+     * outbound staging buffer
+     */
+    private static final int OUT_BUF = 8 * 1024;
+
+    private final AsyncEntityProducer delegate;
+    private final String contentType;
+    private final Deflater deflater = new Deflater(Deflater.DEFAULT_COMPRESSION, /*nowrap=*/true);
+
+    /**
+     * holds compressed bytes not yet sent downstream
+     */
+    private final ByteBuffer pending = ByteBuffer.allocate(OUT_BUF);
+    private final byte[] in = new byte[IN_BUF];
+
+    private final AtomicBoolean delegateEnded = new AtomicBoolean(false);
+    private boolean finished = false;
+
+    public DeflatingAsyncEntityProducer(final AsyncEntityProducer delegate) {
+        this.delegate = Args.notNull(delegate, "delegate");
+        this.contentType = delegate.getContentType();
+        /* place pending into “read-mode” with no data */
+        pending.flip();
+    }
+
+    // ------------------------------------------------------------------ metadata
+
+    @Override
+    public boolean isRepeatable() {
+        return false;
+    }
+
+    @Override
+    public long getContentLength() {
+        return -1;
+    }     // unknown
+
+    @Override
+    public String getContentType() {
+        return contentType;
+    }
+
+    @Override
+    public String getContentEncoding() {
+        return "deflate";
+    }
+
+    @Override
+    public boolean isChunked() {
+        return true;
+    }
+
+    @Override
+    public Set<String> getTrailerNames() {
+        return Collections.emptySet();
+    }
+
+    @Override
+    public int available() {
+        if (pending.hasRemaining()) {
+            return pending.remaining();
+        }
+        return delegate.available();
+    }
+
+    // ------------------------------------------------------------------ core
+
+    @Override
+    public void produce(final DataStreamChannel channel) throws IOException {
+        /* 1 — flush any leftover compressed bytes first */
+        if (flushPending(channel)) {
+            return;  // back-pressure: outer channel could not accept more
+        }
+
+        /* 2 — pull more data from delegate */
+        delegate.produce(new InnerChannel(channel));
+
+        /* 3 — if delegate ended, finish the deflater */
+        if (delegateEnded.get() && !finished) {
+            deflater.finish();
+            deflateToPending();
+            flushPending(channel);
+            if (!pending.hasRemaining()) {
+                finished = true;
+                channel.endStream();
+            }
+        }
+    }
+
+    /**
+     * copy as much as possible from {@link #pending} to the wire
+     */
+    private boolean flushPending(final DataStreamChannel ch) throws IOException {
+        while (pending.hasRemaining()) {
+            final int written = ch.write(pending);
+            if (written == 0) {
+                return true; // back-pressure
+            }
+        }
+        pending.clear().flip();              // no data left → empty read-mode
+        return false;
+    }
+
+    /**
+     * drain {@link #deflater} into {@link #pending}
+     */
+    private void deflateToPending() {
+        /* switch pending to write-mode */
+        pending.compact();
+        final byte[] out = pending.array();
+        int total;
+        do {
+            total = deflater.deflate(out, pending.position(), pending.remaining(),
+                    Deflater.NO_FLUSH);
+            pending.position(pending.position() + total);
+            if (!pending.hasRemaining() && total > 0) {
+                /* buffer full: grow to the next power of two */
+                final ByteBuffer bigger = ByteBuffer.allocate(pending.capacity() * 2);
+                pending.flip();
+                bigger.put(pending);
+                pending.clear();
+                pending.put(bigger);
+            }
+        } while (total > 0);
+        pending.flip(); // back to read-mode
+    }
+
+    // ------------------------------------------------------------------ inner channel that receives raw bytes
+
+    private final class InnerChannel implements DataStreamChannel {
+        private final DataStreamChannel outer;
+
+        InnerChannel(final DataStreamChannel outer) {
+            this.outer = outer;
+        }
+
+        @Override
+        public void requestOutput() {
+            outer.requestOutput();
+        }
+
+        @Override
+        public int write(final ByteBuffer src) throws IOException {
+            int consumed = 0;
+            while (src.hasRemaining()) {
+                final int chunk = Math.min(src.remaining(), in.length);
+                src.get(in, 0, chunk);
+                deflater.setInput(in, 0, chunk);
+                consumed += chunk;
+                deflateToPending();
+                if (flushPending(outer)) {    // honour back-pressure
+                    break;
+                }
+            }
+            return consumed;
+        }
+
+        @Override
+        public void endStream() {
+            delegateEnded.set(true);
+        }
+
+        @Override
+        public void endStream(final List<? extends Header> trailers) {
+            endStream();
+        }
+    }
+
+    // ------------------------------------------------------------------ error / cleanup
+
+    @Override
+    public void failed(final Exception cause) {
+        delegate.failed(cause);
+    }
+
+    @Override
+    public void releaseResources() {
+        delegate.releaseResources();
+        deflater.end();
+    }
+}

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

@@ -0,0 +1,129 @@
+/*
+ * ====================================================================
+ * 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 java.nio.ByteBuffer;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.zip.DataFormatException;
+import java.util.zip.Inflater;
+
+import org.apache.hc.core5.http.Header;
+import org.apache.hc.core5.http.HttpException;
+import org.apache.hc.core5.http.nio.AsyncDataConsumer;
+import org.apache.hc.core5.http.nio.CapacityChannel;
+
+/**
+ * <p>Streaming, non-blocking {@link AsyncDataConsumer} that transparently
+ * inflates a response whose {@code Content-Encoding} is {@code deflate}.
+ * The decoded bytes are pushed straight to the wrapped downstream consumer
+ * while honouring reactor back-pressure.</p>
+ *
+ * <p>The implementation understands both formats that exist “in the wild”: the
+ * raw DEFLATE stream (RFC 1951) and the zlib-wrapped variant (RFC 1950).
+ * If the caller does not specify which one to expect, the first two bytes of
+ * the stream are inspected and the proper decoder is chosen automatically.</p>
+ *
+ * <p>No {@code InputStream}/{@code OutputStream} buffering is used; memory
+ * footprint is bounded and suitable for very large payloads.</p>
+ *
+ * @since 5.6
+ */
+public final class InflatingAsyncDataConsumer implements AsyncDataConsumer {
+
+    private final AsyncDataConsumer downstream;
+    private final Boolean nowrapHint;
+    private Inflater inflater;
+    private boolean formatChosen;
+    private final byte[] out = new byte[8 * 1024];
+    private final AtomicBoolean closed = new AtomicBoolean(false);
+
+    public InflatingAsyncDataConsumer(
+            final AsyncDataConsumer downstream, final Boolean nowrapHint) {
+        this.downstream = downstream;
+        this.nowrapHint = nowrapHint;
+        this.inflater = new Inflater(nowrapHint == null || nowrapHint);
+    }
+
+    @Override
+    public void updateCapacity(final CapacityChannel ch) throws IOException {
+        downstream.updateCapacity(ch);
+    }
+
+    @Override
+    public void consume(final ByteBuffer src) throws IOException {
+        if (closed.get()) {
+            return;
+        }
+
+        if (nowrapHint == null && !formatChosen && src.remaining() >= 2) {
+            src.mark();
+            final int b0 = src.get() & 0xFF;
+            final int b1 = src.get() & 0xFF;
+            src.reset();
+            final boolean zlib = b0 == 0x78 &&
+                    (b1 == 0x01 || b1 == 0x5E || b1 == 0x9C || b1 == 0xDA);
+            if (zlib) {
+                inflater.end();
+                inflater = new Inflater(false);
+            }
+            formatChosen = true;
+        }
+
+        final byte[] in = new byte[src.remaining()];
+        src.get(in);
+        inflater.setInput(in);
+
+        try {
+            int n;
+            while ((n = inflater.inflate(out)) > 0) {
+                downstream.consume(ByteBuffer.wrap(out, 0, n));
+            }
+            if (inflater.needsDictionary()) {
+                throw new IOException("Deflate dictionary required");
+            }
+        } catch (final DataFormatException ex) {
+            throw new IOException("Corrupt DEFLATE stream", ex);
+        }
+    }
+
+    @Override
+    public void streamEnd(final List<? extends Header> trailers)
+            throws HttpException, IOException {
+        if (closed.compareAndSet(false, true)) {
+            inflater.end();
+            downstream.streamEnd(trailers);
+        }
+    }
+
+    @Override
+    public void releaseResources() {
+        inflater = null;
+        downstream.releaseResources();
+    }
+}