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

private void emmit(FlowableEmitter<Message> emitter, String roomId) throws Exception {
    SSLContext sslCtx = SSLContext.getDefault();
    SSLEngine sslEngine = sslCtx.createSSLEngine("stream.gitter.im", 443);
    sslEngine.setUseClientMode(true);

    HttpClient
            .newClient("stream.gitter.im", 443)
            .secure(sslEngine)
            .createGet("/v1/rooms/" + roomId + "/chatMessages")
            .addHeader("Authorization", "Bearer 3cd4820adf59b6a7116f99d92f68a1b786895ce7")
            .flatMap(HttpClientResponse::getContent)
            .filter(bb -> bb.capacity() > 2)
            .map(MessageEncoder::mapToMessage)
            .doOnNext(m -> System.out.println("Log Emit: " + m))
            .subscribe(emitter::onNext, emitter::onError, emitter::onComplete);
}

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

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

/**
 * Attempts to read the content of an error response as {@code text/plain;charset=utf-8}, otherwise the content
 * will be ignored and a string detailing the Content-Type that was not processed.
 * <p>
 * <b>NOTE:</b>
 * <i>
 *     This method MUST be called from the netty-io thread otherwise the content of the response will not be
 *     available because if will be released automatically as soon as the netty-io thread is left.
 * </i>
 * @param resp  The response to attempt to read from
 * @return An {@link Observable} representing the {@code text/plain;charset=utf-8} response content if it existed
 *         or an error message indicating the content-type that was not attempted to read.
 */
@NotNull
static Observable<String> attemptToReadErrorResponse(@NotNull final HttpClientResponse<ByteBuf> resp) {
    final HttpResponseHeaders headers = resp.getHeaders();
    final String contentType = resp.getHeaders().get(HttpHeaderNames.CONTENT_TYPE);
    if (headers.isContentLengthSet() && headers.getContentLength() > 0 ) {
        if (contentType != null && contentType.startsWith("text/plain")) {
            return resp.getContent()
                .map(r -> r.toString(StandardCharsets.UTF_8));
        } else {
            resp.ignoreContent();
            final String errMsg = getErrMsg(contentType);
            return Observable.just(errMsg);
        }
    } else {
        return Observable.just("");
    }
}

@Test
public void attemptToReadErrorResponse_responseContentIgnoredByDefaultWhenNotString() throws Exception {
    final String errMsg = "lies";
    final byte[] bytes = errMsg.getBytes(StandardCharsets.UTF_8);
    final HttpClientResponse<ByteBuf> resp = response(Unpooled.copiedBuffer(bytes), (headers) -> {
        headers.add("Content-Type", "application/json;charset=utf-8");
        headers.add("Content-Length", bytes.length);
    });

    final String err = ResponseUtils.attemptToReadErrorResponse(resp).toBlocking().first();
    assertThat(err).isNotEqualTo("lies");

    try {
        resp.getContent().toBlocking().first();
    } catch (IllegalStateException e) {
        assertThat(e.getMessage()).isEqualTo("Content stream is already disposed.");
    }
}

@Test
public void testMesosStreamIdIsSavedForSuccessfulSubscribeCall() throws Exception {
    final AtomicReference<String> mesosStreamId = new AtomicReference<>(null);

    final Func1<HttpClientResponse<ByteBuf>, Observable<ByteBuf>> f = MesosClient.verifyResponseOk(
        "Subscribe",
        mesosStreamId,
        StringMessageCodec.UTF8_STRING.mediaType()
    );

    final DefaultHttpResponse nettyResponse = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
    nettyResponse.headers().add("Mesos-Stream-Id", "streamId");
    nettyResponse.headers().add("Content-Type", StringMessageCodec.UTF8_STRING.mediaType());
    final HttpClientResponse<ByteBuf> response = new HttpClientResponse<>(
        nettyResponse,
        UnicastContentSubject.create(1000, TimeUnit.MILLISECONDS)
    );

    f.call(response);


    assertThat(mesosStreamId.get()).isEqualTo("streamId");
}

@Test
public void testMesosStreamIdIsNotSavedForUnsuccessfulSubscribeCall() throws Exception {
    final AtomicReference<String> mesosStreamId = new AtomicReference<>(null);

    final Func1<HttpClientResponse<ByteBuf>, Observable<ByteBuf>> f = MesosClient.verifyResponseOk(
        "Subscribe",
        mesosStreamId,
        StringMessageCodec.UTF8_STRING.mediaType()
    );

    final DefaultHttpResponse nettyResponse = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST);
    nettyResponse.headers().add("Mesos-Stream-Id", "streamId");
    nettyResponse.headers().add("Content-Type", StringMessageCodec.UTF8_STRING.mediaType());
    final HttpClientResponse<ByteBuf> response = new HttpClientResponse<>(
        nettyResponse,
        UnicastContentSubject.create(1000, TimeUnit.MILLISECONDS)
    );

    try {
        f.call(response);
    } catch (Mesos4xxException e) {
        // expected
    }

    assertThat(mesosStreamId.get()).isEqualTo(null);
}

