Drop HttpResponseHeaders and add a default method to notify trailing headers, close #1397

Motivation:

HttpResponseHeaders brings an extra allocation and complexity for the
sheer sake of holding the information that those headers are trailing
ones. Trailing headers are actually a rare thing, we could just add an
extra method on AsyncHandler with a default implementation that would
noop.

Modifications:

* drop HttpResponseHeaders
* pass Netty’s HttpHeaders
* introduce AsyncHandler#onTrailingHeadersReceived

Result:

More simple API (no need to unwrap), less allocations

Author: slandelle

client/src/main/java/org/asynchttpclient/AsyncCompletionHandler.java

@@ -16,14 +16,15 @@
  */
 package org.asynchttpclient;
 
+import io.netty.handler.codec.http.HttpHeaders;
+
 import org.asynchttpclient.handler.ProgressAsyncHandler;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 /**
- * An {@link AsyncHandler} augmented with an {@link #onCompleted(Response)} convenience method which gets called
- * when the {@link Response} processing is finished.  This class also implement the {@link ProgressAsyncHandler} callback,
- * all doing nothing except returning {@link org.asynchttpclient.AsyncHandler.State#CONTINUE}
+ * An {@link AsyncHandler} augmented with an {@link #onCompleted(Response)} convenience method which gets called when the {@link Response} processing is finished. This class also
+ * implement the {@link ProgressAsyncHandler} callback, all doing nothing except returning {@link org.asynchttpclient.AsyncHandler.State#CONTINUE}
  *
  * @param <T> Type of the value that will be returned by the associated {@link java.util.concurrent.Future}
  */
@@ -32,41 +33,37 @@ public abstract class AsyncCompletionHandler<T> implements AsyncHandler<T>, Prog
     private static final Logger LOGGER = LoggerFactory.getLogger(AsyncCompletionHandler.class);
     private final Response.ResponseBuilder builder = new Response.ResponseBuilder();
 
