+ * Purely async/streaming: no {@code InputStream}/{@code OutputStream}. Back-pressure is + * honored via {@link #available()} and the I/O reactor’s calls into {@link #produce(DataStreamChannel)}. + * Trailers from the upstream producer are preserved and emitted once the compressed output + * has been fully drained. + *
+ * + *+ * Ensure {@link com.aayushatharva.brotli4j.Brotli4jLoader#ensureAvailability()} has been + * called once at startup; this class also invokes it in a static initializer as a safeguard. + *
+ * + *{@code
+ * AsyncEntityProducer plain = new StringAsyncEntityProducer("hello", ContentType.TEXT_PLAIN);
+ * AsyncEntityProducer br = new DeflatingBrotliEntityProducer(plain); // defaults q=5, lgwin=22
+ * client.execute(new BasicRequestProducer(post, br),
+ * new BasicResponseConsumer<>(new StringAsyncEntityConsumer()),
+ * null);
+ * }
+ *
+ * @see org.apache.hc.core5.http.nio.AsyncEntityProducer
+ * @see org.apache.hc.core5.http.nio.DataStreamChannel
+ * @see com.aayushatharva.brotli4j.encoder.EncoderJNI
+ * @since 5.6
+ */
+public final class DeflatingBrotliEntityProducer implements AsyncEntityProducer {
+
+ private enum State { STREAMING, FINISHING, DONE }
+
+ private final AsyncEntityProducer upstream;
+ private final EncoderJNI.Wrapper encoder;
+
+ private ByteBuffer pendingOut;
+ private List pendingTrailers;
+ private State state = State.STREAMING;
+
+ /**
+ * Create a producer with explicit Brotli params.
+ *
+ * @param upstream upstream entity producer whose bytes will be compressed
+ * @param quality Brotli quality level (see brotli4j documentation)
+ * @param lgwin Brotli window size log2 (see brotli4j documentation)
+ * @param mode Brotli mode hint (GENERIC/TEXT/FONT)
+ * @throws IOException if the native encoder cannot be created
+ * @since 5.6
+ */
+ public DeflatingBrotliEntityProducer(
+ final AsyncEntityProducer upstream,
+ final int quality,
+ final int lgwin,
+ final Encoder.Mode mode) throws IOException {
+ this.upstream = Args.notNull(upstream, "upstream");
+ this.encoder = new EncoderJNI.Wrapper(256 * 1024, quality, lgwin, mode);
+ }
+
+ /**
+ * Convenience constructor mapping {@code 0=GENERIC, 1=TEXT, 2=FONT}.
+ *
+ * @since 5.6
+ */
+ public DeflatingBrotliEntityProducer(
+ final AsyncEntityProducer upstream,
+ final int quality,
+ final int lgwin,
+ final int modeInt) throws IOException {
+ this(upstream, quality, lgwin,
+ modeInt == 1 ? Encoder.Mode.TEXT :
+ modeInt == 2 ? Encoder.Mode.FONT : Encoder.Mode.GENERIC);
+ }
+
+ /**
+ * Create a producer with sensible defaults ({@code quality=5}, {@code lgwin=22}, {@code GENERIC}).
+ *
+ * @since 5.6
+ */
+ public DeflatingBrotliEntityProducer(final AsyncEntityProducer upstream) throws IOException {
+ this(upstream, 5, 22, Encoder.Mode.GENERIC);
+ }
+
+
+ @Override
+ public String getContentType() {
+ return upstream.getContentType();
+ }
+
+ @Override
+ public String getContentEncoding() {
+ return "br";
+ }
+
+ @Override
+ public long getContentLength() {
+ return -1;
+ }
+
+ @Override
+ public boolean isChunked() {
+ return true;
+ }
+
+ @Override
+ public Set+ * Purely async/streaming: no {@code InputStream}/{@code OutputStream}. Back-pressure from + * the I/O reactor is propagated via {@link CapacityChannel}. JNI output buffers are copied + * into small reusable direct {@link java.nio.ByteBuffer}s before handing them to the + * downstream consumer (which may retain them). + *
+ * + *+ * Ensure {@link com.aayushatharva.brotli4j.Brotli4jLoader#ensureAvailability()} has been + * called once at startup; this class also invokes it in a static initializer as a safeguard. + *
+ * + *{@code
+ * AsyncDataConsumer textConsumer = new StringAsyncEntityConsumer();
+ * AsyncDataConsumer brInflating = new InflatingBrotliDataConsumer(textConsumer);
+ * client.execute(producer, new BasicResponseConsumer<>(brInflating), null);
+ * }
+ *
+ * @see org.apache.hc.core5.http.nio.AsyncDataConsumer
+ * @see org.apache.hc.core5.http.nio.CapacityChannel
+ * @see com.aayushatharva.brotli4j.decoder.DecoderJNI
+ * @since 5.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) {
+ this.downstream = downstream;
+ try {
+ this.decoder = new DecoderJNI.Wrapper(8 * 1024);
+ } catch (final IOException e) {
+ throw new RuntimeException("Unable to initialize DecoderJNI", e);
+ }
+ }
+
+ @Override
+ public void updateCapacity(final CapacityChannel capacityChannel) throws IOException {
+ this.capacity = capacityChannel;
+ downstream.updateCapacity(capacityChannel);
+ }
+
+ @Override
+ public void consume(final ByteBuffer src) throws IOException {
+ while (src.hasRemaining()) {
+ final ByteBuffer in = decoder.getInputBuffer();
+ final int xfer = Math.min(src.remaining(), in.remaining());
+ if (xfer == 0) {
+ decoder.push(0);
+ pump();
+ continue;
+ }
+ final int lim = src.limit();
+ src.limit(src.position() + xfer);
+ in.put(src);
+ src.limit(lim);
+
+ decoder.push(xfer);
+ pump();
+ }
+ final CapacityChannel ch = this.capacity;
+ if (ch != null) {
+ ch.update(Integer.MAX_VALUE);
+ }
+ }
+
+ @Override
+ public void streamEnd(final List trailers) throws IOException, HttpException {
+ pump();
+ Asserts.check(decoder.getStatus() == DecoderJNI.Status.DONE || !decoder.hasOutput(),
+ "Truncated brotli stream");
+ downstream.streamEnd(trailers);
+ }
+
+ @Override
+ public void releaseResources() {
+ try {
+ decoder.destroy();
+ } catch (final Throwable ignore) {
+ }
+ downstream.releaseResources();
+ }
+
+ private void pump() throws IOException {
+ for (; ; ) {
+ switch (decoder.getStatus()) {
+ case OK:
+ decoder.push(0);
+ break;
+ case NEEDS_MORE_OUTPUT: {
+ // Pull a decoder-owned buffer; copy before handing off.
+ final ByteBuffer nativeBuf = decoder.pull();
+ if (nativeBuf != null && nativeBuf.hasRemaining()) {
+ final ByteBuffer copy = ByteBuffer.allocateDirect(nativeBuf.remaining());
+ copy.put(nativeBuf).flip();
+ downstream.consume(copy);
+ }
+ break;
+ }
+ case NEEDS_MORE_INPUT:
+ if (decoder.hasOutput()) {
+ final ByteBuffer nativeBuf = decoder.pull();
+ if (nativeBuf != null && nativeBuf.hasRemaining()) {
+ final ByteBuffer copy = ByteBuffer.allocateDirect(nativeBuf.remaining());
+ copy.put(nativeBuf).flip();
+ downstream.consume(copy);
+ break;
+ }
+ }
+ return; // wait for more input
+ case DONE:
+ if (decoder.hasOutput()) {
+ final ByteBuffer nativeBuf = decoder.pull();
+ if (nativeBuf != null && nativeBuf.hasRemaining()) {
+ final ByteBuffer copy = ByteBuffer.allocateDirect(nativeBuf.remaining());
+ copy.put(nativeBuf).flip();
+ downstream.consume(copy);
+ break;
+ }
+ }
+ return;
+ default:
+ // Corrupted stream
+ throw new IOException("Brotli stream corrupted");
+ }
+ }
+ }
+}
\ No newline at end of file
diff --git a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/Brotli4jRuntime.java b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/Brotli4jRuntime.java
new file mode 100644
index 0000000000..f60b1099f2
--- /dev/null
+++ b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/Brotli4jRuntime.java
@@ -0,0 +1,55 @@
+/*
+ * ====================================================================
+ * 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
+ * + * - Client sends a Brotli-compressed request body (Content-Encoding: br) + * - Server decompresses request, then responds with a Brotli-compressed body + * - Client checks the response Content-Encoding and decompresses if needed + *
+ * Notes:
+ * - Encoding uses brotli4j (native JNI); make sure matching native dependency is on the runtime classpath.
+ * - Decoding here uses Commons Compress via CompressorStreamFactory("br").
+ */
+public final class AsyncClientServerBrotliRoundTrip {
+
+ static {
+ Brotli4jLoader.ensureAvailability();
+ }
+
+ private static final String BR = "br";
+
+ public static void main(final String[] args) throws Exception {
+ final HttpServer server = ServerBootstrap.bootstrap()
+ .setListenerPort(0)
+ .setCanonicalHostName("localhost")
+ .register("/echo", new EchoHandler())
+ .create();
+ server.start();
+ final int port = server.getLocalPort();
+ final String url = "http://localhost:" + port + "/echo";
+
+ try (final CloseableHttpAsyncClient client = HttpAsyncClients.createDefault()) {
+ client.start();
+
+ final String requestBody = "Hello Brotli world (round-trip)!";
+ System.out.println("Request (plain): " + requestBody);
+
+ // --- client compresses request ---
+ final byte[] reqCompressed = brotliCompress(requestBody.getBytes(StandardCharsets.UTF_8));
+
+ final SimpleHttpRequest post = SimpleRequestBuilder.post(url)
+ .setHeader(HttpHeaders.CONTENT_TYPE, ContentType.TEXT_PLAIN.toString())
+ .setHeader(HttpHeaders.CONTENT_ENCODING, BR)
+ .build();
+
+ final Future