import org.asynchttpclient.AsyncHttpClientConfig; //导入依赖的package包/类
/**
 * 生成默认的httpclient config
 *
 * @return the config
 */
public static AsyncHttpClientConfig createConfig(int connectTimeout, int requestTimeout) {
    HashedWheelTimer timer = new HashedWheelTimer();
    timer.start();
    DefaultChannelPool channelPool = new DefaultChannelPool(60000,
            -1,
            DefaultChannelPool.PoolLeaseStrategy.LIFO,
            timer,
            3000);

    return new DefaultAsyncHttpClientConfig.Builder()
            .setConnectTimeout(connectTimeout)
            .setRequestTimeout(requestTimeout)
            .setMaxConnectionsPerHost(10000)
            .setValidateResponseHeaders(false)
            .setMaxRequestRetry(0)
            .setChannelPool(channelPool)
            .build();
}
 

import org.asynchttpclient.AsyncHttpClientConfig; //导入依赖的package包/类
public TimeoutsHolder(Timer nettyTimer, NettyResponseFuture<?> nettyResponseFuture, NettyRequestSender requestSender, AsyncHttpClientConfig config) {
    this.nettyTimer = nettyTimer;
    this.nettyResponseFuture = nettyResponseFuture;
    this.requestSender = requestSender;
    this.readTimeoutValue = config.getReadTimeout();

    int requestTimeoutInMs = nettyResponseFuture.getTargetRequest().getRequestTimeout();
    if (requestTimeoutInMs == 0) {
        requestTimeoutInMs = config.getRequestTimeout();
    }

    if (requestTimeoutInMs != -1) {
        requestTimeoutMillisTime = millisTime() + requestTimeoutInMs;
        requestTimeout = newTimeout(new RequestTimeoutTimerTask(nettyResponseFuture, requestSender, this, requestTimeoutInMs), requestTimeoutInMs);
    } else {
        requestTimeoutMillisTime = -1L;
        requestTimeout = null;
    }
}
 

import org.asynchttpclient.AsyncHttpClientConfig; //导入依赖的package包/类
@Test(groups = "standalone")
public void negativeContentTypeTest() throws Exception {

    AsyncHttpClientConfig config = config()//
            .setConnectTimeout(100)//
            .setMaxConnections(50)//
            .setRequestTimeout(5 * 60 * 1000) // 5 minutes
            .build();

    try (AsyncHttpClient client = asyncHttpClient(config)) {
        RequestBuilder requestBuilder = post(getTargetUrl())//
                .setHeader("Content-Type", "message/rfc822")//
                .setBody(new InputStreamBodyGenerator(new ByteArrayInputStream(MY_MESSAGE.getBytes())));

        Future<Response> future = client.executeRequest(requestBuilder.build());

        System.out.println("waiting for response");
        Response response = future.get();
        assertEquals(response.getStatusCode(), 200);
        assertEquals(response.getResponseBody(), MY_MESSAGE);
    }
}
 

import org.asynchttpclient.AsyncHttpClientConfig; //导入依赖的package包/类
/**
 * Create an instance from a given {@link BulletConfig}. Requires {@link DRPCConfig#DRPC_HTTP_CONNECT_TIMEOUT_MS},
 * {@link DRPCConfig#DRPC_HTTP_CONNECT_RETRY_LIMIT}, {@link DRPCConfig#DRPC_SERVERS}, {@link DRPCConfig#DRPC_HTTP_PORT},
 * and {@link DRPCConfig#DRPC_HTTP_PATH} to be set. To be used in PubSub.Context#QUERY_SUBMISSION mode.
 *
 * @param config A non-null config that contains the required settings.
 */
