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

@Test
public void testConcurrentPatch() throws InterruptedException {
    HttpClientRequest<ByteBuf, ByteBuf> post = post(100L);

    Observable<byte[]> slowContent = Observable.just("hello ".getBytes()).repeat()
            .zipWith(Observable.interval(50, TimeUnit.MILLISECONDS).startWith(0L), (data, nop) -> data).take(10);
    Observable<byte[]> fastContent = Observable.just("goodbye ".getBytes()).repeat()
            .zipWith(Observable.interval(10, TimeUnit.MILLISECONDS).startWith(0L), (data, nop) -> data).take(10);

    Iterator<HttpClientResponse<ByteBuf>> iterator = post.map(this::getLocation)
            .flatMap(location -> Observable.merge(
                    patch(location, 0 , slowContent),
                    patch(location, 0, fastContent).delay(120, TimeUnit.MILLISECONDS)))
            .toBlocking().getIterator();

    // the first response should be the failure
    assertThat(iterator.next()).isNotNull()
            .extracting(HttpClientResponse::getStatus).containsExactly(HttpResponseStatus.BAD_REQUEST);

    // the second one should be sucessfull
    assertThat(iterator.next()).isNotNull()
            .extracting(HttpClientResponse::getStatus).containsExactly(HttpResponseStatus.NO_CONTENT);
}

protected static Observable<JsonNode> buildEventStream(String buildId, String lastEventId) {
    AtomicReference<String> _lastBuildEventId = new AtomicReference<>(null);

    HttpClientRequest<ByteBuf, ByteBuf> request = HTTP_CLIENT
            .createGet("/build-export/v1/build/" + buildId + "/events?eventTypes=" + EventProcessor.EVENT_TYPES)
            .setKeepAlive(true);
    if (BASIC_AUTH != null) {
        request = request.addHeader("Authorization", "Basic " + BASIC_AUTH);
    }

    if (lastEventId != null) {
        request = request.addHeader("Last-Event-ID", lastEventId);
    }

    return request
            .flatMap(HttpClientResponse::getContentAsServerSentEvents)
            .doOnNext(serverSentEvent -> _lastBuildEventId.set(serverSentEvent.getEventIdAsString()))
            .doOnSubscribe(() -> LOGGER.info("Streaming events for build: " + buildId))
            .filter(serverSentEvent -> serverSentEvent.getEventTypeAsString().equals("BuildEvent"))
            .map(Application::parse)
            .onErrorResumeNext(t -> {
                LOGGER.info("Error streaming build events of build " + buildId + ", resuming from event id" + _lastBuildEventId.get() + "...");
                return buildEventStream(buildId, _lastBuildEventId.get());
            });
}

@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 Observable<JsonNode> get(String url)
{
    return rxNetty.submit(withHeaders(HttpClientRequest.create(HttpMethod.GET, configuration.getPath() + url)))
            .flatMap(clientResponse ->
            {
                if (clientResponse.getStatus().code() == 200)
                {
                    return clientResponse.getContent();
                }
                else
                {
                    logger.error("Got http {} from zmon", clientResponse.getStatus().code());
                    return Observable.empty();
                }
            })
            .flatMap(this::parse);
}

@Override
public void apply(HttpClientRequest<ByteBuf> request, Authentication auth) {
    logger.info("OAuth2TokenStrategy called for auth: " + auth + " and request: " + request);
    if (auth != null) {
        if (auth instanceof OAuth2Authentication) {
            Object details = auth.getDetails();
            if (details instanceof OAuth2AuthenticationDetails) {
                logger.info("OAuth2 authentication details found");
                OAuth2AuthenticationDetails oauth = (OAuth2AuthenticationDetails) details;
                String accessToken = oauth.getTokenValue();
                String tokenType = oauth.getTokenType() == null ? "Bearer" : oauth.getTokenType();
                request.withHeader("Authorization", tokenType + " " + accessToken);
            } else {
                logger.info("No OAuth2 authentication details found");
            }
        }
    } else {
        logger.warn("OAuth2TokenStrategy enabled, but inbound request does not contain a Spring Security Authentication context");
    }
}

/**
 * Calls the remote service using the provided request, applies error handling and
 * converts the response into a {@link Map}. The entire request and response are
 * executed and handled in a hot observable.
 *
 * @param serviceId the id of the service for which the request is made
 * @param request   the request which has to be executed using RxNetty
 * @return an {@link Observable} emitting the JSON response as a Map with String keys
 * and Object values.
 */
