@Override
protected Future<?> execute() {
    InetSocketAddress destinationAddress;
    try {
        destinationAddress = addressFor(serverHostAndPort, proxyServer);
    } catch (UnknownHostException e) {
        return channel.newFailedFuture(e);
    }

    DefaultSocks4CommandRequest connectRequest = new DefaultSocks4CommandRequest(
            Socks4CommandType.CONNECT, destinationAddress.getHostString(), destinationAddress.getPort());

    addFirstOrReplaceHandler(SOCKS_ENCODER_NAME, Socks4ClientEncoder.INSTANCE);
    addFirstOrReplaceHandler(SOCKS_DECODER_NAME, new Socks4ClientDecoder());
    return writeToChannel(connectRequest);
}
 

@Test
void testProxy_connectionTimeoutFailure_throwsException() throws Exception {
    DYNAMIC_HANDLER.setChannelReadCustomizer((ctx, msg) -> {
        if (msg instanceof DefaultSocks4CommandRequest) {
            ctx.channel().eventLoop().schedule(
                    () -> ctx.fireChannelRead(msg), 50, TimeUnit.MILLISECONDS);
        } else {
            ctx.fireChannelRead(msg);
        }
    });

    final ClientFactory clientFactory = ClientFactory.builder().proxyConfig(
            ProxyConfig.socks4(socksProxyServer.address())).connectTimeoutMillis(1).build();

    final WebClient webClient = WebClient.builder(H1C, backendServer.httpEndpoint())
                                         .factory(clientFactory)
                                         .decorator(LoggingClient.newDecorator())
                                         .build();
    final CompletableFuture<AggregatedHttpResponse> responseFuture =
            webClient.get(PROXY_PATH).aggregate();
    assertThatThrownBy(responseFuture::join).isInstanceOf(CompletionException.class)
                                            .hasCauseInstanceOf(UnprocessedRequestException.class)
                                            .hasRootCauseInstanceOf(ProxyConnectException.class);
    clientFactory.close();
}
 

@Override
protected Object newInitialMessage(ChannelHandlerContext ctx) throws Exception {
    InetSocketAddress raddr = destinationAddress();
    String rhost;
    if (raddr.isUnresolved()) {
        rhost = raddr.getHostString();
    } else {
        rhost = raddr.getAddress().getHostAddress();
    }
    return new DefaultSocks4CommandRequest(
            Socks4CommandType.CONNECT, rhost, raddr.getPort(), username != null? username : "");
}
 

@Test
void testProxyWithUserName() throws Exception {
    final String username = "username";
    DYNAMIC_HANDLER.setChannelReadCustomizer((ctx, msg) -> {
        if (msg instanceof DefaultSocks4CommandRequest) {
            assertThat(username).isEqualTo(((DefaultSocks4CommandRequest) msg).userId());
        }
        ctx.fireChannelRead(msg);
    });

    final ClientFactory clientFactory =
            ClientFactory.builder()
                         .proxyConfig(ProxyConfig.socks4(socksProxyServer.address(), username))
                         .build();

    final WebClient webClient = WebClient.builder(H1C, backendServer.httpEndpoint())
                                         .factory(clientFactory)
                                         .decorator(LoggingClient.newDecorator())
                                         .build();
    final CompletableFuture<AggregatedHttpResponse> responseFuture =
            webClient.get(PROXY_PATH).aggregate();
    final AggregatedHttpResponse response = responseFuture.join();
    assertThat(response.status()).isEqualTo(OK);
    assertThat(response.contentUtf8()).isEqualTo(SUCCESS_RESPONSE);
    assertThat(numSuccessfulProxyRequests).isEqualTo(1);
    clientFactory.close();
}