Java 类io.reactivex.netty.client.RxClient 实例源码

public void close(WebSocketSession webSocketSession) {
    Replicator replicator = replicatorMap.remove(webSocketSession.getId());
    if (replicator != null) {
        try {
            replicator.close();
            webSocketSession.close();
        } catch (IOException e) {
            logger.warn(e.getMessage());
        }
    }
    RxClient rxClient = rxClientMap.get(webSocketSession.getId());
    if (rxClient != null) {
        rxClient.shutdown();
    }
    observableMap.remove(webSocketSession.getId());
    authMap.remove(webSocketSession.getId());
}

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

@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("unchecked")
@Test
public void badStatusCode(){
    HttpClientRequest<ByteBuf> request = mock(HttpClientRequest.class);
    when(request.getUri()).thenReturn("http://someUri.com");

    HttpClientResponse<ByteBuf> response = mock(HttpClientResponse.class);
    when(response.getStatus()).thenReturn(HttpResponseStatus.BAD_REQUEST);

    HttpResponseHeaders httpResponseHeaders = mock(HttpResponseHeaders.class);
    when(httpResponseHeaders.entries()).thenReturn(newArrayList());
    when(response.getHeaders()).thenReturn(httpResponseHeaders);

    Observable<HttpClientResponse<ByteBuf>> observable = Observable.just(response);
    when(rxClient.submit(any(RxClient.ServerInfo.class), eq(request))).thenReturn(observable);

       TestSubscriber<Map<String, Object>> testSubscriber = new TestSubscriber<>();
    nettyServiceCaller.retrieveJsonFromRequest("serviceId", request).toBlocking().subscribe(testSubscriber);
       testSubscriber.assertNoErrors();

       verify(errorHandler).handleNodeWarning(Mockito.eq("serviceId"), Mockito.anyString());
}

@SuppressWarnings("unchecked")
@Test
public void httpRequestReturnsErroneousObservable(){
    HttpClientRequest<ByteBuf> request = mock(HttpClientRequest.class);
    when(request.getUri()).thenReturn("http://someUri.com");

    HttpRequestHeaders httpRequestHeaders = mock(HttpRequestHeaders.class);
    when(httpRequestHeaders.entries()).thenReturn(Lists.newArrayList());
    when(request.getHeaders()).thenReturn(httpRequestHeaders);

    when(rxClient.submit(any(RxClient.ServerInfo.class), eq(request)))
            .thenReturn(Observable.error(new RuntimeException()));

       TestSubscriber<Map<String, Object>> testSubscriber = new TestSubscriber<>();
    nettyServiceCaller.retrieveJsonFromRequest("serviceId", request).toBlocking().subscribe(testSubscriber);
       testSubscriber.assertNoErrors();

       verify(errorHandler).handleNodeError(Mockito.eq("serviceId"), Mockito.anyString(), Mockito.any(RuntimeException.class));
}

public HttpClient(Collection<String> nodes) {
    // searchShard
    // search template

    List<io.reactivex.netty.protocol.http.client.HttpClient<ByteBuf, ByteBuf>> clientsTemp = new ArrayList<>();
    // expect something like "http://%s:%d"
    for (String node : nodes) {
        String[] next = node.split(":");
        // indices admin
        String host = next[1].substring(2); // remove the // of http://
        int port = Integer.parseInt(next[2]);
        HttpClientBuilder<ByteBuf, ByteBuf> clientBuilder = RxNetty.newHttpClientBuilder(host, port);
        clientBuilder.config(new RxClient.ClientConfig.Builder().readTimeout(timeOut, MILLISECONDS).build());
        clientBuilder.withMaxConnections(maxConnections);
        clientsTemp.add(clientBuilder.build());
        logger.info("adding host {}:{}", host, port);
    }
    this.clients = new SnapshotableCopyOnWriteArray<>(clientsTemp);

    clientSupplier = new RoundRobinSupplier<>(clients);

    this.httpAdminClient = new HttpAdminClient(clientSupplier);
}