@Test
public void testVerifyResponseOk_ensuresContentTypeOfResponseMatchesReceiveCodec() throws Exception {
    final Func1<HttpClientResponse<ByteBuf>, Observable<ByteBuf>> f = MesosClient.verifyResponseOk(
        "Subscribe",
        new AtomicReference<>(),
        StringMessageCodec.UTF8_STRING.mediaType()
    );

    final DefaultHttpResponse nettyResponse = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
    nettyResponse.headers().add("Content-Type", "text/html");
    final HttpClientResponse<ByteBuf> response = new HttpClientResponse<>(
        nettyResponse,
        UnicastContentSubject.create(1000, TimeUnit.MILLISECONDS)
    );

    try {
        f.call(response);
    } catch (MesosException e) {
        final String expected = String.format(
            "Response had Content-Type \"%s\" expected \"%s\"",
            "text/html",
            StringMessageCodec.UTF8_STRING.mediaType()
        );
        assertThat(e.getContext().getMessage()).isEqualTo(expected);
    }
}

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

/**
 * Utility function to map an Observable<ByteBuf> to an Observable<Integer> while also
 * updating our counters for metrics sent and errors.
 */
protected Func1<HttpClientResponse<ByteBuf>, Integer> withBookkeeping(final int batchSize) {
    return new Func1<HttpClientResponse<ByteBuf>, Integer>() {
        @Override
        public Integer call(HttpClientResponse<ByteBuf> response) {
            boolean ok = response.getStatus().code() == HTTP_OK;
            if (ok) {
                numMetricsSent.increment(batchSize);
            } else {
                LOGGER.info("Status code: {} - Lost {} metrics",
                        response.getStatus().code(), batchSize);
                numMetricsDroppedHttpErr.increment(batchSize);
            }

            return batchSize;
        }
    };
}

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

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

private Movie[] handleGetMoviesReply(Observable<HttpClientResponse<ByteBuf>> httpGet) {
    return httpGet
            .flatMap(new Func1<HttpClientResponse<ByteBuf>, Observable<Movie[]>>() {
                @Override
                public Observable<Movie[]> call(HttpClientResponse<ByteBuf> httpClientResponse) {
                    return httpClientResponse.getContent().map(new Func1<ByteBuf, Movie[]>() {
                        @Override
                        public Movie[] call(ByteBuf byteBuf) {
                            String[] lines = byteBuf.toString(Charset.defaultCharset()).split("\n");
                            Movie[] movies = new Movie[lines.length];
                            for (int i = 0; i < movies.length; i++) {
                                movies[i] = Movie.from(lines[i]);
                            }
                            return movies;
                        }
                    });
                }
            }).toBlocking().first();
}

/**
 * Given the request it creates an observable which upon subscription issues HTTP call and emits one DocumentServiceResponse.
 * 
 * @param request 
 * @param method
 * @return Observable<DocumentServiceResponse> 
 */
public Observable<DocumentServiceResponse> performRequest(RxDocumentServiceRequest request, HttpMethod method) {

    URI uri = getUri(request);

    HttpClientRequest<ByteBuf> httpRequest = HttpClientRequest.create(method, uri.toString());

    this.fillHttpRequestBaseWithHeaders(request.getHeaders(), httpRequest);
    try {

        if (request.getContentObservable() != null) {

            // TODO validate this
            // convert byte[] to ByteBuf
            // why not use Observable<byte[]> directly?
            Observable<ByteBuf> byteBufObservable = request.getContentObservable()
                    .map(bytes ->  Unpooled.wrappedBuffer(bytes));

            httpRequest.withContentSource(byteBufObservable);
        } else if (request.getContent() != null){
            httpRequest.withContent(request.getContent());
        }

    } catch (Exception e) {
        return Observable.error(e);
    }

    Observable<HttpClientResponse<ByteBuf>> clientResponseObservable = this.httpClient.submit(httpRequest);

    return toDocumentServiceResponse(clientResponseObservable, request);
}

/**
 * Transforms the rxNetty's client response Observable to DocumentServiceResponse Observable.
 * 
 * 
 * Once the the customer code subscribes to the observable returned by the {@link AsyncDocumentClient} CRUD APIs,
 * the subscription goes up till it reaches the source rxNetty's observable, and at that point the HTTP invocation will be made.
 * 
 * @param clientResponseObservable
 * @param request
 * @return {@link Observable}
 */
