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

项目：reactive-playing 文件：RxGitterClient.java

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

项目：ge-export 文件：HttpClientFactory.java

public static HttpClient<ByteBuf, ByteBuf> create(String server, final String portStr) {
    int port = 0;

    try {

        URL url = new URL(defaultToHttps(server));
        if (portStr == null) {
            port = url.getDefaultPort();
        } else if (Integer.parseInt(portStr) > 0){
            port = Integer.parseInt(portStr);
        }
        final HttpClient<ByteBuf, ByteBuf> httpClient = HttpClient.newClient(new InetSocketAddress(
                url.getHost(), port));
        if(url.getProtocol().equals("https")) {
            return httpClient.unsafeSecure();
        } else if (url.getProtocol().equals("http")) {
            return httpClient;
        } else {
            throw new RuntimeException("Unsuported protocol");
        }
    }

    catch(MalformedURLException e){
        throw new RuntimeException(e);
    }
}

项目：wildfly-swarm 文件：SecuredTransportFactory.java

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

项目：ARCHIVE-wildfly-swarm 文件：SecuredTransportFactory.java

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

项目：Prana 文件：HealthCheckHandler.java

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

}

项目：Prana 文件：TestUtils.java

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

项目：ribbon 文件：RxMovieProxyExampleTest.java

@Test
public void testTransportFactoryWithInjection() {
    Injector injector = Guice.createInjector(
            new AbstractModule() {
                @Override
                protected void configure() {
                    bind(ClientConfigFactory.class).to(MyClientConfigFactory.class).in(Scopes.SINGLETON);
                    bind(RibbonTransportFactory.class).to(DefaultRibbonTransportFactory.class).in(Scopes.SINGLETON);
                }
            }
    );

    RibbonTransportFactory transportFactory = injector.getInstance(RibbonTransportFactory.class);
    HttpClient<ByteBuf, ByteBuf> client = transportFactory.newHttpClient("myClient");
    IClientConfig config = ((LoadBalancingHttpClient) client).getClientConfig();
    assertEquals("MyConfig", config.getNameSpace());
}

项目：ribbon 文件：HttpResourceObservableCommand.java

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

项目：azure-documentdb-rxjava 文件：RxGatewayStoreModel.java

public RxGatewayStoreModel(ConnectionPolicy connectionPolicy,
        ConsistencyLevel consistencyLevel,
        QueryCompatibilityMode queryCompatibilityMode,
        String masterKey,
        Map<String, String> resourceTokens,
        UserAgentContainer userAgentContainer,
        EndpointManager globalEndpointManager,
        HttpClient<ByteBuf, ByteBuf> httpClient) {
    this.defaultHeaders = new HashMap<String, String>();
    this.defaultHeaders.put(HttpConstants.HttpHeaders.CACHE_CONTROL,
            "no-cache");
    this.defaultHeaders.put(HttpConstants.HttpHeaders.VERSION,
            HttpConstants.Versions.CURRENT_VERSION);

    if (userAgentContainer == null) {
        userAgentContainer = new UserAgentContainer();
    }

    this.defaultHeaders.put(HttpConstants.HttpHeaders.USER_AGENT, userAgentContainer.getUserAgent());

    if (consistencyLevel != null) {
        this.defaultHeaders.put(HttpConstants.HttpHeaders.CONSISTENCY_LEVEL,
                consistencyLevel.toString());
    }

    this.globalEndpointManager = globalEndpointManager;
    this.queryCompatibilityMode = queryCompatibilityMode;

    this.httpClient = httpClient;
}

项目：tusRx 文件：TestClient.java

public static void main(String[] args) {

        HttpClient.newClient("localhost", 8080)
                  .enableWireLogging("hello-client", LogLevel.ERROR)
                  .createOptions("/files")
                  .doOnNext(resp -> logger.info(resp.toString()))
                  .flatMap(resp -> resp.getContent()
                                       .map(bb -> bb.toString(Charset.defaultCharset())))
                  .toBlocking()
                  .forEach(logger::info);
    }

项目：tusRx 文件：TestServerRule.java

