WebSocket client fails silently on handshake failure

[edwinchoi]

The issue is that WebSocketClient.HandshakeOperator does not check for failure on handshake completion.

The following code reproduces the issue:

final RxServer<HttpServerRequest<ByteBuf>, HttpServerResponse<ByteBuf>> server = RxNetty.createHttpServer(0, new RequestHandler<ByteBuf, ByteBuf>() {
    @Override
    public Observable<Void> handle(HttpServerRequest<ByteBuf> request, HttpServerResponse<ByteBuf> response) {
        response.setStatus(HttpResponseStatus.BAD_REQUEST);
        return Observable.empty();
    }
});
server.start();

final LinkedBlockingQueue<Notification<Throwable>> q = new LinkedBlockingQueue<Notification<Throwable>>();
RxNetty.newWebSocketClientBuilder("localhost", server.getServerPort())
        .withWebSocketVersion(WebSocketVersion.V13)
        .enableWireLogging(LogLevel.ERROR)
        .build()
        .connect()
        .timeout(3, java.util.concurrent.TimeUnit.SECONDS)
        .subscribe(new Subscriber<ObservableConnection<WebSocketFrame, WebSocketFrame>>() {
            private void put(Notification<? extends Throwable> notif) {
                try {
                    q.put((Notification<Throwable>) notif);
                } catch(InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
            public void onNext(ObservableConnection<WebSocketFrame, WebSocketFrame> con) {
                put(Notification.createOnNext(new IllegalStateException("Connection should not have succeeded")));
            }
            public void onCompleted() {
                put(Notification.<Throwable>createOnCompleted());
            }
            public void onError(Throwable cause) {
                put(Notification.createOnNext(cause));
            }
        });
final Notification<Throwable> n = q.poll(5, TimeUnit.SECONDS);
assert n != null;
assert n.isOnNext();
assert n.getValue() instanceof WebSocketHandshakeException;
server.shutdown();

Changing the handshake listener (in WebSocketClient.HandshakeOperator) to the following resolves the issue:

handler.addHandshakeFinishedListener(new ChannelFutureListener() {
    @Override
    public void operationComplete(ChannelFuture future) throws Exception {
        if (future.isCancelled()) {
            originalSubscriber.unsubscribe(); // should this be unsubscribe, or onCompleted?
        } else if (future.isSuccess()) {
            originalSubscriber.onNext(connection);
            originalSubscriber.onCompleted();
        } else {
            originalSubscriber.onError(future.cause());
        }
    }
});

[NiteshKant]

@edwinchoi Thanks for reporting the issue. Looks like you have figured the solution, do you wish to send a PR?

[edwinchoi]

@NiteshKant created a PR.

[NiteshKant]

Fixed via #453