public DRPCQueryResultPubscriber(BulletConfig config) {
    Objects.requireNonNull(config);

    Number connectTimeout = config.getRequiredConfigAs(DRPCConfig.DRPC_HTTP_CONNECT_TIMEOUT_MS, Number.class);
    Number retryLimit = config.getRequiredConfigAs(DRPCConfig.DRPC_HTTP_CONNECT_RETRY_LIMIT, Number.class);
    List<String> servers = (List<String>) config.getRequiredConfigAs(DRPCConfig.DRPC_SERVERS, List.class);
    String protocol = config.getRequiredConfigAs(DRPCConfig.DRPC_HTTP_PROTOCOL, String.class);
    String port = config.getRequiredConfigAs(DRPCConfig.DRPC_HTTP_PORT, String.class);
    String path = config.getRequiredConfigAs(DRPCConfig.DRPC_HTTP_PATH, String.class);
    String function = config.getRequiredConfigAs(DRPCConfig.DRPC_FUNCTION, String.class);

    List<String> endpoints = servers.stream()
                                    .map(url -> String.format(URL_TEMPLATE, protocol, url, port, path, function))
                                    .collect(Collectors.toList());
    this.urls = new RandomPool<>(endpoints);
    AsyncHttpClientConfig clientConfig = new DefaultAsyncHttpClientConfig.Builder()
                                                .setConnectTimeout(connectTimeout.intValue())
                                                .setMaxRequestRetry(retryLimit.intValue())
                                                .setReadTimeout(NO_TIMEOUT)
                                                .setRequestTimeout(NO_TIMEOUT)
                                                .build();
    // This is thread safe
    client = new DefaultAsyncHttpClient(clientConfig);
    responses = new ConcurrentLinkedQueue<>();
}
 

import org.asynchttpclient.AsyncHttpClientConfig; //导入依赖的package包/类
/**
 * Initialize the client config.
 *
 * @param requestTimeout  Timeout to use for the client configuration.
 *
 * @return the set up client
 */
private static AsyncHttpClient initializeWebClient(int requestTimeout) {

    LOG.debug("Druid request timeout: {}ms", requestTimeout);

    // Build the configuration
    AsyncHttpClientConfig config = new DefaultAsyncHttpClientConfig.Builder()
            .setReadTimeout(requestTimeout)
            .setRequestTimeout(requestTimeout)
            .setConnectTimeout(requestTimeout)
            .setConnectionTtl(requestTimeout)
            .setPooledConnectionIdleTimeout(requestTimeout)
            .setFollowRedirect(true)
            .build();

    return new DefaultAsyncHttpClient(config);
}
 

import org.asynchttpclient.AsyncHttpClientConfig; //导入依赖的package包/类
@Test
public void testAcceptWebSocket() {
    TestServer server = testServer(19001);
    running(server, () -> {
        try {
            AsyncHttpClientConfig config = new DefaultAsyncHttpClientConfig.Builder().setMaxRequestRetry(0).build();
            AsyncHttpClient client = new DefaultAsyncHttpClient(config);
            WebSocketClient webSocketClient = new WebSocketClient(client);

            try {
                String serverURL = "ws://localhost:19001/ws";
                WebSocketClient.LoggingListener listener = new WebSocketClient.LoggingListener();
                CompletableFuture<WebSocket> completionStage = webSocketClient.call(serverURL, listener);

                await().until(() -> {
                    assertThat(completionStage).isDone();
                });
            } finally {
                //noinspection ThrowFromFinallyBlock
                client.close();
            }
        } catch (Exception e) {
            fail("Unexpected exception", e);
        }
    });
}
 

import org.asynchttpclient.AsyncHttpClientConfig; //导入依赖的package包/类
/**
 * 为service创建Resty配置
 *
 * @param serviceClz the service clz
 */