public Observable<Map<String, Object>> retrieveJsonFromRequest(String serviceId, HttpClientRequest<ByteBuf> request) {
    RxClient.ServerInfo serverInfo = getServerInfoFromRequestOrClient(request, rxClient);

    return rxClient.submit(serverInfo, request)
            .publish().autoConnect()
            .doOnError(el -> errorHandler.handleNodeError(serviceId, format("Error retrieving node(s) for url {0} with headers {1}: {2}",
                    request.getUri(), request.getHeaders().entries(), el), el))
            .filter(r -> {
                if (r.getStatus().code() < 400) {
                    return true;
                } else {
                    errorHandler.handleNodeWarning(serviceId, "Exception " + r.getStatus() + " for url " + request.getUri() + " with headers " + r.getHeaders().entries());
                    return false;
                }
            })
            .flatMap(AbstractHttpContentHolder::getContent)
            .map(data -> data.toString(Charset.defaultCharset()))
            .map(response -> {
                JacksonJsonParser jsonParser = new JacksonJsonParser();
                return jsonParser.parseMap(response);
            })
            .doOnNext(r -> logger.info("Json retrieved from call: {}", r))
            .onErrorResumeNext(Observable.empty());
}

private Observable<Node> getIndexFromServiceInstance(ServiceInstance serviceInstance, final Object outboundSecurityObject) {
    final String url = uriResolver.resolveHomePageUrl(serviceInstance);
    final String serviceId = serviceInstance.getServiceId().toLowerCase();
    HttpClientRequest<ByteBuf> request = HttpClientRequest.createGet(url);
    applyOutboundSecurityStrategyOnRequest(request, outboundSecurityObject);
    for (Map.Entry<String, String> header : properties.getRequestHeaders().entrySet()) {
        request.withHeader(header.getKey(), header.getValue());
    }

    return caller.retrieveJsonFromRequest(serviceId, request)
            .map(JSONObject::new)
            .concatMap(source -> indexToNodeConverter.convert(serviceInstance.getServiceId().toLowerCase(), url, source))
            .filter(node -> !properties.getFilteredServices().contains(node.getId()))
            .doOnNext(el -> logger.info("Index node {} discovered in url: {}", el.getId(), url))
            .doOnError(e -> logger.error("Error while fetching node: ", e))
            .doOnCompleted(() -> logger.info("Completed emissions of an index node observable for url: " + url))
            .onErrorResumeNext(Observable.empty());
}

protected Observable<Node> getHealthNodesFromService(String serviceId, String url, final Object outboundSecurityObject) {
    HttpClientRequest<ByteBuf> request = HttpClientRequest.createGet(url);
    applyOutboundSecurityStrategyOnRequest(request, outboundSecurityObject);
    for (Entry<String, String> header : properties.getRequestHeaders().entrySet()) {
        request.withHeader(header.getKey(), header.getValue());
    }

    return caller.retrieveJsonFromRequest(serviceId, request)
            .flatMap(el -> healthToNodeConverter.convertToNodes(serviceId, el))
            .filter(node -> !properties.getFilteredServices().contains(node.getId()))
            //TODO: .map(node -> springCloudEnricher.enrich(node))
            .doOnNext(el -> logger.info("Health node {} discovered in url: {}", el.getId(), url))
            .doOnError(e -> logger.error("Error during healthnode fetching: ", e))
            .doOnCompleted(() -> logger.info("Completed emission of a health node observable from url: " + url))
            .onErrorResumeNext(Observable.empty());
}

@SuppressWarnings("unchecked")
@Test
public void getPactUrlsNotFound() throws InterruptedException {

    HttpClientResponse<ByteBuf> urlsNotFoundResponse = mock(HttpClientResponse.class);
    when(urlsNotFoundResponse.getContent()).thenReturn(null);
    when(urlsNotFoundResponse.getStatus()).thenReturn(HttpResponseStatus.NOT_FOUND);
    HttpResponseHeaders httpResponseHeaders = mock(HttpResponseHeaders.class);
    when(httpResponseHeaders.entries()).thenReturn(newArrayList());
    when(urlsNotFoundResponse.getHeaders()).thenReturn(httpResponseHeaders);

    when(rxClient.submit(any(RxClient.ServerInfo.class), any(HttpClientRequest.class)))
            .thenReturn(Observable.just(urlsNotFoundResponse));

    TestSubscriber<Node> testSubscriber = new TestSubscriber<>();
    pactsAggregator.aggregateNodes().toBlocking().subscribe(testSubscriber);
    testSubscriber.assertNoErrors();

    List<Node> nodes = testSubscriber.getOnNextEvents();
    assertThat(nodes).isEmpty();

    verify(publisher).publishEvent(any(SystemEvent.class));
}

