io.netty.handler.codec.socksx.v4.DefaultSocks4CommandResponse #io.netty.handler.proxy.ProxyConnectException源码实例Demo

private void failAndReset(Throwable cause) {
    if (cause instanceof ProxyConnectException) {
        // ProxyConnectException is handled by HttpSessionHandler.exceptionCaught().
        return;
    }

    fail(cause);

    final Http2Error error;
    if (Exceptions.isStreamCancelling(cause)) {
        error = Http2Error.CANCEL;
    } else {
        error = Http2Error.INTERNAL_ERROR;
    }

    if (error.code() != Http2Error.CANCEL.code()) {
        Exceptions.logIfUnexpected(logger, ch,
                                   HttpSession.get(ch).protocol(),
                                   "a request publisher raised an exception", cause);
    }

    if (ch.isActive()) {
        encoder.writeReset(id, streamId(), error);
        ch.flush();
    }
}
 

@Test
void testProxy_protocolUpgrade_notSharableExceptionNotThrown() throws Exception {
    DYNAMIC_HANDLER.setWriteCustomizer((ctx, msg, promise) -> {
        ctx.write(new DefaultSocks4CommandResponse(Socks4CommandStatus.REJECTED_OR_FAILED), promise);
    });
    final ClientFactory clientFactory = ClientFactory.builder().proxyConfig(
            ProxyConfig.socks4(socksProxyServer.address())).build();
    final WebClient webClient = WebClient.builder(SessionProtocol.HTTP, 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();
}
 

@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();
}
 

@Test
void testProxy_responseFailure_throwsException() throws Exception {
    DYNAMIC_HANDLER.setWriteCustomizer((ctx, msg, promise) -> {
        ctx.write(new DefaultSocks4CommandResponse(Socks4CommandStatus.REJECTED_OR_FAILED), promise);
    });
    final ClientFactory clientFactory = ClientFactory.builder().proxyConfig(
            ProxyConfig.socks4(socksProxyServer.address())).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();
}
 

@Test
void testProxyServerImmediateClose() throws Exception {
    DYNAMIC_HANDLER.setChannelReadCustomizer((ctx, msg) -> {
        ctx.close();
    });
    try (ClientFactory clientFactory = ClientFactory.builder().proxyConfig(
            ProxyConfig.socks4(socksProxyServer.address())).build()) {
        final WebClient webClient = WebClient.builder(H1C, backendServer.httpEndpoint())
                                             .factory(clientFactory)
                                             .decorator(LoggingClient.newDecorator())
                                             .build();
        final CompletableFuture<AggregatedHttpResponse> responseFuture =
                webClient.get(PROXY_PATH).aggregate();
        await().timeout(Duration.ofSeconds(10)).until(responseFuture::isCompletedExceptionally);
        assertThatThrownBy(responseFuture::join).isInstanceOf(CompletionException.class)
                                                .hasCauseInstanceOf(UnprocessedRequestException.class)
                                                .hasRootCauseInstanceOf(ProxyConnectException.class);
    }
}
 

@Override
protected void exceptionCaught(Throwable cause) {
    try {
        if (cause instanceof ProxyConnectException) {
            LOG.info("A ProxyConnectException occurred on ProxyToServerConnection: " + cause.getMessage());
            connectionFlow.fail(cause);
        } else if (cause instanceof IOException) {
            // IOExceptions are expected errors, for example when a server drops the connection. rather than flood
            // the logs with stack traces for these expected exceptions, log the message at the INFO level and the
            // stack trace at the DEBUG level.
            LOG.info("An IOException occurred on ProxyToServerConnection: " + cause.getMessage());
            LOG.debug("An IOException occurred on ProxyToServerConnection", cause);
        } else if (cause instanceof RejectedExecutionException) {
            LOG.info("An executor rejected a read or write operation on the ProxyToServerConnection (this is normal if the proxy is shutting down). Message: " + cause.getMessage());
            LOG.debug("A RejectedExecutionException occurred on ProxyToServerConnection", cause);
        } else {
            LOG.error("Caught an exception on ProxyToServerConnection", cause);
        }
    } finally {
        if (!is(DISCONNECTED)) {
            LOG.info("Disconnecting open connection to server");
            disconnect();
        }
    }
    // This can happen if we couldn't make the initial connection due
    // to something like an unresolved address, for example, or a timeout.
    // There will not have been be any requests written on an unopened
    // connection, so there should not be any further action to take here.
}
 

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    if (cause instanceof ProxyConnectException) {
        sessionPromise.tryFailure(UnprocessedRequestException.of(cause));
        return;
    }
    setPendingException(ctx, new ClosedSessionException(cause));
    if (!(cause instanceof IOException)) {
        ctx.close();
    } else {
        // Netty will close the connection automatically on an IOException.
    }
}