public void initContextForService(Class serviceClz) {
    RestyService restyService = (RestyService) serviceClz.getDeclaredAnnotation(RestyService.class);
    if (restyService == null) {
        return;
    }
    String serviceName = restyService.serviceName();

    // class->@RestyService
    this.storeRestyService(serviceClz, restyService);
    AsyncHttpClientConfig httpClientConfig = AsyncHttpConfigFactory.createConfig(restyService.connectTimeout(), restyService.requestTimeout());
    HttpClientWrapper clientHolder = new HttpClientWrapper(httpClientConfig);
    httpClientPool.putIfAbsent(serviceName, clientHolder);


    SpringAnnotationWrapper wrapper = new SpringAnnotationWrapper();

    RestyCommandConfig commandProperties = new RestyCommandConfig.DefaultRestyCommandConfig();
    processRestyServiceAnnotation(restyService, commandProperties);


    for (Method method : serviceClz.getMethods()) {
        //存储 httpMethod 和 restyCommandConfig
        RestyMethod restyMethod = method.getDeclaredAnnotation(RestyMethod.class);
        if (restyMethod != null) {
            storeRestyMethod(method, restyMethod);
            processRestyMethodAnnotation(restyMethod, commandProperties);
        }
        commandPropertiesMap.putIfAbsent(method, commandProperties);

        // 存储 httpMethod 和 requestTemplate
        RestyRequestTemplate restyRequestTemplate = wrapper.processAnnotation(serviceClz, method);
        requestTemplateMap.putIfAbsent(method, restyRequestTemplate);

        serviceMethodTable.put(serviceName, restyRequestTemplate.getPath(), method);
    }
    log.info("RestyCommandContext初始化成功！");
}
 

import org.asynchttpclient.AsyncHttpClientConfig; //导入依赖的package包/类
private Bootstrap newBootstrap(Class<? extends Channel> socketChannelClass, EventLoopGroup eventLoopGroup, AsyncHttpClientConfig config) {
    @SuppressWarnings("deprecation")
    Bootstrap bootstrap = new Bootstrap().channel(socketChannelClass).group(eventLoopGroup)//
            // default to PooledByteBufAllocator
            .option(ChannelOption.ALLOCATOR, config.isUsePooledMemory() ? PooledByteBufAllocator.DEFAULT : UnpooledByteBufAllocator.DEFAULT)//
            .option(ChannelOption.TCP_NODELAY, config.isTcpNoDelay())//
            .option(ChannelOption.SO_REUSEADDR, config.isSoReuseAddress())//
            .option(ChannelOption.AUTO_CLOSE, false);

    if (config.getConnectTimeout() > 0) {
        bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, config.getConnectTimeout());
    }

    if (config.getSoLinger() >= 0) {
        bootstrap.option(ChannelOption.SO_LINGER, config.getSoLinger());
    }

    if (config.getSoSndBuf() >= 0) {
        bootstrap.option(ChannelOption.SO_SNDBUF, config.getSoSndBuf());
    }

    if (config.getSoRcvBuf() >= 0) {
        bootstrap.option(ChannelOption.SO_RCVBUF, config.getSoRcvBuf());
    }

    for (Entry<ChannelOption<Object>, Object> entry : config.getChannelOptions().entrySet()) {
        bootstrap.option(entry.getKey(), entry.getValue());
    }

    return bootstrap;
}
 

import org.asynchttpclient.AsyncHttpClientConfig; //导入依赖的package包/类
public NettyFileBody(File file, long offset, long length, AsyncHttpClientConfig config) {
    if (!file.isFile()) {
        throw new IllegalArgumentException(String.format("File %s is not a file or doesn't exist", file.getAbsolutePath()));
    }
    this.file = file;
    this.offset = offset;
    this.length = length;
    this.config = config;
}
 

import org.asynchttpclient.AsyncHttpClientConfig; //导入依赖的package包/类
public NettyChannelConnector(InetAddress localAddress,//
        List<InetSocketAddress> remoteAddresses,//
        AsyncHandler<?> asyncHandler,//
        TimeoutsHolder timeoutsHolder,//
        AtomicBoolean closed,//
        AsyncHttpClientConfig config) {
    this.localAddress = localAddress != null ? new InetSocketAddress(localAddress, 0) : null;
    this.remoteAddresses = remoteAddresses;
    this.asyncHandlerExtensions = toAsyncHandlerExtensions(asyncHandler);
    this.timeoutsHolder = assertNotNull(timeoutsHolder, "timeoutsHolder");
    this.closed = closed;
    this.connectionTtlEnabled = config.getConnectionTtl() > 0;
}
 