public HttpClient<ByteBuf, ByteBuf> getHttpClient () {
    if (client == null) {
        HttpClient<ByteBuf, ByteBuf> _client = embedded ? HttpClient.newClient(getServerAddress()) :  HttpClient.newClient(new InetSocketAddress("localhost", 8080));
        client = _client;
    }
    return client;
}

项目：MarketData 文件：RxNettyEventEventStreamClient.java

private Observable<String> initializeStream() {
    HttpClient<ByteBuf, ServerSentEvent> client =
            RxNetty.createHttpClient("localhost", port, PipelineConfigurators.<ByteBuf>clientSseConfigurator());

    return client.submit(HttpClientRequest.createGet("/hello")).
            flatMap(response -> {
                printResponseHeader(response);
                return response.getContent();
            }).map(serverSentEvent -> serverSentEvent.contentAsString());
}

项目：rxnetty-nio-test 文件：RxJavaNioTest.java

@Test
public void nioTest() throws Exception {
  HttpServer<ByteBuf, ByteBuf> server = getServer();

  TestSubscriber<String> ts = new TestSubscriber<>();

  long start = System.currentTimeMillis();

  // we use 10 since the default rxnetty thread pool size is 8
  // you could also shrink the pool down for the same effect
  // but I couldn't be bothered finding the settings
  Observable.range(1, 10)
    // flatMap runs async Observables concurrently
    .flatMap(i ->
      HttpClient.newClient(server.getServerAddress())
        .createGet("/" + i)
        .flatMap(response ->
          response.getContent()
            .map(bytes ->
              bytes.toString(Charset.defaultCharset()) + " " +
                "[response received on " + Thread.currentThread().getName() +
                " at " + (System.currentTimeMillis() - start) + "]"
            )
        )
    )
    .doOnNext(System.out::println)
    .subscribe(ts);

  ts.awaitTerminalEvent();

  server.shutdown();
}

项目：hawkular-apm 文件：NettyNoResponseHttpITest.java

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

项目：hawkular-apm 文件：NettyNoResponseHttpITest.java

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

项目：hawkular-apm 文件：NettyNoResponseHttpITest.java

@Test
public void testPUT() 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.createPut(PATH_3);
    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_3, testProducer.getUri());
    assertTrue(testProducer.getProperties(Constants.PROP_HTTP_QUERY).isEmpty());
    assertEquals("PUT", testProducer.getOperation());
    assertEquals("PUT", testProducer.getProperties("http_method").iterator().next().getValue());
}

项目：hawkular-apm 文件：NettyHttpITest.java

@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())))
            .single().toBlocking().toFuture().get(5, TimeUnit.SECONDS);
    assertEquals(HELLO_WORLD, result1);

    // Check stored traces (including 1 for the test client)
    Wait.until(() -> getApmMockServer().getTraces().size() == 1);
    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());
}

项目：hawkular-apm 文件：NettyHttpITest.java

@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())))
            .single().toBlocking().toFuture().get(5, TimeUnit.SECONDS);
    assertEquals(HELLO_WORLD, result1);

    // Check stored traces (including 1 for the test client)
    Wait.until(() -> getApmMockServer().getTraces().size() == 1);
    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());
}

项目：hawkular-apm 文件：NettyHttpITest.java

@Test
public void testPUT() 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.createPut(PATH_3);
    req1.writeStringContent(Observable.just(HELLO_THERE));

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

    // Check stored traces (including 1 for the test client)
    Wait.until(() -> getApmMockServer().getTraces().size() == 1);
    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_3, testProducer.getUri());
    assertTrue(testProducer.getProperties(Constants.PROP_HTTP_QUERY).isEmpty());
    assertEquals("PUT", testProducer.getOperation());
    assertEquals("PUT", testProducer.getProperties("http_method").iterator().next().getValue());
}

项目：mesos-rxjava 文件：SinkSubscriber.java

SinkSubscriber(
    @NotNull final HttpClient<ByteBuf, ByteBuf> httpClient,
    @NotNull final Func1<Send, Observable<HttpClientRequest<ByteBuf>>> createPost
) {
    this.httpClient = httpClient;
    this.createPost = createPost;
}

