HTTP Redirect fails when there is a connection pool

[allenxwang]

The following test, which is slightly modified from the existing one in HttpClientTest, fails:

    @Test
    public void testRedirect() {
        HttpClientConfig.Builder builder = new HttpClientConfig.Builder(null);
        HttpClientConfig config = builder.readTimeout(20000, TimeUnit.MILLISECONDS)
                .build();
        HttpClient<ByteBuf, ByteBuf> client = new HttpClientBuilder<ByteBuf, ByteBuf>("localhost", port)
                .config(config)
                .withMaxConnections(2) // this line is added 
                .build();
        String content = invokeBlockingCall(client, "test/redirect?port=" + port);
        assertEquals("Hello world", content);
    }

The test fails with the following stack trace, which means there is no invocation of the onNext() to the blocking observable.

java.util.NoSuchElementException: Sequence contains no elements
    at rx.operators.OperatorSingle$1.onCompleted(OperatorSingle.java:78)
    at rx.observers.SerializedObserver.onCompleted(SerializedObserver.java:96)
    at rx.observers.SerializedSubscriber.onCompleted(SerializedSubscriber.java:44)
    at rx.operators.OperatorMerge$1.onCompleted(OperatorMerge.java:52)
    at rx.operators.OperatorMap$1.onCompleted(OperatorMap.java:43)
    at io.reactivex.netty.protocol.http.client.RedirectOperator$RedirectSubscriber.onCompleted(RedirectOperator.java:208)
    at io.reactivex.netty.protocol.http.client.RedirectOperator$RedirectSubscriber.onCompleted(RedirectOperator.java:208)
    at io.reactivex.netty.protocol.http.client.RequestProcessingOperator$1$2$1.onCompleted(RequestProcessingOperator.java:75)
    at rx.Observable$29.onCompleted(Observable.java:6081)
    at rx.observers.SafeSubscriber.onCompleted(SafeSubscriber.java:74)
    at rx.subjects.SubjectSubscriptionManager$SubjectObserver.onCompleted(SubjectSubscriptionManager.java:247)
    at rx.Notification.accept(Notification.java:116)
    at rx.subjects.PublishSubject$2.call(PublishSubject.java:81)
    at rx.subjects.PublishSubject$2.call(PublishSubject.java:73)
    at rx.subjects.SubjectSubscriptionManager$1.call(SubjectSubscriptionManager.java:106)
    at rx.subjects.SubjectSubscriptionManager$1.call(SubjectSubscriptionManager.java:47)
    at rx.Observable.subscribe(Observable.java:6177)
    at rx.Observable.subscribe(Observable.java:6077)
    at io.reactivex.netty.protocol.http.client.RequestProcessingOperator$1$2.call(RequestProcessingOperator.java:72)
    at io.reactivex.netty.protocol.http.client.RequestProcessingOperator$1$2.call(RequestProcessingOperator.java:1)
    at rx.Observable$12.onNext(Observable.java:3682)
    at rx.operators.OperatorDoOnEach$1.onNext(OperatorDoOnEach.java:61)
    at rx.subjects.SubjectSubscriptionManager$SubjectObserver.onNext(SubjectSubscriptionManager.java:261)
    at rx.subjects.PublishSubject.onNext(PublishSubject.java:131)
    at io.reactivex.netty.pipeline.ObservableAdapter.channelRead(ObservableAdapter.java:40)
    at io.netty.channel.DefaultChannelHandlerContext.invokeChannelRead(DefaultChannelHandlerContext.java:338)
    at io.netty.channel.DefaultChannelHandlerContext.fireChannelRead(DefaultChannelHandlerContext.java:324)
    at io.netty.channel.ChannelInboundHandlerAdapter.channelRead(ChannelInboundHandlerAdapter.java:86)
    at io.netty.channel.DefaultChannelHandlerContext.invokeChannelRead(DefaultChannelHandlerContext.java:338)
    at io.netty.channel.DefaultChannelHandlerContext.fireChannelRead(DefaultChannelHandlerContext.java:324)
    at io.netty.channel.ChannelInboundHandlerAdapter.channelRead(ChannelInboundHandlerAdapter.java:86)
    at io.reactivex.netty.protocol.http.client.ClientRequestResponseConverter.channelRead(ClientRequestResponseConverter.java:94)
    at io.netty.channel.DefaultChannelHandlerContext.invokeChannelRead(DefaultChannelHandlerContext.java:338)
    at io.netty.channel.DefaultChannelHandlerContext.fireChannelRead(DefaultChannelHandlerContext.java:324)
    at io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:103)
    at io.netty.channel.DefaultChannelHandlerContext.invokeChannelRead(DefaultChannelHandlerContext.java:338)
    at io.netty.channel.DefaultChannelHandlerContext.fireChannelRead(DefaultChannelHandlerContext.java:324)
    at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:153)
    at io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:148)
    at io.netty.channel.DefaultChannelHandlerContext.invokeChannelRead(DefaultChannelHandlerContext.java:338)
    at io.netty.channel.DefaultChannelHandlerContext.fireChannelRead(DefaultChannelHandlerContext.java:324)
    at io.netty.handler.timeout.ReadTimeoutHandler.channelRead(ReadTimeoutHandler.java:149)
    at io.netty.channel.DefaultChannelHandlerContext.invokeChannelRead(DefaultChannelHandlerContext.java:338)
    at io.netty.channel.DefaultChannelHandlerContext.fireChannelRead(DefaultChannelHandlerContext.java:324)
    at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:785)
    at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:126)
    at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:485)
    at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:452)
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:346)
    at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:101)
    at java.lang.Thread.run(Thread.java:695)

[NiteshKant]

The issue here was that as part of sending (onNext()) the first response (redirect response), we were submitting the redirect request.
If the connection is reused (for connection pool), it resets the content subject (reacting to ConnectionReuseEvent) in ClientRequestResponseConverter.
The ClientRequestResponseConverter was referring to the "current content subject" while invoking an onComplete(), which is invoked when the content on the first response is over.
Now, in the above situation, since the content subject is changed between processing headers and processing content (in the same method), the onComplete() is called on the content subject of the second response and not the first as it is expected.

The fix is two folds:

1. Move the redirect call in RedirectOperator to onComplete()/onError() of the first observer.
2. In ClientRequestResponseConverter use the subject that was present when the channelRead() was received and do not refer to the instance level state of the subject.

From the perspective of this bug, 1) is sufficient but 2) is the correct way of handling the shared state.