@SuppressWarnings("unchecked")
@Test
public void onErrorWhenGettingNodeOne() {
    HttpClientResponse<ByteBuf> urlsResponse = mock(HttpClientResponse.class);
    ByteBuf byteBuf = (new PooledByteBufAllocator()).directBuffer();
    ByteBufUtil.writeUtf8(byteBuf, onePactSource);
    when(urlsResponse.getContent()).thenReturn(Observable.just(byteBuf));
    when(urlsResponse.getStatus()).thenReturn(HttpResponseStatus.OK);

    when(rxClient.submit(any(RxClient.ServerInfo.class), any(HttpClientRequest.class)))
            .thenReturn(Observable.just(urlsResponse), Observable.error(new RuntimeException()));

    TestSubscriber<Node> testSubscriber = new TestSubscriber<>();
    pactsAggregator.aggregateNodes().toBlocking().subscribe(testSubscriber);
    testSubscriber.assertError(RuntimeException.class);

    verify(publisher).publishEvent(any(SystemEvent.class));
}

@SuppressWarnings({"rawtypes", "unchecked"})
@Test
public void shouldGetMappingNodesFromService() {
    when(properties.getRequestHeaders()).thenReturn(requestHeaders());
    Map retrievedMap = new HashMap();
    Observable retrievedMapObservable = Observable.just(retrievedMap);
    when(caller.retrieveJsonFromRequest(anyString(), any(HttpClientRequest.class)))
            .thenReturn(retrievedMapObservable);
    mockStatic(MappingsToNodeConverter.class);
    PowerMockito.when(MappingsToNodeConverter.convertToNodes(anyString(), anyMap()))
            .thenReturn(Observable.from(correctNodes()));

    TestSubscriber<Node> testSubscriber = new TestSubscriber<>();
    aggregator.getMappingNodesFromService("testService", "testUrl", null).toBlocking().subscribe(testSubscriber);
    List<Node> nodes = testSubscriber.getOnNextEvents();

    verify(caller, times(1)).retrieveJsonFromRequest(eq("testService"), requestCaptor.capture());
    assertThat(requestCaptor.getValue().getHeaders().entries()).usingElementComparator(stringEntryComparator())
            .containsExactlyElementsOf(requestHeaders().entrySet());
    assertThat(nodes).containsOnly(new Node("Node1"), new Node("Node2"));
}

@SuppressWarnings({"rawtypes", "unchecked"})
@Test
public void shouldReturnEmptyObservableOnEmptySourceObservable() {
    when(properties.getRequestHeaders()).thenReturn(requestHeaders());
    Observable retrievedMapObservable = Observable.empty();
    when(caller.retrieveJsonFromRequest(anyString(), any(HttpClientRequest.class)))
            .thenReturn(retrievedMapObservable);

    TestSubscriber<Node> testSubscriber = new TestSubscriber<>();
    aggregator.getMappingNodesFromService("testService", "testUrl", null).toBlocking().subscribe(testSubscriber);
    testSubscriber.assertNoValues();
    testSubscriber.assertCompleted();

    verify(caller, times(1)).retrieveJsonFromRequest(eq("testService"), requestCaptor.capture());
    assertThat(requestCaptor.getValue().getHeaders().entries()).usingElementComparator(stringEntryComparator())
            .containsExactlyElementsOf(requestHeaders().entrySet());
}

@SuppressWarnings({"rawtypes", "unchecked"})
@Test
public void shouldReturnEmptyObservableOnErroneousConversion() {
    when(properties.getRequestHeaders()).thenReturn(requestHeaders());
    Map retrievedMap = new HashMap();
    Observable retrievedMapObservable = Observable.just(retrievedMap).publish().autoConnect();
    when(caller.retrieveJsonFromRequest(anyString(), any(HttpClientRequest.class)))
            .thenReturn(retrievedMapObservable);
    mockStatic(MappingsToNodeConverter.class);
    PowerMockito.when(MappingsToNodeConverter.convertToNodes(anyString(), anyMap()))
            .thenThrow(new RuntimeException("Error1"));

    TestSubscriber<Node> testSubscriber = new TestSubscriber<>();
    aggregator.getMappingNodesFromService("testService", "testUrl", null).toBlocking().subscribe(testSubscriber);
    testSubscriber.getOnNextEvents();
    testSubscriber.assertNoValues();
    testSubscriber.assertCompleted();

    verify(caller, times(1)).retrieveJsonFromRequest(eq("testService"), requestCaptor.capture());
    assertThat(requestCaptor.getValue().getHeaders().entries()).usingElementComparator(stringEntryComparator())
            .containsExactlyElementsOf(requestHeaders().entrySet());
}

@Test
@SuppressWarnings("unchecked")
public void emptyObservableFromIndexCallShouldReturnZeroNodes() throws InterruptedException {
    when(discoveryClient.getServices()).thenReturn(Collections.singletonList("service"));
    ServiceInstance instance = mock(ServiceInstance.class);
    when(discoveryClient.getInstances("service")).thenReturn(Collections.singletonList(instance));

    when(instance.getServiceId()).thenReturn("service");
    when(instance.getUri()).thenReturn(URI.create("http://localhost:8089/service"));

    when(caller.retrieveJsonFromRequest(eq("service"), any(HttpClientRequest.class))).thenReturn(Observable.empty());

    TestSubscriber<Node> testSubscriber = new TestSubscriber<>();
    indexesAggregator.aggregateNodes().toBlocking().subscribe(testSubscriber);

    List<Node> nodes = testSubscriber.getOnNextEvents();
    assertThat(nodes).hasSize(0);
}