项目：mesos-rxjava 文件：TcpSocketProxyTest.java

@Test
public void testConnectionTerminatedOnClose() throws Exception {
    final TcpSocketProxy proxy = new TcpSocketProxy(
        new InetSocketAddress("localhost", 0),
        new InetSocketAddress("localhost", server.getServerPort())
    );
    proxy.start();

    final int listenPort = proxy.getListenPort();
    final HttpClient<ByteBuf, ByteBuf> client = RxNetty.createHttpClient("localhost", listenPort);

    final String first = client.submit(HttpClientRequest.createGet("/"))
        .flatMap(AbstractHttpContentHolder::getContent)
        .map(bb -> bb.toString(StandardCharsets.UTF_8))
        .toBlocking()
        .first();

    assertThat(first).isEqualTo("Hello World");
    LOGGER.info("first request done");
    proxy.shutdown();
    if (proxy.isShutdown()) {
        proxy.close();
    } else {
        fail("proxy should have been shutdown");
    }

    try {
        final URI uri = URI.create(String.format("http://localhost:%d/", listenPort));
        uri.toURL().getContent();
        fail("Shouldn't have been able to get content");
    } catch (IOException e) {
        // expected
    }
}

项目：reactivesocket-websocket-rxnetty 文件：ClientServerTest.java

@BeforeClass
public static void setup() {
    ReactiveSocketWebSocketServer serverHandler = ReactiveSocketWebSocketServer.create(
            requestResponsePayload -> {
                String requestResponse = byteToString(requestResponsePayload.getData());
                if (requestResponse.startsWith("h")) {
                    return Single.just(utf8EncodedPayloadData(requestResponse + " world"));
                } else if ("test".equals(requestResponse)) {
                    return Single.just(utf8EncodedPayloadData("test response"));
                } else {
                    return Single.error(new RuntimeException("Not Found"));
                }
            } ,
            requestStreamPayload -> {
                String requestStream = byteToString(requestStreamPayload.getData());
                return Observable.just(requestStream, "world").map(n -> utf8EncodedPayloadData(n));
            } , null, null, null);

    server = HttpServer.newServer()
            // .clientChannelOption(ChannelOption.AUTO_READ, true)
            // .enableWireLogging(LogLevel.ERROR)
            .start((req, resp) -> {
                return resp.acceptWebSocketUpgrade(serverHandler::acceptWebsocket);
            });

    client = HttpClient.newClient("localhost", server.getServerPort()).enableWireLogging(LogLevel.ERROR)
            .createGet("/rs")
            .requestWebSocketUpgrade()
            .flatMap(WebSocketResponse::getWebSocketConnection)
            .map(ReactiveSocketWebSocketClient::create)
            .toBlocking().single();

    client.connect()
            .subscribe(v -> {
            } , t -> t.printStackTrace());
}

项目：reactivesocket-websocket-rxnetty 文件：ReactiveSocketWebSocketServerTest.java

@Test
public void test() {
    // create protocol with handlers
    ReactiveSocketWebSocketServer handler = ReactiveSocketWebSocketServer.create(
            requestResponsePayload -> {
                String requestResponse = byteToString(requestResponsePayload.getData()); 
                return Single.just(utf8EncodedPayloadData("hello" + requestResponse));
            } ,
            requestStreamPayload -> {
                String requestStream = byteToString(requestStreamPayload.getData());
                return just("a_" + requestStream, "b_" + requestStream).map(n -> utf8EncodedPayloadData(n));
            } , null, null, null);

    // start server with protocol
    HttpServer<ByteBuf, ByteBuf> server = HttpServer.newServer();
    int port = server.getServerPort();
    server.start((request, response) -> {
        return response.acceptWebSocketUpgrade(handler::acceptWebsocket);
    });

    // TODO send actual requests
    HttpClient.newClient("localhost", server.getServerPort())
            .createGet("/")
            .requestWebSocketUpgrade();

    server.shutdown();
}

项目：triathlon 文件：MarathonClient.java