private Observable<DocumentServiceResponse> toDocumentServiceResponse(Observable<HttpClientResponse<ByteBuf>> clientResponseObservable, 
        RxDocumentServiceRequest request) {

    return clientResponseObservable.flatMap(clientResponse -> {

        // header key/value pairs
        HttpResponseHeaders httpResponseHeaders = clientResponse.getHeaders();
        HttpResponseStatus httpResponseStatus = clientResponse.getStatus();

        Observable<InputStream> inputStreamObservable;

        if (request.getOperationType() == OperationType.Delete) {
            // for delete we don't expect any body
            inputStreamObservable = Observable.just(null);
        } else {
            // transforms the observable<ByteBuf> to Observable<InputStream>
            inputStreamObservable = toInputStream(clientResponse.getContent());
        }

        Observable<StoreResponse> storeResponseObservable = inputStreamObservable
                .map(contentInputStream -> {
                    try {
                        // If there is any error in the header response this throws exception
                        validateOrThrow(request, httpResponseStatus, httpResponseHeaders, contentInputStream);

                        // transforms to Observable<StoreResponse>
                        return toStoreResponse(httpResponseStatus, httpResponseHeaders, contentInputStream);
                    } catch (Exception e) {
                        throw Exceptions.propagate(e);
                    }
                });

        return storeResponseObservable;

    }).map(storeResponse -> new DocumentServiceResponse(storeResponse));
}

@Test
public void testTusResumable() {
    HttpClientRequest<ByteBuf, ByteBuf> request = serverRule.getHttpClient().createOptions("/files");
    request.flatMap(HttpClientResponse::discardContent).ignoreElements();
    request.subscribe(resp -> logger.info(resp.toString()), x -> logger.error(x.toString()), ()-> logger.info("done"));
    request
        .doOnNext(resp-> logger.info(resp.toString()))
        .map(resp -> resp.getHeader("Tus-Resumable"))
        .toBlocking()
        .forEach(hv -> assertEquals("1.0.0", hv));
}

@Test
public void testStatus() {
    Iterable<HttpClientResponse<ByteBuf>> response = serverRule.getHttpClient()
            .createPost("/files")
            .addHeader("Tus-Resumable", "1.0.0")
            .addHeader("Upload-Length", "100")
            .toBlocking()
            .toIterable();
    Iterator<HttpClientResponse<ByteBuf>> it = response.iterator();
    assertTrue(it.hasNext());
    assertEquals(HttpResponseStatus.CREATED, it.next().getStatus());
}

@Test
public void testLocation() {
    Iterable<HttpClientResponse<ByteBuf>> response = serverRule.getHttpClient().createPost("/files")
            .addHeader("Tus-Resumable", "1.0.0")
            .addHeader("Upload-Length", "100")
            .toBlocking()
            .toIterable();
    Iterator<HttpClientResponse<ByteBuf>> it = response.iterator();
    assertTrue(it.hasNext());
    assertNotNull(it.next().getHeader("Location"));
}

public void testCannotUploadMoreThanUploadLength() {
    Observable<byte[]> contentTooBig = Observable.just(new byte[101]);
    HttpClientResponse<ByteBuf> patchResponse = post(100L).map(this::getLocation).flatMap(location -> { 
        return patch(location, 0L, contentTooBig);
    }).toBlocking().first();

    assertThat(patchResponse.getStatus()).isEqualTo(HttpResponseStatus.BAD_REQUEST);

}

private static Observable<ServerSentEvent> buildStream(Instant since, String lastStreamedBuildId) {

        AtomicReference<String> _lastBuildId = new AtomicReference<>(null);

        final String buildsSinceUri = "/build-export/v1/builds/since/" + String.valueOf(since.toEpochMilli());
        LOGGER.info("Builds uri: " + buildsSinceUri);

        HttpClientRequest<ByteBuf, ByteBuf> request = HTTP_CLIENT
                .createGet(buildsSinceUri)
                .setKeepAlive(true);
        if (BASIC_AUTH != null) {
            request = request.addHeader("Authorization", "Basic " + BASIC_AUTH);
        }

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

        return request
                .flatMap(HttpClientResponse::getContentAsServerSentEvents)
                .doOnNext(serverSentEvent -> _lastBuildId.set(serverSentEvent.getEventIdAsString()))
                .doOnSubscribe(() -> LOGGER.info("Streaming builds..."))
                .onErrorResumeNext(t -> {
                    LOGGER.info("Error streaming builds, resuming from build id: " + _lastBuildId.get());
                    return buildStream(since, _lastBuildId.get());
                });
    }

private static void printResponseHeader(HttpClientResponse<ServerSentEvent> response) {
    System.out.println("New response received.");
    System.out.println("========================");
    System.out.println(response.getHttpVersion().text() + ' ' + response.getStatus().code()
            + ' ' + response.getStatus().reasonPhrase());
    for (Map.Entry<String, String> header : response.getHeaders().entries()) {
        System.out.println(header.getKey() + ": " + header.getValue());
    }
}

