Simplifying Aggregated Client Response usage

[NiteshKant]

Problem

HttpClientResponse follows the model that is suitable for streaming payloads i.e. HttpClientResponse.getContent() returns an Observable which can contain multiple elements. Since, this is the only way to use an HttpClientResponse, even in cases when the generated response is not streaming i.e. the content Observable produces a single element, the API does not simplify this use case.
The following is the typical interaction:

RxNetty.createHttpGet("http://localhost:8888/hello")
               .flatMap(new Func1<HttpClientResponse<ByteBuf>, Observable<ByteBuf>>() {
                   @Override
                   public Observable<ByteBuf> call(HttpClientResponse<ByteBuf> response) {
                       /// Interpret headers .....
                       return response.getContent();
                   }
               })
               .subscribe(new Action1<ByteBuf>() {
                   @Override
                   public void call(ByteBuf byteBuf) {
                       // handle Content
                   }
               });

Although the above works, it is a relatively cumbersome way of handling a non-streaming response.

Proposal

The following proposes a more specialized non-streaming interaction by defining a new response class (we can discuss an appropriate name):

public class FullHttpClientResponseHolder<T> {

    private final HttpClientResponse<T> response;
    private final T content;

    public FullHttpClientResponseHolder(HttpClientResponse<T> response, T content) {
        this.response = response;
        this.content = content;
    }

    public HttpClientResponse<T> getResponse() {
        return response;
    }

    public T getContent() {
        return content;
    }
}

and an operator:

public class FullResponseOperator<T>
        implements Observable.Operator<FullHttpClientResponseHolder<T>, HttpClientResponse<T>> {

    @Override
    public Subscriber<? super HttpClientResponse<T>> call(Subscriber<? super FullHttpClientResponseHolder<T>> child) {
        return new Subscriber<HttpClientResponse<T>>() {
            @Override
            public void onCompleted() {
                // Content complete propagates to the child subscriber.
            }

            @Override
            public void onError(Throwable e) {
                child.onError(e);
            }

            @Override
            public void onNext(HttpClientResponse<T> response) {
                response.getContent()
                        .take(1)
                        .map(new Func1<T, FullHttpClientResponseHolder<T>>() {
                            @Override
                            public FullHttpClientResponseHolder<T> call(T t) {
                                return new FullHttpClientResponseHolder<T>(response, t);
                            }
                        }).subscribe(child);
            }
        };
    }
}

Using which, the client can now look like:

        RxNetty.createHttpGet("http://localhost:8888/hello")
               .lift(new FullResponseOperator<ByteBuf>())
               .subscribe(new Action1<FullHttpClientResponseHolder<ByteBuf>>() {
                   @Override
                   public void call(FullHttpClientResponseHolder<ByteBuf> holder) {
                       holder.getResponse(); // handle headers
                       holder.getContent(); // handle content
                   }
               });

This approach is more convenient for non-streaming responses as it provides both content and headers at the time. It also does not add any overhead as the response is already aggregated by netty.

Possible issues

Since, the proposed operator disregards any content elements > 1, if this is used with a streaming response, it will terminate the stream after one onNext().
If we can name the operator in a way which is more intuitive to the users, this may not be that big an issue.