@Test
public void shouldGetHealthNodesFromService() {
    when(properties.getRequestHeaders()).thenReturn(requestHeaders());
    Map retrievedMap = new HashMap();
    Observable retrievedMapObservable = Observable.just(retrievedMap);
    when(caller.retrieveJsonFromRequest(anyString(), any(HttpClientRequest.class)))
            .thenReturn(retrievedMapObservable);
    when(converter.convertToNodes(anyString(), anyMap()))
            .thenReturn(Observable.from(correctNodes()));

    TestSubscriber<Node> testSubscriber = new TestSubscriber<>();
    aggregator.getHealthNodesFromService("testService", "testUrl", null).toBlocking().subscribe(testSubscriber);
    List<Node> nodes = testSubscriber.getOnNextEvents();

    verify(caller, times(1)).retrieveJsonFromRequest(eq("testService"), requestCaptor.capture());
    assertThat(requestCaptor.getValue().getHeaders().entries()).usingElementComparator(stringEntryComparator())
            .containsExactlyElementsOf(requestHeaders().entrySet());
    assertThat(nodes).containsOnly(new Node("Node1"), new Node("Node2"));
}

@Test
public void shouldReturnEmptyObservableOnEmptySourceObservable() {
    when(properties.getRequestHeaders()).thenReturn(requestHeaders());
    Observable retrievedMapObservable = Observable.empty();
    when(caller.retrieveJsonFromRequest(anyString(), any(HttpClientRequest.class)))
            .thenReturn(retrievedMapObservable);

    TestSubscriber<Node> testSubscriber = new TestSubscriber<>();
    aggregator.getHealthNodesFromService("testService", "testUrl", null).toBlocking().subscribe(testSubscriber);
    testSubscriber.assertNoValues();
    testSubscriber.assertCompleted();

    verify(caller, times(1)).retrieveJsonFromRequest(eq("testService"), requestCaptor.capture());
    assertThat(requestCaptor.getValue().getHeaders().entries()).usingElementComparator(stringEntryComparator())
            .containsExactlyElementsOf(requestHeaders().entrySet());
}

@Test
public void shouldReturnEmptyObservableOnErroneousSourceObservable() {
    when(properties.getRequestHeaders()).thenReturn(requestHeaders());
    Map retrievedMap = new HashMap();
    Observable retrievedMapObservable = Observable.just(retrievedMap).publish().autoConnect();
    retrievedMapObservable.map(o -> o.toString());
    when(caller.retrieveJsonFromRequest(anyString(), any(HttpClientRequest.class)))
            .thenReturn(retrievedMapObservable);

    TestSubscriber<Node> testSubscriber = new TestSubscriber<>();
    aggregator.getHealthNodesFromService("testService", "testUrl", null).toBlocking().subscribe(testSubscriber);
    testSubscriber.assertNoValues();
    testSubscriber.assertCompleted();

    verify(caller, times(1)).retrieveJsonFromRequest(eq("testService"), requestCaptor.capture());
    assertThat(requestCaptor.getValue().getHeaders().entries()).usingElementComparator(stringEntryComparator())
            .containsExactlyElementsOf(requestHeaders().entrySet());
}

@Test
public void shouldReturnEmptyObservableOnExceptionsInSourceObservable() {
    when(properties.getRequestHeaders()).thenReturn(requestHeaders());
    Map retrievedMap = new HashMap();
    Observable retrievedMapObservable = Observable.just(retrievedMap);
    retrievedMapObservable.doOnNext(o -> {
        throw new IllegalArgumentException();
    });
    when(caller.retrieveJsonFromRequest(anyString(), any(HttpClientRequest.class)))
            .thenReturn(retrievedMapObservable);

    TestSubscriber<Node> testSubscriber = new TestSubscriber<>();
    aggregator.getHealthNodesFromService("testService", "testUrl", null).toBlocking().subscribe(testSubscriber);
    testSubscriber.getOnNextEvents();
    testSubscriber.assertNoValues();
    testSubscriber.assertCompleted();

    verify(caller, times(1)).retrieveJsonFromRequest(eq("testService"), requestCaptor.capture());
    assertThat(requestCaptor.getValue().getHeaders().entries()).usingElementComparator(stringEntryComparator())
            .containsExactlyElementsOf(requestHeaders().entrySet());
}