@Override
public void onExecutionSuccess(ExecutionContext<HttpClientRequest<ByteBuf>> context, HttpClientResponse<ByteBuf> response, ExecutionInfo info) {
    KeycloakSecurityContext securityContext = (KeycloakSecurityContext) context.get(KeycloakSecurityContextAssociation.class.getName());
    if (securityContext != null) {
        KeycloakSecurityContextAssociation.associate(securityContext);
    } else {
        KeycloakSecurityContextAssociation.disassociate();
    }
}

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

    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);

    HttpClientResponse<ByteBuf> pactTwoResponse = mock(HttpClientResponse.class);
    ByteBuf byteBuf3 = (new PooledByteBufAllocator()).directBuffer();
    ByteBufUtil.writeUtf8(byteBuf3, pactTwo);
    when(pactTwoResponse.getContent()).thenReturn(Observable.just(byteBuf3));
    when(pactTwoResponse.getStatus()).thenReturn(HttpResponseStatus.OK);

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


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

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

    assertThat(nodes.get(0).getId()).isEqualTo("consumer2");
    assertThat(nodes.get(0).getLane()).isEqualTo(0);
    assertThat(nodes.get(0).getLinkedToNodeIds()).contains("pn:provider2");
    assertThat(nodes.get(0).getDetails().get("url")).isEqualTo("http://someserver.be:7000/pacts/provider/provider2/consumer/consumer2/version/1.0.0");
    assertThat(nodes.get(0).getDetails().get("type")).isEqualTo(NodeTypes.UI_COMPONENT);
    assertThat(nodes.get(0).getDetails().get("status")).isEqualTo("UP");
}

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

    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);

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

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


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

@Test
public void testGET() throws InterruptedException, ExecutionException, TimeoutException {
    SocketAddress serverAddress = new InetSocketAddress("127.0.0.1", server.getServerPort());

    /*Create a new client for the server address*/
    HttpClient<ByteBuf, ByteBuf> client = HttpClient.newClient(serverAddress);
    HttpClientRequest<ByteBuf, ByteBuf> req1 = client.createGet(PATH_1 + "?" + QUERY_1);

    Object result1 = req1
            .flatMap((HttpClientResponse<ByteBuf> resp) -> resp.getContent()
                    .map(bb -> bb.toString(Charset.defaultCharset())))
            .singleOrDefault(null).toBlocking().toFuture().get(5, TimeUnit.SECONDS);

    assertNull(result1);

    Wait.until(() -> getApmMockServer().getTraces().size() == 1);

    // Check stored traces (including 1 for the test client)
    assertEquals(1, getApmMockServer().getTraces().size());

    List<Producer> producers = NodeUtil.findNodes(getApmMockServer().getTraces().get(0).getNodes(), Producer.class);

    assertEquals("Expecting 1 producers", 1, producers.size());

    Producer testProducer = producers.get(0);

    assertEquals(PATH_1, testProducer.getUri());
    assertEquals(QUERY_1, testProducer.getProperties(Constants.PROP_HTTP_QUERY).iterator().next().getValue());
    assertEquals("GET", testProducer.getOperation());
    assertEquals("GET", testProducer.getProperties("http_method").iterator().next().getValue());
}

@Test
public void testPOST() throws InterruptedException, ExecutionException, TimeoutException {
    SocketAddress serverAddress = new InetSocketAddress("127.0.0.1", server.getServerPort());

    /*Create a new client for the server address*/
    HttpClient<ByteBuf, ByteBuf> client = HttpClient.newClient(serverAddress);
    HttpClientRequest<ByteBuf, ByteBuf> req1 = client.createPost(PATH_2);
    req1.writeStringContent(Observable.just(HELLO_THERE));

    Object result1 = req1
            .flatMap((HttpClientResponse<ByteBuf> resp) -> resp.getContent()
                    .map(bb -> bb.toString(Charset.defaultCharset())))
            .singleOrDefault(null).toBlocking().toFuture().get(5, TimeUnit.SECONDS);

    assertNull(result1);

    Wait.until(() -> getApmMockServer().getTraces().size() == 1);

    // Check stored traces (including 1 for the test client)
    assertEquals(1, getApmMockServer().getTraces().size());

    List<Producer> producers = NodeUtil.findNodes(getApmMockServer().getTraces().get(0).getNodes(), Producer.class);

    assertEquals("Expecting 1 producers", 1, producers.size());

    Producer testProducer = producers.get(0);

    assertEquals(PATH_2, testProducer.getUri());
    assertTrue(testProducer.getProperties(Constants.PROP_HTTP_QUERY).isEmpty());
    assertEquals("POST", testProducer.getOperation());
    assertEquals("POST", testProducer.getProperties("http_method").iterator().next().getValue());
}