-    /**
-     * {@inheritDoc}
-     */
-    public State onBodyPartReceived(final HttpResponseBodyPart content) throws Exception {
+    @Override
+    public State onBodyPartReceived(HttpResponseBodyPart content) throws Exception {
         builder.accumulate(content);
         return State.CONTINUE;
     }
 
-    /**
-     * {@inheritDoc}
-     */
-    public State onStatusReceived(final HttpResponseStatus status) throws Exception {
+    @Override
+    public State onStatusReceived(HttpResponseStatus status) throws Exception {
         builder.reset();
         builder.accumulate(status);
         return State.CONTINUE;
     }
 
-    /**
-     * {@inheritDoc}
-     */
-    public State onHeadersReceived(final HttpResponseHeaders headers) throws Exception {
+    @Override
+    public State onHeadersReceived(HttpHeaders headers) throws Exception {
         builder.accumulate(headers);
         return State.CONTINUE;
     }
 
-    /**
-     * {@inheritDoc}
-     */
+    @Override
+    public State onTrailingHeadersReceived(HttpHeaders headers) throws Exception {
+        builder.accumulate(headers);
+        return State.CONTINUE;
+    }
+
+    @Override
     public final T onCompleted() throws Exception {
         return onCompleted(builder.build());
     }
 
-    /**
-     * {@inheritDoc}
-     */
+    @Override
     public void onThrowable(Throwable t) {
         LOGGER.debug(t.getMessage(), t);
     }
@@ -85,28 +82,30 @@ public void onThrowable(Throwable t) {
      *
      * @return a {@link org.asynchttpclient.AsyncHandler.State} telling to CONTINUE or ABORT the current processing.
      */
+    @Override
     public State onHeadersWritten() {
         return State.CONTINUE;
     }
 
     /**
-     * Invoked when the content (a {@link java.io.File}, {@link String} or {@link java.io.InputStream} has been fully
-     * written on the I/O socket.
+     * Invoked when the content (a {@link java.io.File}, {@link String} or {@link java.io.InputStream} has been fully written on the I/O socket.
      *
      * @return a {@link org.asynchttpclient.AsyncHandler.State} telling to CONTINUE or ABORT the current processing.
      */
+    @Override
     public State onContentWritten() {
         return State.CONTINUE;
     }
 
     /**
      * Invoked when the I/O operation associated with the {@link Request} body as been progressed.
      *
-     * @param amount  The amount of bytes to transfer
+     * @param amount The amount of bytes to transfer
      * @param current The amount of bytes transferred
-     * @param total   The total number of bytes transferred
+     * @param total The total number of bytes transferred
      * @return a {@link org.asynchttpclient.AsyncHandler.State} telling to CONTINUE or ABORT the current processing.
      */
+    @Override
     public State onContentWriteProgress(long amount, long current, long total) {
         return State.CONTINUE;
     }

client/src/main/java/org/asynchttpclient/AsyncHandler.java

@@ -15,6 +15,8 @@
  */
 package org.asynchttpclient;
 
+import io.netty.handler.codec.http.HttpHeaders;
+
 
 /**
  * An asynchronous handler or callback which gets invoked as soon as some data is available when
@@ -89,14 +91,23 @@ enum State {
     State onStatusReceived(HttpResponseStatus responseStatus) throws Exception;
 
     /**
-     * Invoked as soon as the HTTP headers has been received. Can potentially be invoked more than once if a broken server
-     * sent trailing headers.
+     * Invoked as soon as the HTTP headers have been received.
      *
      * @param headers the HTTP headers.
      * @return a {@link State} telling to CONTINUE or ABORT the current processing.
      * @throws Exception if something wrong happens
      */
-    State onHeadersReceived(HttpResponseHeaders headers) throws Exception;
+    State onHeadersReceived(HttpHeaders headers) throws Exception;
+    
+    /**
+     * Invoked when trailing headers have been received. 
+     * @param headers
+     * @return
+     * @throws Exception
+     */
+    default State onTrailingHeadersReceived(HttpHeaders headers) throws Exception {
+        return State.CONTINUE;
+    }
 
     /**
      * Invoked once the HTTP response processing is finished.

client/src/main/java/org/asynchttpclient/HttpResponseHeaders.java

@@ -172,15 +172,15 @@ public interface Response {
     class ResponseBuilder {
         private final List<HttpResponseBodyPart> bodyParts = new ArrayList<>(1);
         private HttpResponseStatus status;
-        private HttpResponseHeaders headers;
+        private HttpHeaders headers;
 
         public ResponseBuilder accumulate(HttpResponseStatus status) {
             this.status = status;
             return this;
         }
 
-        public ResponseBuilder accumulate(HttpResponseHeaders headers) {
-            this.headers = this.headers == null ? headers : new HttpResponseHeaders(this.headers.getHeaders().add(headers.getHeaders()), true);
+        public ResponseBuilder accumulate(HttpHeaders headers) {
+            this.headers = this.headers == null ? headers : this.headers.add(headers);
             return this;
         }
 

client/src/main/java/org/asynchttpclient/Response.java

@@ -12,13 +12,14 @@
  */
 package org.asynchttpclient.filter;
 
+import io.netty.handler.codec.http.HttpHeaders;
+
+import java.io.IOException;
+
 import org.asynchttpclient.AsyncHandler;
-import org.asynchttpclient.HttpResponseHeaders;
 import org.asynchttpclient.HttpResponseStatus;
 import org.asynchttpclient.Request;
 
-import java.io.IOException;
-
 /**
  * A {@link FilterContext} can be used to decorate {@link Request} and {@link AsyncHandler} from a list of {@link RequestFilter}.
  * {@link RequestFilter} gets executed before the HTTP request is made to the remote server. Once the response bytes are
@@ -68,9 +69,9 @@ public HttpResponseStatus getResponseStatus() {
     }
 
     /**
-     * @return the response {@link HttpResponseHeaders}
+     * @return the response {@link HttpHeaders}
      */
-    public HttpResponseHeaders getResponseHeaders() {
+    public HttpHeaders getResponseHeaders() {
         return b.headers;
     }
 
@@ -94,7 +95,7 @@ public static class FilterContextBuilder<T> {
         private HttpResponseStatus responseStatus = null;
         private boolean replayRequest = false;
         private IOException ioException = null;
-        private HttpResponseHeaders headers;
+        private HttpHeaders headers;
 
         public FilterContextBuilder() {
         }
@@ -130,7 +131,7 @@ public FilterContextBuilder<T> responseStatus(HttpResponseStatus responseStatus)
             return this;
         }
 
-        public FilterContextBuilder<T> responseHeaders(HttpResponseHeaders headers) {
+        public FilterContextBuilder<T> responseHeaders(HttpHeaders headers) {
             this.headers = headers;
             return this;
         }

client/src/main/java/org/asynchttpclient/filter/FilterContext.java

@@ -12,6 +12,8 @@
  */
 package org.asynchttpclient.handler;
 
+import io.netty.handler.codec.http.HttpHeaders;
+
 import java.io.FilterInputStream;
 import java.io.IOException;
 import java.io.InputStream;
@@ -23,7 +25,6 @@
 
 import org.asynchttpclient.AsyncHandler;
 import org.asynchttpclient.HttpResponseBodyPart;
-import org.asynchttpclient.HttpResponseHeaders;
 import org.asynchttpclient.HttpResponseStatus;
 import org.asynchttpclient.Response;
 
@@ -98,6 +99,7 @@ public BodyDeferringAsyncHandler(final OutputStream os) {
         this.responseSet = false;
     }
 
+    @Override
     public void onThrowable(Throwable t) {
         this.throwable = t;
         // Counting down to handle error cases too.
@@ -121,17 +123,26 @@ public void onThrowable(Throwable t) {
         }
     }
 
+    @Override
     public State onStatusReceived(HttpResponseStatus responseStatus) throws Exception {
         responseBuilder.reset();
         responseBuilder.accumulate(responseStatus);
         return State.CONTINUE;
     }
 
-    public State onHeadersReceived(HttpResponseHeaders headers) throws Exception {
+    @Override
+    public State onHeadersReceived(HttpHeaders headers) throws Exception {
+        responseBuilder.accumulate(headers);
+        return State.CONTINUE;
+    }
+    
+    @Override
+    public State onTrailingHeadersReceived(HttpHeaders headers) throws Exception {
         responseBuilder.accumulate(headers);
         return State.CONTINUE;
     }
 
+    @Override
     public State onBodyPartReceived(HttpResponseBodyPart bodyPart) throws Exception {
         // body arrived, flush headers
         if (!responseSet) {
@@ -152,6 +163,7 @@ protected void closeOut() throws IOException {
         }
     }
 
+    @Override
     public Response onCompleted() throws IOException {
 
         if (!responseSet) {
@@ -242,6 +254,7 @@ public BodyDeferringInputStream(final Future<Response> future, final BodyDeferri
          * Closes the input stream, and "joins" (wait for complete execution
          * together with potential exception thrown) of the async request.
          */
+        @Override
         public void close() throws IOException {
             // close
             super.close();

client/src/main/java/org/asynchttpclient/handler/BodyDeferringAsyncHandler.java

@@ -18,7 +18,6 @@
 
 import org.asynchttpclient.AsyncCompletionHandlerBase;
 import org.asynchttpclient.HttpResponseBodyPart;
-import org.asynchttpclient.HttpResponseHeaders;
 import org.asynchttpclient.Response;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -97,8 +96,14 @@ public void headers(HttpHeaders headers) {
     }
 
     @Override
-    public State onHeadersReceived(final HttpResponseHeaders headers) throws Exception {
-        fireOnHeaderReceived(headers.getHeaders());
+    public State onHeadersReceived(final HttpHeaders headers) throws Exception {
+        fireOnHeaderReceived(headers);
+        return super.onHeadersReceived(headers);
+    }
+    
+    @Override
+    public State onTrailingHeadersReceived(HttpHeaders headers) throws Exception {
+        fireOnHeaderReceived(headers);
         return super.onHeadersReceived(headers);
     }