import io.vertx.rxjava.core.http.HttpClient; //导入依赖的package包/类
private void doDispatch(RoutingContext ctx, String path, ServiceReference ref) {
  HttpClientRequest req = ref.getAs(HttpClient.class).request(ctx.request().method(), path,
      res -> res.bodyHandler(body -> check(res.statusCode() >= 500,
          () -> serviceUnavailable(ctx, res.statusMessage()),
          () -> chain(ctx.response(), r -> res.headers().addAll(r.headers()))
              .setStatusCode(res.statusCode())
              .end(body),
          () -> discovery.release(ref))));
  req.headers().addAll(ctx.request().headers());
  // TODO: 24.08.2017 pass user auth data to service?, (in header)
  if (ctx.getBody() != null) {
    req.end(ctx.getBody());
    return;
  }
  req.end();
}
 

import io.vertx.rxjava.core.http.HttpClient; //导入依赖的package包/类
@Test
public void testHello(TestContext test) {
	HttpClient client = Vertx.vertx().createHttpClient();
	HttpClientRequest req = client.request(HttpMethod.GET, 8080, "localhost", "/hello");

	req.toObservable()
		.flatMap(res -> {
			if (res.statusCode() != 200) {
				test.fail("status.code -> " + res.statusCode());
			}
			return res.toObservable();
		})
		.subscribe(data -> {
			logger.info("Server content " + data.toString("UTF-8"));
		});

	req.end();
}
 

import io.vertx.rxjava.core.http.HttpClient; //导入依赖的package包/类
@Test
public void testIndex(TestContext test) {
	HttpClient client = Vertx.vertx().createHttpClient();
	HttpClientRequest req = client.request(HttpMethod.GET, 8080, "localhost", "/");

	req.toObservable()
		.flatMap(res -> {
			test.assertEquals(res.statusCode(), 303);
			test.assertEquals(res.headers().get(HttpHeaders.LOCATION), is("hello"));

			return res.toObservable();
		})
		.subscribe(data -> {
			logger.info("Server content " + data.toString("UTF-8"));
		});

	req.end();
}
 

import io.vertx.rxjava.core.http.HttpClient; //导入依赖的package包/类
@Test
public void testHttpClient() {
  HttpServer server = vertx.createHttpServer(new HttpServerOptions().setPort(8080));
  server.requestStream().handler(req -> {
    req.response().setChunked(true).end("some_content");
  });
  try {
    server.listen(ar -> {
      HttpClient client = vertx.createHttpClient(new HttpClientOptions());
      client.request(HttpMethod.GET, 8080, "localhost", "/the_uri", resp -> {
        Buffer content = Buffer.buffer();
        Observable<Buffer> observable = resp.toObservable();
        observable.forEach(content::appendBuffer, err -> fail(), () -> {
          assertEquals("some_content", content.toString("UTF-8"));
          testComplete();
        });
      }).end();
    });
    await();
  } finally {
    server.close();
  }
}
 

import io.vertx.rxjava.core.http.HttpClient; //导入依赖的package包/类
@Test
public void testWebsocketClient() {
  HttpServer server = vertx.createHttpServer(new HttpServerOptions().setPort(8080));
  server.websocketStream().handler(ws -> {
    ws.write(Buffer.buffer("some_content"));
    ws.close();
  });
  server.listen(ar -> {
    HttpClient client = vertx.createHttpClient(new HttpClientOptions());
    client.websocket(8080, "localhost", "/the_uri", ws -> {
      Buffer content = Buffer.buffer();
      Observable<Buffer> observable = ws.toObservable();
      observable.forEach(content::appendBuffer, err -> fail(), () -> {
        server.close();
        assertEquals("some_content", content.toString("UTF-8"));
        testComplete();
      });
    });
  });
  await();
}
 

import io.vertx.rxjava.core.http.HttpClient; //导入依赖的package包/类
@Test
public void testWebsocketClientFlatMap() {
  HttpServer server = vertx.createHttpServer(new HttpServerOptions().setPort(8080));
  server.websocketStream().handler(ws -> {
    ws.write(Buffer.buffer("some_content"));
    ws.close();
  });
  server.listen(ar -> {
    HttpClient client = vertx.createHttpClient(new HttpClientOptions());
    Buffer content = Buffer.buffer();
    client.
        websocketStream(8080, "localhost", "/the_uri").
        toObservable().
        flatMap(WebSocket::toObservable).
        forEach(content::appendBuffer, err -> fail(), () -> {
          server.close();
          assertEquals("some_content", content.toString("UTF-8"));
          testComplete();
        });
  });
  await();
}
 

import io.vertx.rxjava.core.http.HttpClient; //导入依赖的package包/类
public RecentPurchasesHandler(HttpClient client, SharedData sharedData) {
    this.client = client;

    usersCache = sharedData.getLocalMap("users");
    productsCache = sharedData.getLocalMap("products");
    userPurchasesCache = sharedData.getLocalMap("user-purchases");
    productPurchasesCache = sharedData.getLocalMap("product-purchases");
}
 

import io.vertx.rxjava.core.http.HttpClient; //导入依赖的package包/类
public void websocketClient(Vertx vertx) {
  HttpClient client = vertx.createHttpClient(new HttpClientOptions());
  client.websocketStream(8080, "localhost", "/the_uri").toObservable().subscribe(
      ws -> {
        // Use the websocket
      },
      error -> {
        // Could not connect
      }
  );
}
 