@Test
public void shouldReturnEmptyObservableOnErroneousConversion() {
    when(properties.getRequestHeaders()).thenReturn(requestHeaders());
    Map retrievedMap = new HashMap();
    Observable retrievedMapObservable = Observable.just(retrievedMap).publish().autoConnect();
    when(caller.retrieveJsonFromRequest(anyString(), any(HttpClientRequest.class)))
            .thenReturn(retrievedMapObservable);
    when(converter.convertToNodes(anyString(), anyMap()))
            .thenThrow(new RuntimeException("Error1"));

    TestSubscriber<Node> testSubscriber = new TestSubscriber<>();
    aggregator.getHealthNodesFromService("testService", "testUrl", null).toBlocking().subscribe(testSubscriber);
    testSubscriber.getOnNextEvents();
    testSubscriber.assertNoValues();
    testSubscriber.assertCompleted();

    verify(caller, times(1)).retrieveJsonFromRequest(eq("testService"), requestCaptor.capture());
    assertThat(requestCaptor.getValue().getHeaders().entries()).usingElementComparator(stringEntryComparator())
            .containsExactlyElementsOf(requestHeaders().entrySet());
}

@Test
public void testUserAgentContains_MesosRxJavaCore_RxNetty() throws Exception {
    final String clientName = "unit-tests";
    final MesosClient<String, String> client = MesosClientBuilder.<String, String>newBuilder()
        .sendCodec(StringMessageCodec.UTF8_STRING)
        .receiveCodec(StringMessageCodec.UTF8_STRING)
        .mesosUri(URI.create("http://localhost:12345"))
        .applicationUserAgentEntry(literal(clientName, "latest"))
        .subscribe("subscribe")
        .processStream(events -> events.map(e -> Optional.<SinkOperation<String>>empty()))
        .build();

    final HttpClientRequest<ByteBuf> request = client.createPost
        .call("ACK")
        .toBlocking()
        .first();

    final Map<String, String> headers = headersToMap(request.getHeaders());
    assertThat(headers).containsKeys("User-Agent");
    final String ua = headers.get("User-Agent");
    assertThat(ua).startsWith(String.format("%s/%s", clientName, "latest"));
    assertThat(ua).contains("mesos-rxjava-client/");
}

@Test
public void testRequestUriFromPassedUri() throws Exception {
    final Func1<String, Observable<HttpClientRequest<ByteBuf>>> createPost = MesosClient.curryCreatePost(
        URI.create("http://localhost:12345/glavin/api/v1/scheduler"),
        StringMessageCodec.UTF8_STRING,
        StringMessageCodec.UTF8_STRING,
        new UserAgent(
            literal("testing", "latest")
        ),
        new AtomicReference<>(null)
    );

    final HttpClientRequest<ByteBuf> request = createPost.call("something")
        .toBlocking()
        .first();

    assertThat(request.getUri()).isEqualTo("/glavin/api/v1/scheduler");
}

@Test
public void testMesosStreamIdAddedToRequestWhenNonNull() throws Exception {
    final Func1<String, Observable<HttpClientRequest<ByteBuf>>> createPost = MesosClient.curryCreatePost(
        URI.create("http://localhost:12345/api/v1/scheduler"),
        StringMessageCodec.UTF8_STRING,
        StringMessageCodec.UTF8_STRING,
        new UserAgent(
            literal("testing", "latest")
        ),
        new AtomicReference<>("streamId")
    );

    final HttpClientRequest<ByteBuf> request = createPost.call("something")
        .toBlocking()
        .first();

    final Map<String, String> headers = headersToMap(request.getHeaders());
    assertThat(headers).containsKeys("Mesos-Stream-Id");
    assertThat(headers.get("Mesos-Stream-Id")).isEqualTo("streamId");
}

@Test
public void testMesosStreamIdNotPresentWhenNull() throws Exception {
    final Func1<String, Observable<HttpClientRequest<ByteBuf>>> createPost = MesosClient.curryCreatePost(
        URI.create("http://localhost:12345/api/v1/scheduler"),
        StringMessageCodec.UTF8_STRING,
        StringMessageCodec.UTF8_STRING,
        new UserAgent(
            literal("testing", "latest")
        ),
        new AtomicReference<>(null)
    );

    final HttpClientRequest<ByteBuf> request = createPost.call("something")
        .toBlocking()
        .first();

    final Map<String, String> headers = headersToMap(request.getHeaders());
    assertThat(headers).doesNotContainKeys("Mesos-Stream-Id");
}

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

