Java 类io.reactivex.netty.protocol.http.client.HttpClientPipelineConfigurator 实例源码

@Override
public HttpClient<ByteBuf, ByteBuf> newHttpClient(final IClientConfig config) {
    final List<ExecutionListener<HttpClientRequest<ByteBuf>, HttpClientResponse<ByteBuf>>> listeners = new ArrayList<>();
    listeners.add(createBearerHeaderAdder());
    final PipelineConfiguratorComposite<HttpClientResponse<ByteBuf>, HttpClientRequest<ByteBuf>> pipelineConfigurator = new PipelineConfiguratorComposite<HttpClientResponse<ByteBuf>,
            HttpClientRequest<ByteBuf>>(new HttpClientPipelineConfigurator<ByteBuf, ByteBuf>(),
                                        new HttpObjectAggregationConfigurator(maxChunkSize));
    final LoadBalancingHttpClient<ByteBuf, ByteBuf> client = LoadBalancingHttpClient.<ByteBuf, ByteBuf>builder()
            .withClientConfig(config)
            .withExecutorListeners(listeners)
            .withRetryHandler(getDefaultHttpRetryHandlerWithConfig(config))
            .withPipelineConfigurator(pipelineConfigurator)
            .withPoolCleanerScheduler(RibbonTransport.poolCleanerScheduler)
            .build();

    return client;
}

@Override
public HttpClient<ByteBuf, ByteBuf> newHttpClient(final IClientConfig config) {
    final List<ExecutionListener<HttpClientRequest<ByteBuf>, HttpClientResponse<ByteBuf>>> listeners = new ArrayList<>();
    listeners.add(createBearerHeaderAdder());
    final PipelineConfiguratorComposite<HttpClientResponse<ByteBuf>, HttpClientRequest<ByteBuf>> pipelineConfigurator = new PipelineConfiguratorComposite<HttpClientResponse<ByteBuf>, 
            HttpClientRequest<ByteBuf>>(new HttpClientPipelineConfigurator<ByteBuf, ByteBuf>(),
            new HttpObjectAggregationConfigurator(maxChunkSize));
    final LoadBalancingHttpClient<ByteBuf, ByteBuf> client = LoadBalancingHttpClient.<ByteBuf, ByteBuf>builder()
            .withClientConfig(config)
            .withExecutorListeners(listeners)
            .withRetryHandler(getDefaultHttpRetryHandlerWithConfig(config))
            .withPipelineConfigurator(pipelineConfigurator)
            .withPoolCleanerScheduler(RibbonTransport.poolCleanerScheduler)
            .build();

    return client;
}

private LoadBalancingHttpClient<ByteBuf, ByteBuf> getClient(String vip) {
    LoadBalancingHttpClient<ByteBuf, ByteBuf> client = httpClients.get(vip);
    if (client == null) {
        IClientConfig config = IClientConfig.Builder.newBuilder("prana_backend").
                withDefaultValues().
                withDeploymentContextBasedVipAddresses(vip).
                build().
                set(IClientConfigKey.Keys.MaxTotalConnections, 2000).
                set(IClientConfigKey.Keys.MaxConnectionsPerHost, 2000).
                set(IClientConfigKey.Keys.OkToRetryOnAllOperations, false).
                set(IClientConfigKey.Keys.NIWSServerListClassName, DiscoveryEnabledNIWSServerList.class.getName());

        client = RibbonTransport.newHttpClient(new HttpClientPipelineConfigurator<ByteBuf, ByteBuf>(), config);
        httpClients.putIfAbsent(vip, client);

    }
    return client;
}

@Override
public void configureNewPipeline(ChannelPipeline pipeline) {
    httpClientPipelineConfigurator.configureNewPipeline(pipeline);
    if (null != pipeline.get(HttpClientPipelineConfigurator.REQUEST_RESPONSE_CONVERTER_HANDLER_NAME)) {
        pipeline.addBefore(HttpClientPipelineConfigurator.REQUEST_RESPONSE_CONVERTER_HANDLER_NAME,
                           SSEInboundHandler.NAME, SSEClientPipelineConfigurator.SSE_INBOUND_HANDLER);
    } else {
        // Assuming that the underlying HTTP configurator knows what its doing. It will mostly fail though.
        pipeline.addLast(SSEInboundHandler.NAME, SSEClientPipelineConfigurator.SSE_INBOUND_HANDLER);
    }
}

public SseOverHttpClientPipelineConfigurator() {
    this(new HttpClientPipelineConfigurator<I, Object>());
}

public SseOverHttpClientPipelineConfigurator(HttpClientPipelineConfigurator<I, ?> httpClientPipelineConfigurator) {
    this.httpClientPipelineConfigurator = httpClientPipelineConfigurator;
}

public static <I, O> PipelineConfigurator<HttpClientResponse<O>, HttpClientRequest<I>> httpClientConfigurator() {
    return new PipelineConfiguratorComposite<HttpClientResponse<O>, HttpClientRequest<I>>(new HttpClientPipelineConfigurator<I, O>(),
                                                                              new HttpObjectAggregationConfigurator());
}