import org.asynchttpclient.AsyncHttpClientConfig; //导入依赖的package包/类
public NettyRequestSender(AsyncHttpClientConfig config,//
        ChannelManager channelManager,//
        Timer nettyTimer,//
        AtomicBoolean closed) {
    this.config = config;
    this.channelManager = channelManager;
    this.nettyTimer = nettyTimer;
    this.closed = closed;
    requestFactory = new NettyRequestFactory(config);
}
 

import org.asynchttpclient.AsyncHttpClientConfig; //导入依赖的package包/类
public NettyResponseStatus(Uri uri, AsyncHttpClientConfig config, HttpResponse response, Channel channel) {
    super(uri, config);
    this.response = response;
    if (channel != null) {
        remoteAddress = channel.remoteAddress();
        localAddress = channel.localAddress();
    } else {
        remoteAddress = null;
        localAddress = null;
    }
}
 

import org.asynchttpclient.AsyncHttpClientConfig; //导入依赖的package包/类
public Interceptors(//
        AsyncHttpClientConfig config,//
        ChannelManager channelManager,//
        NettyRequestSender requestSender) {
    this.config = config;
    unauthorized401Interceptor = new Unauthorized401Interceptor(channelManager, requestSender);
    proxyUnauthorized407Interceptor = new ProxyUnauthorized407Interceptor(channelManager, requestSender);
    continue100Interceptor = new Continue100Interceptor(requestSender);
    redirect30xInterceptor = new Redirect30xInterceptor(channelManager, config, requestSender);
    connectSuccessInterceptor = new ConnectSuccessInterceptor(channelManager, requestSender);
    responseFiltersInterceptor = new ResponseFiltersInterceptor(config, requestSender);
    hasResponseFilters = !config.getResponseFilters().isEmpty();
}
 

import org.asynchttpclient.AsyncHttpClientConfig; //导入依赖的package包/类
public AsyncHttpClientHandler(AsyncHttpClientConfig config,//
        ChannelManager channelManager,//
        NettyRequestSender requestSender) {
    this.config = config;
    this.channelManager = channelManager;
    this.requestSender = requestSender;
    interceptors = new Interceptors(config, channelManager, requestSender);
    hasIOExceptionFilters = !config.getIoExceptionFilters().isEmpty();
}
 

import org.asynchttpclient.AsyncHttpClientConfig; //导入依赖的package包/类
protected void configureSslEngine(SSLEngine sslEngine, AsyncHttpClientConfig config) {
    sslEngine.setUseClientMode(true);
    if (!config.isAcceptAnyCertificate()) {
        SSLParameters params = sslEngine.getSSLParameters();
        params.setEndpointIdentificationAlgorithm("HTTPS");
        sslEngine.setSSLParameters(params);
    }

    if (isNonEmpty(config.getEnabledProtocols()))
        sslEngine.setEnabledProtocols(config.getEnabledProtocols());

    if (isNonEmpty(config.getEnabledCipherSuites()))
        sslEngine.setEnabledCipherSuites(config.getEnabledCipherSuites());
}
 

import org.asynchttpclient.AsyncHttpClientConfig; //导入依赖的package包/类
private SslContext getSslContext(AsyncHttpClientConfig config) throws SSLException {
    if (config.getSslContext() != null)
        return config.getSslContext();

    SslContextBuilder sslContextBuilder = SslContextBuilder.forClient()//
            .sslProvider(config.isUseOpenSsl() ? SslProvider.OPENSSL : SslProvider.JDK)//
            .sessionCacheSize(config.getSslSessionCacheSize())//
            .sessionTimeout(config.getSslSessionTimeout());

    if (config.isAcceptAnyCertificate())
        sslContextBuilder.trustManager(InsecureTrustManagerFactory.INSTANCE);

    return sslContextBuilder.build();
}
 

