HTTPClient should discard a connection when response isn't completely consumed.

[NiteshKant]

The scenario when the response of an HTTP request is not consumed by the user, HTTPClient should dispose the underlying connection and not reuse it again for any other request.

Failure to do this will result in a subsequent request getting an older response (if arrived). If the older response never arrives, then the new request qualifies as a pipeline request and hence the response will never come (as the responses should be ordered and the first one never completed)

The following code demonstrates this problem:

final TestScheduler serverTestScheduler = Schedulers.test();
        final AtomicBoolean delayedResponse = new AtomicBoolean(true);
        RxNetty.newHttpServerBuilder(9999, new RequestHandler<ByteBuf, ByteBuf>() {
            @Override
            public Observable<Void> handle(HttpServerRequest<ByteBuf> request,
                                           final HttpServerResponse<ByteBuf> response) {
                if (!delayedResponse.get()) {
                    response.setStatus(HttpResponseStatus.NOT_FOUND);
                    return response.close(true);
                }
                return Observable.interval(1, TimeUnit.DAYS, serverTestScheduler)
                                 .flatMap(new Func1<Long, Observable<Void>>() {
                                     @Override
                                     public Observable<Void> call(Long aLong) {
                                         response.setStatus(HttpResponseStatus.NO_CONTENT);
                                         return response.close(true);
                                     }
                                 });
            }
        }).enableWireLogging(LogLevel.ERROR).build().start();


        HttpClientResponse<ByteBuf> response = RxNetty.createHttpClient("localhost", 9999)
                                                      .submit(HttpClientRequest.createGet(""))
                                                      .timeout(1, TimeUnit.SECONDS)
                                                      .doOnError(new Action1<Throwable>() {
                                                          @Override
                                                          public void call(Throwable throwable) {
                                                              System.out.println(">>>>>>>>>>>>>>");
                                                              throwable.printStackTrace();
                                                              serverTestScheduler.advanceTimeBy(1, TimeUnit.DAYS);
                                                              delayedResponse.set(false);
                                                          }
                                                      })
                                                      .retry(1).toBlocking().single();
        Assert.assertEquals(HttpResponseStatus.NOT_FOUND, response.getStatus());

Currently for read timeouts, we do dispose the connections but the same is not done for user initiated unsubscribes prior to response completion.

Streaming responses

How does this play with streaming responses?
For endless streams (eg: SSE) we should discard the connection, for protocols where the stream end may be well defined (Websockets) we should try to end the stream on unsubscribe.

[NiteshKant]

This was fixed as part of issue #250 for disposing connections if returned to pool (on unsubscribe) before response is completed. Issue #223 fixed this for SSE where the connections receiving are never reused.