@Override
protected RxClient<I, O> createRxClient(Server server) {
    ClientBuilder<I, O> builder = RxNetty.newTcpClientBuilder(server.getHost(), server.getPort());
    if (pipelineConfigurator != null) {
        builder.pipelineConfigurator(pipelineConfigurator);
    }
    Integer connectTimeout = getProperty(IClientConfigKey.Keys.ConnectTimeout, null, DefaultClientConfigImpl.DEFAULT_CONNECT_TIMEOUT);
    builder.channelOption(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeout);
    if (isPoolEnabled()) {
        builder.withConnectionPoolLimitStrategy(poolStrategy)
        .withIdleConnectionsTimeoutMillis(idleConnectionEvictionMills)
        .withPoolIdleCleanupScheduler(poolCleanerScheduler);
    } else {
        builder.withNoConnectionPooling();
    }
    RxClient<I, O> client = builder.build();
    return client;
}

public Observable<DatagramPacket> submit(final String content) {
    return LoadBalancerCommand.<DatagramPacket>builder()
            .withLoadBalancerContext(lbContext)
            .build()
            .submit(new ServerOperation<DatagramPacket>() {
                @Override
                public Observable<DatagramPacket> call(Server server) {
                    RxClient<DatagramPacket, DatagramPacket> rxClient = getOrCreateRxClient(server);
                    return rxClient.connect().flatMap(new Func1<ObservableConnection<DatagramPacket, DatagramPacket>, Observable<? extends DatagramPacket>>() {
                        @Override
                        public Observable<? extends DatagramPacket> call(ObservableConnection<DatagramPacket, DatagramPacket> connection) {
                            connection.writeStringAndFlush(content);
                            return connection.getInput().timeout(10, TimeUnit.MILLISECONDS).take(1);
                        }
                    });
                }
            });
}

public static RxClient.ServerInfo getServerInfoFromRequestOrClient(HttpClientRequest<ByteBuf> request,
                                                                   CompositeHttpClient<ByteBuf, ByteBuf> rxClient) {
    RxClient.ServerInfo serverInfo = rxClient.getDefaultServer();

    try {
        URI uri = new URI(request.getUri());

        final String host = uri.getHost();
        if (null != host) {
            int port = uri.getPort();
            if (port < 0) {
                String scheme = uri.getScheme();
                if (null != scheme) {
                    if ("http".equals(scheme)) {
                        port = 80;
                    } else if ("https".equals(scheme)) {
                        port = 443;
                    }
                }
            }
            serverInfo = new RxClient.ServerInfo(host, port);
        }
    } catch (URISyntaxException e) {
        logger.error("Could not extract server info from request: {0}", e.getMessage());
    }

    return serverInfo;
}

private Observable<String> getPactUrlsFromBroker(Object outboundSecurityObject) {
    logger.info("Discovering pact urls");
    final String url = pactBrokerUrl + latestPactsUrl;
    HttpClientRequest<ByteBuf> request = HttpClientRequest.createGet(url);
    applyOutboundSecurityStrategyOnRequest(request, outboundSecurityObject);
    for (Map.Entry<String, String> header : properties.getRequestHeaders().entrySet()) {
        request.withHeader(header.getKey(), header.getValue());
    }

    RxClient.ServerInfo serverInfo = NettyServiceCaller.getServerInfoFromRequestOrClient(request, rxClient);

    return rxClient.submit(serverInfo, request)
            .filter(r -> {
                if (r.getStatus().code() < 400) {
                    return true;
                } else {
                    String warning = "Exception " + r.getStatus() + " for call " + url + " with headers " + r.getHeaders().entries();
                    logger.warn(warning);
                    publisher.publishEvent(new SystemEvent(warning));
                    return false;
                }
            })
            .flatMap(response -> response.getContent())
            .map(data -> data.toString(Charset.defaultCharset()))
            .doOnError(el -> {
                String error = MessageFormat.format("Could not convert ByteBuf to String for call {0}: {1}",
                        request.getUri(), el);
                logger.error(error);
                publisher.publishEvent(new SystemEvent(error, el));
            })
            .onErrorReturn(Throwable::toString)
            .map(response -> {
                logger.info("logging response: " + response);
                return response;
            })
            .map(response -> JsonPath.<List<String>>read(response, selfHrefJsonPath))
            .map(jsonList -> Observable.from(jsonList))
            .flatMap(el -> (Observable<String>) el.map(obj -> (String) obj))
            .doOnNext(pactUrl -> logger.info("Pact url discovered: " + pactUrl));
}

