Handle onTrailingHeadersReceived in AHC extras for RxJava 1 and 2 (AsyncHttpClient#1398)

Override default implementation and forward it to delegates, add tests.

Author: twz123

extras/rxjava/src/main/java/org/asynchttpclient/extras/rxjava/single/AbstractSingleSubscriberBridge.java

@@ -56,6 +56,11 @@ public State onHeadersReceived(HttpHeaders headers) throws Exception {
         return subscriber.isUnsubscribed() ? abort() : delegate().onHeadersReceived(headers);
     }
 
+    @Override
+    public State onTrailingHeadersReceived(HttpHeaders headers) throws Exception {
+        return subscriber.isUnsubscribed() ? abort() : delegate().onTrailingHeadersReceived(headers);
+    }
+
     @Override
     public Void onCompleted() {
         if (delegateTerminated.getAndSet(true)) {

extras/rxjava/src/test/java/org/asynchttpclient/extras/rxjava/single/AsyncHttpSingleTest.java

@@ -82,6 +82,9 @@ public void testSuccessfulCompletion() throws Exception {
                 bridge.onBodyPartReceived(null);
                 verify(handler).onBodyPartReceived(null);
 
+                bridge.onTrailingHeadersReceived(null);
+                verify(handler).onTrailingHeadersReceived(null);
+
                 bridge.onCompleted();
                 verify(handler).onCompleted();
             } catch (final Throwable t) {
@@ -135,6 +138,9 @@ public void testSuccessfulCompletionWithProgress() throws Exception {
                 progressBridge.onBodyPartReceived(null);
                 inOrder.verify(handler).onBodyPartReceived(null);
 
+                bridge.onTrailingHeadersReceived(null);
+                verify(handler).onTrailingHeadersReceived(null);
+
                 progressBridge.onCompleted();
                 inOrder.verify(handler).onCompleted();
             } catch (final Throwable t) {

extras/rxjava2/src/main/java/org/asynchttpclient/extras/rxjava2/maybe/AbstractMaybeAsyncHandlerBridge.java

@@ -81,6 +81,11 @@ public final State onHeadersReceived(HttpHeaders headers) throws Exception {
         return emitter.isDisposed() ? disposed() : delegate().onHeadersReceived(headers);
     }
 
+    @Override
+    public State onTrailingHeadersReceived(HttpHeaders headers) throws Exception {
+        return emitter.isDisposed() ? disposed() : delegate().onTrailingHeadersReceived(headers);
+    }
+
     /**
      * {@inheritDoc}
      *

extras/rxjava2/src/test/java/org/asynchttpclient/extras/rxjava2/maybe/AbstractMaybeAsyncHandlerBridgeTest.java

@@ -92,6 +92,9 @@ public void forwardsEvents() throws Exception {
         /* when */ underTest.onBodyPartReceived(bodyPart);
         then(delegate).should(times(2)).onBodyPartReceived(bodyPart);
 
+        /* when */ underTest.onTrailingHeadersReceived(headers);
+        then(delegate).should().onTrailingHeadersReceived(headers);
+
         /* when */ underTest.onCompleted();
         then(delegate).should().onCompleted();
         then(emitter).should().onSuccess(this);
@@ -199,6 +202,7 @@ public Object[][] httpEvents() {
                 { named("onStatusReceived", () -> underTest.onStatusReceived(status)) }, //
                 { named("onHeadersReceived", () -> underTest.onHeadersReceived(headers)) }, //
                 { named("onBodyPartReceived", () -> underTest.onBodyPartReceived(bodyPart)) }, //
+                { named("onTrailingHeadersReceived", () -> underTest.onTrailingHeadersReceived(headers)) }, //
         };
     }