public void execute(IndicesExistsRequest request, final ActionListener<IndicesExistsResponse> listener) {
    logger.debug("indices exists request {}", request);
    try {
        RequestUriBuilder uriBuilder = new RequestUriBuilder(Strings.arrayToCommaDelimitedString(request.indices()));

        uriBuilder.addQueryParameter("local", request.local());
        uriBuilder.addIndicesOptions(request);

        indicesAdminClient.getHttpClient().submit(HttpClientRequest.<ByteBuf>create(HttpMethod.HEAD, uriBuilder.toString()))
                .flatMap(HANDLES_404)
                .flatMap(new Func1<HttpClientResponse<ByteBuf>, Observable<IndicesExistsResponse>>() {
                    @Override
                    public Observable<IndicesExistsResponse> call(final HttpClientResponse<ByteBuf> response) {
                        return IndicesExistsResponse.parse(response.getStatus().code());
                    }
                })
                .single()
                .subscribe(new ListenerCompleterObserver<>(listener));

    } catch (Exception e) {
        listener.onFailure(e);
    }
}

public void execute(ClearScrollRequest request, final ActionListener<ClearScrollResponse> listener) {
    logger.debug("clear scroll request {}", request);
    try {
        RequestUriBuilder uriBuilder = new RequestUriBuilder()
                .addEndpoint("_search/scroll");

        uriBuilder.addQueryParameter("scroll_id", Strings.collectionToCommaDelimitedString(request.getScrollIds()));
        httpClient.getHttpClient().submit(HttpClientRequest.createDelete(uriBuilder.toString()))
                .flatMap(HANDLES_404)
                .flatMap(new Func1<HttpClientResponse<ByteBuf>, Observable<ClearScrollResponse>>() {
                    @Override
                    public Observable<ClearScrollResponse> call(final HttpClientResponse<ByteBuf> response) {
                        return response.getContent().flatMap(new Func1<ByteBuf, Observable<ClearScrollResponse>>() {
                            @Override
                            public Observable<ClearScrollResponse> call(ByteBuf byteBuf) {
                                return ClearScrollResponse.parse(response.getStatus().code());
                            }
                        });
                    }
                })
                .single()
                .subscribe(new ListenerCompleterObserver<>(listener));
    } catch (Exception e) {
        listener.onFailure(e);
    }
}

private static HttpClientRequest<ByteBuf> compress(
        HttpClientRequest<ByteBuf> req, byte[] entity) {
    if (entity.length >= MIN_COMPRESS_SIZE) {
        req.withHeader(HttpHeaders.Names.CONTENT_ENCODING, HttpHeaders.Values.GZIP);
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        try (GZIPOutputStream gzip = new GZIPOutputStream(baos)) {
            gzip.write(entity);
        } catch (IOException e) {
            // This isn't expected to occur
            throw new RuntimeException("failed to gzip request payload", e);
        }
        req.withContent(baos.toByteArray());
    } else {
        req.withContent(entity);
    }
    return req;
}

private Observable<HttpClientResponse<ByteBuf>> getResponse(String externalHealthCheckURL) {
    String host = "localhost";
    int port = DEFAULT_APPLICATION_PORT;
    String path = "/healthcheck";
    try {
        URL url = new URL(externalHealthCheckURL);
        host = url.getHost();
        port = url.getPort();
        path = url.getPath();
    } catch (MalformedURLException e) {
        //continue
    }
    Integer timeout = DynamicProperty.getInstance("prana.host.healthcheck.timeout").getInteger(DEFAULT_CONNECTION_TIMEOUT);
    HttpClient<ByteBuf, ByteBuf> httpClient = RxNetty.<ByteBuf, ByteBuf>newHttpClientBuilder(host, port)
            .pipelineConfigurator(PipelineConfigurators.<ByteBuf, ByteBuf>httpClientConfigurator())
            .channelOption(ChannelOption.CONNECT_TIMEOUT_MILLIS, timeout)
            .build();
    return httpClient.submit(HttpClientRequest.createGet(path));

}

public static String getResponse(HttpClientRequest<ByteBuf> request, HttpClient<ByteBuf, ByteBuf> client) {
    return client.submit(request).flatMap(new Func1<HttpClientResponse<ByteBuf>, Observable<String>>() {
        @Override
        public Observable<String> call(HttpClientResponse<ByteBuf> response) {
            return response.getContent().map(new Func1<ByteBuf, String>() {
                @Override
                public String call(ByteBuf byteBuf) {
                    return byteBuf.toString(Charset.defaultCharset());
                }
            });
        }
    }).onErrorFlatMap(new Func1<OnErrorThrowable, Observable<String>>() {
        @Override
        public Observable<String> call(OnErrorThrowable onErrorThrowable) {
            throw onErrorThrowable;
        }
    }).toBlocking().first();
}

public HttpResourceObservableCommand(HttpClient<ByteBuf, ByteBuf> httpClient,
                                     HttpClientRequest<ByteBuf> httpRequest, String hystrixCacheKey,
                                     Map<String, Object> requestProperties,
                                     FallbackHandler<T> fallbackHandler,
                                     ResponseValidator<HttpClientResponse<ByteBuf>> validator,
                                     Class<? extends T> classType,
                                     HystrixObservableCommand.Setter setter) {
    super(setter);
    this.httpClient = httpClient;
    this.fallbackHandler = fallbackHandler;
    this.validator = validator;
    this.httpRequest = httpRequest;
    this.hystrixCacheKey = hystrixCacheKey;
    this.classType = classType;
    this.requestProperties = requestProperties;
}