public Observable<HttpClientResponse<ByteBuf>> postMessage(String message) {
    PipelineConfigurator<HttpClientResponse<ByteBuf>, HttpClientRequest<ByteBuf>> pipelineConfigurator
            = PipelineConfigurators.httpClientConfigurator();

    HttpClient<ByteBuf, ByteBuf> client = RxNetty.<ByteBuf, ByteBuf>newHttpClientBuilder(networkAddress.getIpAddress(), port)
            .pipelineConfigurator(pipelineConfigurator)
            .enableWireLogging(LogLevel.ERROR).build();

    HttpClientRequest<ByteBuf> request = HttpClientRequest.createPost("/v2/apps");
    request.withRawContentSource(Observable.just(message), StringTransformer.DEFAULT_INSTANCE);
    request.withHeader("Content-Type", "application/json");
    return client.submit(request);
}

项目：triathlon 文件：HealthCheck.java

private HttpClientResponse<ByteBuf> getResponse(String serviceUrl) throws MalformedURLException, InterruptedException, ExecutionException, TimeoutException {
    String host, path;
    int port;

    URL url = new URL(serviceUrl);
    host = url.getHost();
    port = url.getPort();
    path = url.getPath();
    System.out.println(url);

    HttpClient<ByteBuf, ByteBuf> httpClient = RxNetty.<ByteBuf, ByteBuf>newHttpClientBuilder(host, port)
            .pipelineConfigurator(PipelineConfigurators.<ByteBuf, ByteBuf>httpClientConfigurator())
            .build();
    return httpClient.submit(HttpClientRequest.createGet(path)).toBlocking().toFuture().get(checkTimeout, TimeUnit.MILLISECONDS);
}

项目：argos-dashboard 文件：DefaultHystrixClusterMonitor.java

@Override
public Observable<String> observeJson() {
    if(jsonObservable != null) {
        return jsonObservable;
    }

    HttpClientRequest<ByteBuf> request = HttpClientRequest.createGet(url.getPath() + "?" + url.getQuery());
    int port = url.getPort() < 0 ? url.getDefaultPort() : url.getPort();
    HttpClient<ByteBuf, ServerSentEvent> client = RxNetty.<ByteBuf, ServerSentEvent>newHttpClientBuilder(url.getHost(), port)
            .withNoConnectionPooling()
            .pipelineConfigurator(PipelineConfigurators.<ByteBuf>clientSseConfigurator())
            .build();


    jsonObservable = client.submit(request)
            .doOnError(t -> LOG.error("Error connecting to " + url, t))
            .flatMap(response -> {
                        if (response.getStatus().code() != 200) {
                            return Observable.error(new RuntimeException("Failed to connect: " + response.getStatus()));
                        }

                        return response.getContent()
                                .doOnSubscribe(() -> LOG.info("Turbine => Aggregate Stream from URL: " + url))
                                .doOnUnsubscribe(() -> LOG.info("Turbine => Unsubscribing Stream: " + url))
                                .map(ServerSentEvent::contentAsString);
                    }
            )
            .timeout(120, TimeUnit.SECONDS)
            .retryWhen(attempts -> attempts.zipWith(Observable.range(1, Integer.MAX_VALUE), (k, i) -> i)
                    .flatMap(n -> {
                        int waitTimeSeconds = Math.min(6, n) * 10; // wait in 10 second increments up to a max of 1 minute
                        LOG.info("Turbine => Retrying connection to: " + this.url + " in {} seconds", waitTimeSeconds);
                        return Observable.timer(waitTimeSeconds, TimeUnit.SECONDS);
                    })
            )
            .repeat()
            .share();

    return jsonObservable;
}

项目：kha 文件：ConditionsCommand.java

public ConditionsCommand(WeatherRequest request,
                         WundergroundDeviceConfiguration configuration,
                         HttpClient<ByteBuf, ByteBuf> httpClient, ObjectMapper mapper) {
    super(Setter
                    .withGroupKey(HystrixCommandGroupKey.Factory.asKey("Wunderground"))
                    .andCommandKey(HystrixCommandKey.Factory.asKey("ConditionsCommand"))
    );
    this.request = request;
    this.apiKey = configuration.getApiKey();
    this.httpClient = httpClient;
    this.mapper = mapper;
}

项目：kha 文件：WundergroundTemperatureSensor.java