private Observable<Node> getNodesFromPacts(final String url, final Object outboundSecurityObject) {
    logger.info("Discovering pact urls");
    HttpClientRequest<ByteBuf> request = HttpClientRequest.createGet(url);

    applyOutboundSecurityStrategyOnRequest(request, outboundSecurityObject);
    for (Map.Entry<String, String> header : properties.getRequestHeaders().entrySet()) {
        request.withHeader(header.getKey(), header.getValue());
    }

    RxClient.ServerInfo serverInfo = NettyServiceCaller.getServerInfoFromRequestOrClient(request, rxClient);

    return rxClient.submit(serverInfo, request)
            .filter(r -> {
                if (r.getStatus().code() < 400) {
                    return true;
                } else {
                    String warning = "Exception " + r.getStatus() + " for call " + url + " with properties " + r.getHeaders().entries();
                    logger.warn(warning);
                    publisher.publishEvent(new SystemEvent(warning));
                    return false;
                }
            })
            .flatMap(response -> response.getContent())
            .map(data -> data.toString(Charset.defaultCharset()))
            .doOnError(el -> {
                String error = MessageFormat.format("Could not convert ByteBuf to String for call {0}: {1}",
                        request.getUri(), el);
                logger.error(error);
                publisher.publishEvent(new SystemEvent(error, el));
            })
            .onErrorReturn(Throwable::toString)
            .map(response -> pactToNodeConverter.convert(response, url))
            .filter(node -> !properties.getFilteredServices().contains(node.getId()))
            .doOnNext(node -> logger.info("Pact node discovered in url: " + url));
}

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