protected LoadBalancingHttpClient(Builder<I, O> builder) {
    super(builder.lb, builder.config, new RequestSpecificRetryHandler(true, true, builder.retryHandler, null), builder.pipelineConfigurator, builder.poolCleanerScheduler);
    requestIdHeaderName = getProperty(IClientConfigKey.Keys.RequestIdHeaderName, null, null);
    requestIdProvider = (requestIdHeaderName != null) 
                      ? new HttpRequestIdProvider(requestIdHeaderName, RxContexts.DEFAULT_CORRELATOR)
                      : null;
    this.listeners = new CopyOnWriteArrayList<ExecutionListener<HttpClientRequest<I>, HttpClientResponse<O>>>(builder.listeners);
    defaultCommandBuilder = LoadBalancerCommand.<HttpClientResponse<O>>builder()
            .withLoadBalancerContext(lbContext)
            .withListeners(this.listeners)
            .withClientConfig(builder.config)
            .withRetryHandler(builder.retryHandler)
            .build();
    this.responseToErrorPolicy = builder.responseToErrorPolicy;
    this.backoffStrategy = builder.backoffStrategy;
}

@Test
public void testObservable() throws Exception {
    HttpClientRequest<ByteBuf> request = HttpClientRequest.createGet(SERVICE_URI + "testAsync/person");
    LoadBalancingHttpClient<ByteBuf, ByteBuf> observableClient = RibbonTransport.newHttpClient();
    Observable<HttpClientResponse<ByteBuf>> response = observableClient.submit(request);
    Person person = getPersonObservable(response).toBlocking().single();
    assertEquals(EmbeddedResources.defaultPerson, person);
    final HttpClientListener listener = observableClient.getListener();
    assertEquals(1, listener.getPoolAcquires());
    assertEquals(1, listener.getConnectionCount());
    waitUntilTrueOrTimeout(1000, new Func0<Boolean>() {
        @Override
        public Boolean call() {
            return listener.getPoolReleases() == 1;
        }
    });
}

@Test
public void testSubmitToAbsoluteURI() throws Exception {
    HttpClientRequest<ByteBuf> request = HttpClientRequest.createGet(SERVICE_URI + "testAsync/person");
    LoadBalancingHttpClient<ByteBuf, ByteBuf> observableClient = RibbonTransport.newHttpClient();
    // final List<Person> result = Lists.newArrayList();
    Observable<HttpClientResponse<ByteBuf>> response = observableClient.submit(request);
    Person person = getPersonObservable(response).toBlocking().single();
    assertEquals(EmbeddedResources.defaultPerson, person);
    // need to sleep to wait until connection is released
    final HttpClientListener listener = observableClient.getListener();
    assertEquals(1, listener.getConnectionCount());
    assertEquals(1, listener.getPoolAcquires());
    waitUntilTrueOrTimeout(1000, new Func0<Boolean>() {
        @Override
        public Boolean call() {
            return listener.getPoolReleases() == 1;
        }
    });
}

@Test
public void testPoolReuse() throws Exception {
    HttpClientRequest<ByteBuf> request = HttpClientRequest.createGet(SERVICE_URI + "testAsync/person");
    LoadBalancingHttpClient<ByteBuf, ByteBuf> observableClient = RibbonTransport.newHttpClient(
            IClientConfig.Builder.newBuilder().withDefaultValues()
            .withMaxAutoRetries(1)
            .withMaxAutoRetriesNextServer(1).build());
    Observable<HttpClientResponse<ByteBuf>> response = observableClient.submit(request);
    Person person = getPersonObservable(response).toBlocking().single();
    assertEquals(EmbeddedResources.defaultPerson, person);
    response = observableClient.submit(request);
    person = getPersonObservable(response).toBlocking().single();
    assertEquals(EmbeddedResources.defaultPerson, person);
    final HttpClientListener listener = observableClient.getListener();
    assertEquals(2, listener.getPoolAcquires());
    waitUntilTrueOrTimeout(1000, new Func0<Boolean>() {
        @Override
        public Boolean call() {
            return listener.getPoolReleases() == 2;
        }
    });
    assertEquals(1, listener.getConnectionCount());
    assertEquals(1, listener.getPoolReuse());
}