public WundergroundTemperatureSensor(WundergroundDeviceConfiguration configuration,
                                     HttpClient<ByteBuf, ByteBuf> httpClient,
                                     ObjectMapper mapper) {
    WeatherRequest request = new WeatherRequest()
            .forCityInCountry(configuration.getCity(), configuration.getCountry());
    this.command = new ConditionsCommand(request, configuration, httpClient, mapper);
}

项目：kha 文件：WundergroundPressureSensor.java

public WundergroundPressureSensor(WundergroundDeviceConfiguration configuration,
                                  HttpClient<ByteBuf, ByteBuf> httpClient,
                                  ObjectMapper mapper) {
    WeatherRequest request = new WeatherRequest()
                    .forCityInCountry(configuration.getCity(), configuration.getCountry());
    this.command = new ConditionsCommand(request, configuration, httpClient, mapper);
}

项目：kha 文件：WundergroundSensorCreator.java

@Override
public Sensor<? extends Quantity> createFromBuilder(SensorBuilder<WundergroundDevice> builder) {
    WundergroundDevice device = builder.getDevice();
    WundergroundDeviceConfiguration configuration = device.getConfiguration();
    HttpClient<ByteBuf, ByteBuf> httpClient = device.getHttpClient();
    ObjectMapper mapper = device.getMapper();
    switch (builder.getType()) {
        case WundergroundTemperatureSensor.TYPE:
            return new WundergroundTemperatureSensor(configuration, httpClient, mapper);
        case WundergroundPressureSensor.TYPE:
            return new WundergroundPressureSensor(configuration, httpClient, mapper);
        default:
            throw new IllegalArgumentException(format("Can't process type %s", builder.getType()));
    }
}

项目：ReactiveLab 文件：LoadBalancerFactory.java

public LoadBalancer<HttpClientHolder<ByteBuf, ServerSentEvent>> forVip(String targetVip) {
    Observable<MembershipEvent<Host>> eurekaHostSource = membershipSource.forInterest(Interests.forVips(targetVip), instanceInfo -> {
        String ipAddress = instanceInfo.getDataCenterInfo()
                .getAddresses().stream()
                .filter(na -> na.getProtocolType() == ProtocolType.IPv4)
                .collect(Collectors.toList()).get(0).getIpAddress();
        HashSet<ServicePort> servicePorts = instanceInfo.getPorts();
        ServicePort portToUse = servicePorts.iterator().next();
        return new Host(ipAddress, portToUse.getPort());
    });

    final Map<Host, HttpClientHolder<ByteBuf, ServerSentEvent>> hostVsHolders = new ConcurrentHashMap<>();

    String lbName = targetVip + "-lb";
    return LoadBalancers.newBuilder(eurekaHostSource.map(
            hostEvent -> {
                HttpClient<ByteBuf, ServerSentEvent> client = clientPool.getClientForHost(hostEvent.getClient());
                HttpClientHolder<ByteBuf, ServerSentEvent> holder;
                if (hostEvent.getType() == MembershipEvent.EventType.REMOVE) {
                    holder = hostVsHolders.remove(hostEvent.getClient());
                } else {
                    holder = new HttpClientHolder<>(client);
                    hostVsHolders.put(hostEvent.getClient(), holder);
                }
                return new MembershipEvent<>(hostEvent.getType(), holder);
            })).withWeightingStrategy(new LinearWeightingStrategy<>(new RxNettyPendingRequests<>()))
               .withName(lbName)
            .withFailureDetector(new RxNettyFailureDetector<>()).build();
}

项目：ribbon 文件：SSEClient.java

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

项目：ribbon 文件：LoadBalancingHttpClient.java

/**
 * Convert an HttpClientRequest to a ServerOperation 
 * 
 * @param server
 * @param request
 * @param rxClientConfig
 * @return
 */