import io.vertx.rxjava.core.http.HttpClient; //导入依赖的package包/类
public void httpClientRequest(Vertx vertx) {
  HttpClient client = vertx.createHttpClient(new HttpClientOptions());
  HttpClientRequest request = client.request(HttpMethod.GET, 8080, "localhost", "/the_uri");
  request.toObservable().subscribe(
      response -> {
        // Process the response
      },
      error -> {
        // Could not connect
      }
  );
  request.end();
}
 

import io.vertx.rxjava.core.http.HttpClient; //导入依赖的package包/类
/**
 * @see #get(HttpClient, int, String, String, MultiMap)
 */
public static Observable<HttpClientResponse> get(HttpClient client, String requestURI) {
  return Observable.create(subscriber -> {
    HttpClientRequest req = client.get(requestURI);
    Observable<HttpClientResponse> resp = req.toObservable();
    resp.subscribe(subscriber);
    req.end();
  });
}
 

import io.vertx.rxjava.core.http.HttpClient; //导入依赖的package包/类
@Test
public void testHttpClientConnectionFailure() {
  HttpClient client = vertx.createHttpClient(new HttpClientOptions());
  HttpClientRequest req = client.request(HttpMethod.GET, 9998, "255.255.255.255", "/the_uri");
  Observable<HttpClientResponse> obs = req.toObservable();
  obs.subscribe(
      buffer -> fail(),
      err -> testComplete(),
      this::fail);
  req.end();
  await();
}
 

import io.vertx.rxjava.core.http.HttpClient; //导入依赖的package包/类
@Test
public void testHttpClientConnectionFailureFlatMap() {
  HttpClient client = vertx.createHttpClient(new HttpClientOptions());
  HttpClientRequest req = client.request(HttpMethod.GET, 9998, "255.255.255.255", "/the_uri");
  Observable<HttpClientResponse> obs = req.toObservable();
  obs.flatMap(HttpClientResponse::toObservable).forEach(
      buffer -> fail(),
      err -> testComplete(),
      this::fail);
  req.end();
  await();
}
 

import io.vertx.rxjava.core.http.HttpClient; //导入依赖的package包/类
@Test
public void testGetHelper() throws Exception {
  CountDownLatch listenLatch = new CountDownLatch(1);
  HttpServer server = vertx.createHttpServer();
  AtomicInteger count = new AtomicInteger();
  server.requestHandler(req -> {
    req.response().end(Buffer.buffer("request=" + count.getAndIncrement()));
  }).listen(8080, onSuccess(s -> {
    listenLatch.countDown();
  }));
  awaitLatch(listenLatch);
  HttpClient client = vertx.createHttpClient();
  Observable<HttpClientResponse> obs = io.vertx.rxjava.core.RxHelper.get(client, 8080, "localhost", "/foo");
  List<Buffer> bodies = Collections.synchronizedList(new ArrayList<>());
  CountDownLatch reqLatch = new CountDownLatch(1);
  obs.subscribe(resp -> {
    resp.toObservable().subscribe(bodies::add, this::fail, reqLatch::countDown);
  }, this::fail);
  awaitLatch(reqLatch);
  obs.subscribe(resp -> {
    resp.toObservable().subscribe(bodies::add, this::fail, () -> {
      assertEquals(Arrays.asList("request=0", "request=1"), bodies.stream().map(Buffer::toString).collect(Collectors.toList()));
      testComplete();
    });
  }, this::fail);
  await();
}
 

import io.vertx.rxjava.core.http.HttpClient; //导入依赖的package包/类
/**
 * Download a file
 * @param downloadUrl the URL to download from
 * @param dest the destination file
 * @return a single emitting exactly one item once the file has been
 * downloaded
 */
private Single<Void> doDownload(String downloadUrl, AsyncFile dest) {
  ObservableFuture<Void> observable = RxHelper.observableFuture();
  Handler<AsyncResult<Void>> handler = observable.toHandler();
  
  HttpClientOptions options = new HttpClientOptions();
  if (downloadUrl.startsWith("https")) {
    options.setSsl(true);
  }
  HttpClient client = vertx.createHttpClient(options);
  HttpClientRequest req = client.getAbs(downloadUrl);
  
  req.exceptionHandler(t -> handler.handle(Future.failedFuture(t)));
  
  req.handler(res -> {
    if (res.statusCode() != 200) {
      handler.handle(Future.failedFuture(new HttpException(res.statusCode(),
          res.statusMessage())));
      return;
    }
    
    // get content-length
    int length;
    int[] read = { 0 };
    int[] lastOutput = { 0 };
    String contentLength = res.getHeader("Content-Length");
    if (contentLength != null) {
      length = Integer.parseInt(contentLength);
    } else {
      length = 0;
    }
    
    res.exceptionHandler(t -> handler.handle(Future.failedFuture(t)));
    
    // download file contents, log progress
    res.handler(buf -> {
      read[0] += buf.length();
      if (lastOutput[0] == 0 || read[0] - lastOutput[0] > 1024 * 2048) {
        logProgress(length, read[0]);
        lastOutput[0] = read[0];
      }
      dest.write(buf);
    });
    
    res.endHandler(v -> {
      logProgress(length, read[0]);
      handler.handle(Future.succeededFuture());
    });
  });
  
  req.end();
  
  return observable.toSingle();
}