@Test
public void testPostWithByteBuf() throws Exception {
    Person myPerson = new Person("netty", 5);
    ObjectMapper mapper = new ObjectMapper();
    byte[] raw = mapper.writeValueAsBytes(myPerson);
    ByteBuf buffer = Unpooled.copiedBuffer(raw);
    HttpClientRequest<ByteBuf> request = HttpClientRequest.createPost(SERVICE_URI + "testAsync/person")
            .withHeader("Content-type", "application/json")
            .withHeader("Content-length", String.valueOf(raw.length))
            .withContent(buffer);
    LoadBalancingHttpClient<ByteBuf, ByteBuf> observableClient = RibbonTransport.newHttpClient(
            DefaultClientConfigImpl.getClientConfigWithDefaultValues().set(CommonClientConfigKey.ReadTimeout, 10000));
    Observable<HttpClientResponse<ByteBuf>> response = observableClient.submit(request);
    Person person = getPersonObservable(response).toBlocking().single();
    assertEquals(myPerson, person);
}

@Test
public void testStreamWithLoadBalancer() throws Exception {
    // NettyHttpLoadBalancerErrorHandler errorHandler = new NettyHttpLoadBalancerErrorHandler(1, 3, true);
    // IClientConfig config = DefaultClientConfigImpl.getClientConfigWithDefaultValues().withProperty(CommonClientConfigKey.ConnectTimeout, "1000");
    IClientConfig config = IClientConfig.Builder.newBuilder().withRetryOnAllOperations(true)
            .withMaxAutoRetries(1)
            .withMaxAutoRetriesNextServer(3)
            .build();
    BaseLoadBalancer lb = new BaseLoadBalancer(new DummyPing(), new AvailabilityFilteringRule());
    LoadBalancingHttpClient<ByteBuf, ServerSentEvent> lbObservables = (LoadBalancingHttpClient<ByteBuf, ServerSentEvent>) RibbonTransport.newSSEClient(lb, config);
    HttpClientRequest<ByteBuf> request = HttpClientRequest.createGet("/testAsync/personStream");
    List<Person> result = Lists.newArrayList();
    Server goodServer = new Server("localhost:" + port);
    Server badServer = new Server("localhost:12245");
    List<Server> servers = Lists.newArrayList(badServer, badServer, badServer, goodServer);
    lb.setServersList(servers);
    result = getPersonListFromResponse(lbObservables.submit(request, null, null));
    assertEquals(EmbeddedResources.entityStream, result);
}

@edu.umd.cs.findbugs.annotations.SuppressWarnings
public static void main(String[] args) throws Exception {
    LoadBalancingHttpClient<ByteBuf, ByteBuf> client = RibbonTransport.newHttpClient();
    HttpClientRequest<ByteBuf> request = HttpClientRequest.createGet("http://www.google.com/");
    final CountDownLatch latch = new CountDownLatch(1);
    client.submit(request)
        .toBlocking()
        .forEach(new Action1<HttpClientResponse<ByteBuf>>() {
            @Override
            public void call(HttpClientResponse<ByteBuf> t1) {
                System.out.println("Status code: " + t1.getStatus());
                t1.getContent().subscribe(new Action1<ByteBuf>() {

                    @Override
                    public void call(ByteBuf content) {
                        System.out.println("Response content: " + content.toString(Charset.defaultCharset()));
                        latch.countDown();
                    }

                });
            }
        });
    latch.await(2, TimeUnit.SECONDS);
}

private Observable<Void> registerMovie(Movie movie) {
    HttpClientRequest<ByteBuf> httpRequest = HttpClientRequest.createPost("/movies")
            .withHeader("X-Platform-Version", "xyz")
            .withHeader("X-Auth-Token", "abc")
            .withRawContentSource(Observable.just(movie), new RxMovieTransformer());

    return client.submit(httpRequest).flatMap(new Func1<HttpClientResponse<ByteBuf>, Observable<Void>>() {
        @Override
        public Observable<Void> call(HttpClientResponse<ByteBuf> httpClientResponse) {
            if (httpClientResponse.getStatus().code() / 100 != 2) {
                return Observable.error(new RuntimeException(
                        format("HTTP request failed (status code=%s)", httpClientResponse.getStatus())));
            }
            return Observable.empty();
        }
    });
}

private Observable<Void> updateRecommendation(String user, Movie movie) {
    HttpClientRequest<ByteBuf> httpRequest = HttpClientRequest.createPost(format("/users/%s/recommendations", user))
            .withHeader("X-Platform-Version", "xyz")
            .withHeader("X-Auth-Token", "abc")
            .withRawContentSource(Observable.just(movie.getId()), new StringTransformer());

    return client.submit(httpRequest).flatMap(new Func1<HttpClientResponse<ByteBuf>, Observable<Void>>() {
        @Override
        public Observable<Void> call(HttpClientResponse<ByteBuf> httpClientResponse) {
            if (httpClientResponse.getStatus().code() / 100 != 2) {
                return Observable.error(new RuntimeException(
                        format("HTTP request failed (status code=%s)", httpClientResponse.getStatus())));
            }
            return Observable.empty();
        }
    });
}