protected ServerOperation<HttpClientResponse<O>> requestToOperation(final HttpClientRequest<I> request, final ClientConfig rxClientConfig) {
    Preconditions.checkNotNull(request);

    return new ServerOperation<HttpClientResponse<O>>() {
        final AtomicInteger count = new AtomicInteger(0);

        @Override
        public Observable<HttpClientResponse<O>> call(Server server) {
            HttpClient<I,O> rxClient = getOrCreateRxClient(server);
            setHostHeader(request, server.getHost());

            Observable<HttpClientResponse<O>> o;
            if (rxClientConfig != null) {
                o = rxClient.submit(request, rxClientConfig);
            } 
            else {
                o = rxClient.submit(request);
            }

            return o.concatMap(new Func1<HttpClientResponse<O>, Observable<HttpClientResponse<O>>>() {
                @Override
                public Observable<HttpClientResponse<O>> call(HttpClientResponse<O> t1) {
                    if (t1.getStatus().code()/100 == 4 || t1.getStatus().code()/100 == 5)
                        return responseToErrorPolicy.call(t1, backoffStrategy.call(count.getAndIncrement()));
                    else
                        return Observable.just(t1);
                }
            });
        }
    };
}

项目：WSPerfLab 文件：WSClient.java

public WSClient(String host, int port, int firstStep, int stepSize, int stepDuration, String query) {
    this.host = host;
    this.port = port;
    this.firstStep = firstStep;
    this.stepSize = stepSize;
    this.stepDuration = stepDuration;
    this.query = query;

    System.out.println("Starting client with hostname: " + host + "  port: " + port + "  first-step: " + firstStep + "  step-size: " + stepSize + "  step-duration: " + stepDuration + "s  query: " + query);

    httpClient = new HttpClientBuilder<ByteBuf, ByteBuf>(this.host, this.port)
            .withMaxConnections(15000)
            .config(new HttpClient.HttpClientConfig.Builder().readTimeout(1, TimeUnit.MINUTES).build())
            .build();
    stats = new ConnectionPoolMetricListener();
    httpClient.subscribe(stats);

    client = httpClient.submit(HttpClientRequest.createGet(this.query))
            .flatMap(response -> {
                if (response.getStatus().code() == 200) {
                    counter.increment(CounterEvent.SUCCESS);
                } else {
                    counter.increment(CounterEvent.HTTP_ERROR);
                }
                return response.getContent().doOnNext(bb -> {
                    counter.add(CounterEvent.BYTES, bb.readableBytes());
                });
            }).doOnError((t) -> {
                if (t instanceof PoolExhaustedException) {
                    counter.increment(CounterEvent.POOL_EXHAUSTED);
                } else {
                    counter.increment(CounterEvent.NETTY_ERROR);
                }
            });
}

项目：WSPerfLab 文件：TestRouteBasic.java

public TestRouteBasic(String backendHost, int backendPort) {
    host = backendHost;
    port = backendPort;
    client = new HttpClientBuilder<ByteBuf, ByteBuf>(host, port)
            .withMaxConnections(10000)
            .config(new HttpClient.HttpClientConfig.Builder().readTimeout(1, TimeUnit.MINUTES).build())
            .build();
    stats = new ConnectionPoolMetricListener();
    client.subscribe(stats);
}

项目：RxNettyProxy 文件：ExternalRequestExecutor.java

public ExternalRequestExecutor(final HttpClient<ByteBuf, ByteBuf> client) {
    this.client = client;
}

项目：elasticsearch-http 文件：HttpClusterAdminClient.java

public HttpClusterAdminClient(Supplier<HttpClient<ByteBuf, ByteBuf>> httpClient) {
    this.httpClient = httpClient;
}

项目：elasticsearch-http 文件：HttpClusterAdminClient.java

public io.reactivex.netty.protocol.http.client.HttpClient<ByteBuf, ByteBuf> getHttpClient() {
    return httpClient.get();
}

项目：elasticsearch-http 文件：HttpIndicesAdminClient.java

public HttpIndicesAdminClient(Supplier<io.reactivex.netty.protocol.http.client.HttpClient<ByteBuf, ByteBuf>> httpClient) {
    this.httpClient = httpClient;
}

项目：elasticsearch-http 文件：HttpIndicesAdminClient.java

public io.reactivex.netty.protocol.http.client.HttpClient<ByteBuf, ByteBuf> getHttpClient() {
    return httpClient.get();
}