import org.asynchttpclient.AsyncHttpClientConfig; //导入依赖的package包/类
@Override
public SSLEngine newSslEngine(AsyncHttpClientConfig config, String peerHost, int peerPort) {
    // FIXME should be using ctx allocator
    SSLEngine sslEngine = sslContext.newEngine(ByteBufAllocator.DEFAULT, peerHost, peerPort);
    configureSslEngine(sslEngine, config);
    return sslEngine;
}
 

import org.asynchttpclient.AsyncHttpClientConfig; //导入依赖的package包/类
/**
 * @param config the global config
 * @param request the request
 * @return the proxy server to be used for this request (can be null)
 */
public static ProxyServer getProxyServer(AsyncHttpClientConfig config, Request request) {
    ProxyServer proxyServer = request.getProxyServer();
    if (proxyServer == null) {
        ProxyServerSelector selector = config.getProxyServerSelector();
        if (selector != null) {
            proxyServer = selector.select(request.getUri());
        }
    }
    return proxyServer != null && !proxyServer.isIgnoredForHost(request.getUri().getHost()) ? proxyServer : null;
}
 

import org.asynchttpclient.AsyncHttpClientConfig; //导入依赖的package包/类
/**
 * Tests that a head request can be made
 * 
 * @throws IOException
 * @throws ExecutionException
 * @throws InterruptedException
 */
@Test(groups = "standalone")
public void testRetryNonBlocking() throws IOException, InterruptedException, ExecutionException {

    AsyncHttpClientConfig config = config()//
            .setKeepAlive(true)//
            .setMaxConnections(100)//
            .setConnectTimeout(60000)//
            .setRequestTimeout(30000)//
            .build();

    try (AsyncHttpClient client = asyncHttpClient(config)) {
        List<ListenableFuture<Response>> res = new ArrayList<>();
        for (int i = 0; i < 32; i++) {
            res.add(testMethodRequest(client, 3, "servlet", UUID.randomUUID().toString()));
        }

        StringBuilder b = new StringBuilder();
        for (ListenableFuture<Response> r : res) {
            Response theres = r.get();
            assertEquals(200, theres.getStatusCode());
            b.append("==============\r\n");
            b.append("Response Headers\r\n");
            HttpHeaders heads = theres.getHeaders();
            b.append(heads + "\r\n");
            b.append("==============\r\n");
        }
        System.out.println(b.toString());
        System.out.flush();
    }
}
 

import org.asynchttpclient.AsyncHttpClientConfig; //导入依赖的package包/类
@Test(groups = "standalone")
public void testRetryNonBlockingAsyncConnect() throws IOException, InterruptedException, ExecutionException {

    AsyncHttpClientConfig config = config()//
            .setKeepAlive(true)//
            .setMaxConnections(100)//
            .setConnectTimeout(60000)//
            .setRequestTimeout(30000)//
            .build();

    try (AsyncHttpClient client = asyncHttpClient(config)) {
        List<ListenableFuture<Response>> res = new ArrayList<>();
        for (int i = 0; i < 32; i++) {
            res.add(testMethodRequest(client, 3, "servlet", UUID.randomUUID().toString()));
        }

        StringBuilder b = new StringBuilder();
        for (ListenableFuture<Response> r : res) {
            Response theres = r.get();
            assertEquals(theres.getStatusCode(), 200);
            b.append("==============\r\n");
            b.append("Response Headers\r\n");
            HttpHeaders heads = theres.getHeaders();
            b.append(heads + "\r\n");
            b.append("==============\r\n");
        }
        System.out.println(b.toString());
        System.out.flush();
    }
}