@SuppressWarnings("unchecked")
@Test
public void onErrorWhenGettingPactsUrl() {
    when(rxClient.submit(any(RxClient.ServerInfo.class), any(HttpClientRequest.class)))
            .thenReturn(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("unchecked")
@Test
public void shouldReturnMapOfTwo(){
    String carJson ="{ \"brand\" : \"Mercedes\", \"doors\" : 5 }";

    HttpClientRequest<ByteBuf> request = mock(HttpClientRequest.class);
    when(request.getUri()).thenReturn("http://someUri.com");

    HttpClientResponse<ByteBuf> response = mock(HttpClientResponse.class);
    when(response.getStatus()).thenReturn(OK);
    HttpResponseHeaders httpResponseHeaders = mock(HttpResponseHeaders.class);
    when(httpResponseHeaders.entries()).thenReturn(newArrayList());
    when(response.getHeaders()).thenReturn(httpResponseHeaders);
       ByteBuf byteBuf = (new PooledByteBufAllocator()).directBuffer();
       ByteBufUtil.writeUtf8(byteBuf, carJson);
    when(response.getContent()).thenReturn(Observable.just(byteBuf));

    Observable<HttpClientResponse<ByteBuf>> observable = Observable.just(response);
    when(rxClient.submit(any(RxClient.ServerInfo.class), eq(request))).thenReturn(observable);

       TestSubscriber<Map<String, Object>> testSubscriber = new TestSubscriber<>();
    nettyServiceCaller.retrieveJsonFromRequest("serviceId", request).toBlocking().subscribe(testSubscriber);
       testSubscriber.assertNoErrors();

       List<Map<String, Object>> json = testSubscriber.getOnNextEvents();

       assertThat(json.size()).isEqualTo(1);
       assertThat(json.get(0)).containsKey("brand");
       assertThat(json.get(0).get("brand")).isEqualTo("Mercedes");
       assertThat(json.get(0)).containsKey("doors");
       assertThat(json.get(0).get("doors")).isEqualTo(5);
}

@Override
protected RxClient<I, O> createRxClient(Server server) {
    UdpClientBuilder<I, O> builder = RxNetty.newUdpClientBuilder(server.getHost(), server.getPort());
    if (pipelineConfigurator != null) {
        builder.pipelineConfigurator(pipelineConfigurator);
    }
    return builder.build();
}

@Override
protected HttpClient<I, ServerSentEvent> getOrCreateRxClient(Server server) {
    HttpClientBuilder<I, ServerSentEvent> clientBuilder =
            new HttpClientBuilder<I, ServerSentEvent>(server.getHost(), server.getPort()).pipelineConfigurator(pipelineConfigurator);
    int requestConnectTimeout = getProperty(IClientConfigKey.Keys.ConnectTimeout, null, DefaultClientConfigImpl.DEFAULT_CONNECT_TIMEOUT);
    RxClient.ClientConfig rxClientConfig = new HttpClientConfig.Builder().build();

    HttpClient<I, ServerSentEvent> client = clientBuilder.channelOption(
            ChannelOption.CONNECT_TIMEOUT_MILLIS, requestConnectTimeout).config(rxClientConfig).build();
    return client;
}

/** 
 * Construct an RxClient.ClientConfig from an IClientConfig
 * 
 * @param requestConfig
 * @return
 */
private RxClient.ClientConfig getRxClientConfig(IClientConfig requestConfig) {
    if (requestConfig == null) {
        return DEFAULT_RX_CONFIG;
    }
    int requestReadTimeout = getProperty(IClientConfigKey.Keys.ReadTimeout, requestConfig, 
                                         DefaultClientConfigImpl.DEFAULT_READ_TIMEOUT);
    Boolean followRedirect = getProperty(IClientConfigKey.Keys.FollowRedirects, requestConfig, null);
    HttpClientConfig.Builder builder = new HttpClientConfig.Builder().readTimeout(requestReadTimeout, TimeUnit.MILLISECONDS);
    if (followRedirect != null) {
        builder.setFollowRedirect(followRedirect);
    }
    return builder.build();        
}

@Test
public void testUdpClientWithoutTimeout() throws Exception {
    int port = choosePort();
    UdpServer<DatagramPacket, DatagramPacket> server = new HelloUdpServer(port, 0).createServer();
    server.start();
    BaseLoadBalancer lb = new BaseLoadBalancer();
    lb.setServersList(Lists.newArrayList(new Server("localhost", port)));
    RxClient<DatagramPacket, DatagramPacket> client = RibbonTransport.newUdpClient(lb,
            DefaultClientConfigImpl.getClientConfigWithDefaultValues());
    try {
        String response = client.connect().flatMap(new Func1<ObservableConnection<DatagramPacket, DatagramPacket>,
                Observable<DatagramPacket>>() {
            @Override
            public Observable<DatagramPacket> call(ObservableConnection<DatagramPacket, DatagramPacket> connection) {
                connection.writeStringAndFlush("Is there anybody out there?");
                return connection.getInput();
            }
        }).take(1)
                .map(new Func1<DatagramPacket, String>() {
                    @Override
                    public String call(DatagramPacket datagramPacket) {
                        return datagramPacket.content().toString(Charset.defaultCharset());
                    }
                })
                .toBlocking()
                .first();
        assertEquals(HelloUdpServer.WELCOME_MSG, response);
    } finally {
        server.shutdown();
    }
}

@Override
protected void handleTextMessage(final WebSocketSession session, final TextMessage message) throws Exception {
    RxClient rxClient = rxClientMap.get(session.getId());
    if (rxClient == null) {
        session.close();
        return;
    }
    final String command = message.getPayload();
    if (command.equalsIgnoreCase("sync")) {
        sync(session);
    } else {
        String auth = authMap.get(session.getId());

        observableMap.get(session.getId())
                .subscribe(connection -> {
                    if (auth != null) {
                        connection.writeAndFlush("auth " + auth + "\r\n" + command + "\r\n");
                    } else {
                        connection.writeAndFlush(command + "\r\n");
                    }
                    connection.getInput()
                            .distinct()
                            .subscribe(response -> {
                                String requestMessage;
                                if (response instanceof DefaultLastBulkStringRedisContent) {
                                    DefaultLastBulkStringRedisContent content = (DefaultLastBulkStringRedisContent) response;
                                    byte[] bytes = new byte[content.content().readableBytes()];
                                    content.content().readBytes(bytes);
                                    requestMessage = new String(bytes);
                                } else if (response instanceof IntegerRedisMessage) {
                                    requestMessage = String.valueOf(((IntegerRedisMessage) response).value());
                                } else if (response instanceof ErrorRedisMessage) {
                                    requestMessage = ((ErrorRedisMessage) response).content();
                                } else if (response instanceof FullBulkStringRedisMessage) {
                                    requestMessage = "nil";
                                } else if (response instanceof SimpleStringRedisMessage) {
                                    requestMessage = ((SimpleStringRedisMessage) response).content();
                                } else if (response instanceof BulkStringHeaderRedisMessage || response instanceof DefaultBulkStringRedisContent || response instanceof ArrayHeaderRedisMessage) {
                                    return;
                                } else {
                                    requestMessage = response.toString();
                                }

                                try {
                                    session.sendMessage(new TextMessage(requestMessage));
                                } catch (IOException e) {
                                    logger.warn(e.getMessage());
                                }
                            });
                });
    }
}

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

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

    HttpClientResponse<ByteBuf> pactOneResponse = mock(HttpClientResponse.class);
    ByteBuf byteBuf2 = (new PooledByteBufAllocator()).directBuffer();
    ByteBufUtil.writeUtf8(byteBuf2, pactOne);
    when(pactOneResponse.getContent()).thenReturn(Observable.just(byteBuf2));
    when(pactOneResponse.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(pactOneResponse), Observable.just(pactTwoResponse));


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

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

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

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

public static RxClient<ByteBuf, ByteBuf> newTcpClient(ILoadBalancer loadBalancer, IClientConfig config) {
    return new LoadBalancingTcpClient<ByteBuf, ByteBuf>(loadBalancer, config, getDefaultRetryHandlerWithConfig(config), null, poolCleanerScheduler);
}

public static <I, O> RxClient<I, O> newTcpClient(ILoadBalancer loadBalancer, PipelineConfigurator<O, I> pipelineConfigurator, 
        IClientConfig config, RetryHandler retryHandler) {
    return new LoadBalancingTcpClient<I, O>(loadBalancer, config, retryHandler, pipelineConfigurator, poolCleanerScheduler);
}

public static <I, O> RxClient<I, O> newTcpClient(PipelineConfigurator<O, I> pipelineConfigurator, 
        IClientConfig config) {
    return new LoadBalancingTcpClient<I, O>(config, getDefaultRetryHandlerWithConfig(config), pipelineConfigurator, poolCleanerScheduler);    
}

public static RxClient<ByteBuf, ByteBuf> newTcpClient(IClientConfig config) {
    return new LoadBalancingTcpClient<ByteBuf, ByteBuf>(config, getDefaultRetryHandlerWithConfig(config), null, poolCleanerScheduler);    
}

public static RxClient<DatagramPacket, DatagramPacket> newUdpClient(ILoadBalancer loadBalancer, IClientConfig config) {
    return new LoadBalancingUdpClient<DatagramPacket, DatagramPacket>(loadBalancer, config, getDefaultRetryHandlerWithConfig(config), null);
}

public static RxClient<DatagramPacket, DatagramPacket> newUdpClient(IClientConfig config) {
    return new LoadBalancingUdpClient<DatagramPacket, DatagramPacket>(config, getDefaultRetryHandlerWithConfig(config), null);
}

public static <I, O> RxClient<I, O> newUdpClient(ILoadBalancer loadBalancer, PipelineConfigurator<O, I> pipelineConfigurator, 
        IClientConfig config, RetryHandler retryHandler) {
    return new LoadBalancingUdpClient<I, O>(loadBalancer, config, retryHandler, pipelineConfigurator);
}

public static <I, O> RxClient<I, O> newUdpClient(PipelineConfigurator<O, I> pipelineConfigurator, IClientConfig config) {
    return new LoadBalancingUdpClient<I, O>(config, getDefaultRetryHandlerWithConfig(config), pipelineConfigurator);
}

public RxClient<ByteBuf, ByteBuf> newTcpClient(IClientConfig config) {
    return RibbonTransport.newTcpClient(config);
}

public RxClient<DatagramPacket, DatagramPacket> newUdpClient(IClientConfig config) {
    return RibbonTransport.newUdpClient(config);
}

public final RxClient<ByteBuf, ByteBuf> newTcpClient(String name) {
    IClientConfig config = clientConfigFactory.newConfig();
    config.loadProperties(name);
    return newTcpClient(config);
}

public RxClient<DatagramPacket, DatagramPacket> newUdpClient(String name) {
    IClientConfig config = clientConfigFactory.newConfig();
    config.loadProperties(name);
    return newUdpClient(config);
}

public static <I, O> RxClient<I, O> createTcpClient(String host, int port, PipelineConfigurator<O, I> configurator) {
    return new ClientBuilder<I, O>(host, port).pipelineConfigurator(configurator).build();
}

public static RxClient<ByteBuf, ByteBuf> createTcpClient(String host, int port) {
    return new ClientBuilder<ByteBuf, ByteBuf>(host, port).build();
}