import io.vertx.reactivex.core.buffer.Buffer; //导入依赖的package包/类
@Override
public void handle(RoutingContext rc) {
  Long albumId = PathUtil.parseLongParam(rc.pathParam("albumId"));
  if (albumId == null) {
    rc.next();
    return;
  }

  // Currently there's an issue with rx.Buffer as value type, so workaround with core Buffer
  LocalMap<Long, io.vertx.core.buffer.Buffer> covers = rc.vertx().sharedData().getLocalMap("covers");
  io.vertx.core.buffer.Buffer cached = covers.get(albumId);
  if (cached != null) {
    rc.response().end(Buffer.newInstance(cached));
    return;
  }

  download(albumId)
    .doOnSuccess(buffer -> covers.put(albumId, buffer.getDelegate()))
    .subscribe(rc.response()::end, rc::fail, () -> rc.fail(404));
}
 

import io.vertx.reactivex.core.buffer.Buffer; //导入依赖的package包/类
private Maybe<Buffer> download(Long albumId) {
  return dbClient.rxGetConnection().flatMap(sqlConnection -> {
    return findAlbum(sqlConnection, albumId).doAfterTerminate(sqlConnection::close);
  }).flatMapMaybe(album -> {
    String mbAlbumId = album.getString("mbAlbumId");
    return mbAlbumId == null ? Maybe.empty() : Maybe.just(mbAlbumId);
  }).flatMap(mbAlbumId -> {
    return webClient
      .getAbs("http://coverartarchive.org")
      .uri("/release/" + mbAlbumId + "/front")
      .as(BodyCodec.buffer())
      .rxSend()
      .map(HttpResponse::body)
      .toMaybe();
  });
}
 

import io.vertx.reactivex.core.buffer.Buffer; //导入依赖的package包/类
@Override
public void start() throws Exception {
    MqttClient client = MqttClient.create(vertx);

    client.rxConnect(PORT, HOST)
        .flatMapPublisher(ack ->
            Flowable.interval(1, TimeUnit.SECONDS)
                .flatMapSingle(l -> {
                    JsonObject payload = new JsonObject()
                        .put("uuid", id)
                        .put("data",
                            random.nextInt(100));
                    return client
                        .rxPublish("/data",
                            Buffer.buffer(payload.encode()),
                            MqttQoS.AT_MOST_ONCE, false, false);
                }))
        .subscribe();
}
 

import io.vertx.reactivex.core.buffer.Buffer; //导入依赖的package包/类
private void delegateWithCircuitBreaker(RoutingContext rc) {
    HttpEndpoint.rxGetWebClient(discovery, svc -> svc.getName().equals("currency-3rdparty-service"))
        .flatMap(client -> {

            // TODO
            // Use the circuit breaker (circuit) to call the service. Use the rxExecuteCommandWithFallback` method.
            // This methods takes 2 parameters: the first one if a function taking a `Future` as parameter and
            // needs to report the success or failure on this future. The second method is a function providing
            // the fallback result. You must provide a JSON object as response. For the fallback use:
            // new JsonObject()
            //      .put("amount", rc.getBodyAsJson().getDouble("amount"))
            //      .put("currency", "USD"))
            // In the first function, use the given client, emit a POST request on / containing the incoming
            // payload (rc.getBodyAsJson()). Extract the response payload as JSON (bodyAsJsonObject). Don't
            // forget to subscribe (you can use subscribe(toObserver(fut)). You can have a look to the `delegate`
            // method as example.
            // -----
            return Single.just(new JsonObject().put("amount", 0.0).put("currency", "N/A"));
        })
        // ----
        .map(JsonObject::toBuffer)
        .map(Buffer::new)

        .subscribe(toObserver(rc));
}
 

import io.vertx.reactivex.core.buffer.Buffer; //导入依赖的package包/类
/**
 * Method to check the proxy requesting to convert the current portfolio to EUR.
 *
 * @param rc the routing context
 */
private void convertPortfolioToEuro(RoutingContext rc) {
    EventBusService.getServiceProxy(discovery, svc -> svc.getName().equals("portfolio"), PortfolioService.class,
        ar -> {
            if (ar.failed()) {
                rc.fail(ar.cause());
            } else {
                ar.result().evaluate(res -> {
                    if (res.failed()) {
                        rc.fail(res.cause());
                    } else {
                        JsonObject payload = new JsonObject().put("amount", res.result()).put("currency", "EUR");
                        rc.setBody(new Buffer(payload.toBuffer()));
                        delegateWithCircuitBreaker(rc);
                    }
                });
            }
        });
}
 

import io.vertx.reactivex.core.buffer.Buffer; //导入依赖的package包/类
private Single<HttpResponse<Buffer>> callService(
    Pair<ClientRequest, ServiceMetadata> serviceData, HttpMethod method) {
  final Single<HttpResponse<Buffer>> httpResponse;

  final ClientRequest serviceRequest = serviceData.getLeft();
  final ServiceMetadata serviceMetadata = serviceData.getRight();

  final HttpRequest<Buffer> request = webClient
      .request(method, serviceMetadata.getPort(), serviceMetadata.getDomain(),
          serviceRequest.getPath());

  updateRequestQueryParams(request, serviceMetadata);
  updateRequestHeaders(request, serviceRequest, serviceMetadata);
  overrideRequestHeaders(request, serviceMetadata);

  if (!serviceRequest.getFormAttributes().isEmpty()) {
    httpResponse = request.rxSendForm(serviceRequest.getFormAttributes());
  } else {
    httpResponse = request.rxSend();
  }

  return httpResponse;
}
 

import io.vertx.reactivex.core.buffer.Buffer; //导入依赖的package包/类
private void updateRequestHeaders(HttpRequest<Buffer> request, ClientRequest serviceRequest,
    ServiceMetadata serviceMetadata) {

  MultiMap filteredHeaders = getFilteredHeaders(serviceRequest.getHeaders(),
      serviceMetadata.getAllowedRequestHeaderPatterns());
  filteredHeaders.names().forEach(
      headerName -> filteredHeaders.getAll(headerName)
          .forEach(value -> request.headers().add(headerName, value)));

  if (customRequestHeader.containsKey("name") && customRequestHeader.containsKey("value")) {
    request.headers().set(
        customRequestHeader.getString("name"),
        customRequestHeader.getString("value")
    );
  }
}
 

import io.vertx.reactivex.core.buffer.Buffer; //导入依赖的package包/类
public void delayFlowable(HttpServer server) {
  server.requestHandler(request -> {
    if (request.method() == HttpMethod.POST) {

      // Stop receiving buffers
      request.pause();

      checkAuth(res -> {

        // Now we can receive buffers again
        request.resume();

        if (res.succeeded()) {
          Flowable<Buffer> flowable = request.toFlowable();
          flowable.subscribe(buff -> {
            // Get buffers
          });
        }
      });
    }
  });
}
 

import io.vertx.reactivex.core.buffer.Buffer; //导入依赖的package包/类
@Test
public void testGet() {
  int times = 5;
  waitFor(times);
  HttpServer server = vertx.createHttpServer(new HttpServerOptions().setPort(8080));
  server.requestStream().handler(req -> req.response().setChunked(true).end("some_content"));
  try {
    server.listen(ar -> {
      client = WebClient.wrap(vertx.createHttpClient(new HttpClientOptions()));
      Single<HttpResponse<Buffer>> single = client
              .get(8080, "localhost", "/the_uri")
              .as(BodyCodec.buffer())
              .rxSend();
      for (int i = 0; i < times; i++) {
        single.subscribe(resp -> {
          Buffer body = resp.body();
          assertEquals("some_content", body.toString("UTF-8"));
          complete();
        }, this::fail);
      }
    });
    await();
  } finally {
    server.close();
  }
}
 

import io.vertx.reactivex.core.buffer.Buffer; //导入依赖的package包/类
@Test
public void testPost() {
  int times = 5;
  waitFor(times);
  HttpServer server = vertx.createHttpServer(new HttpServerOptions().setPort(8080));
  server.requestStream().handler(req -> req.bodyHandler(buff -> {
    assertEquals("onetwothree", buff.toString());
    req.response().end();
  }));
  try {
    server.listen(ar -> {
      client = WebClient.wrap(vertx.createHttpClient(new HttpClientOptions()));
      Observable<Buffer> stream = Observable.just(Buffer.buffer("one"), Buffer.buffer("two"), Buffer.buffer("three"));
      Single<HttpResponse<Buffer>> single = client
              .post(8080, "localhost", "/the_uri")
              .rxSendStream(stream);
      for (int i = 0; i < times; i++) {
        single.subscribe(resp -> complete(), this::fail);
      }
    });
    await();
  } finally {
    server.close();
  }
}
 

import io.vertx.reactivex.core.buffer.Buffer; //导入依赖的package包/类
@Test
public void testResponseMissingBody() throws Exception {
  int times = 5;
  waitFor(times);
  HttpServer server = vertx.createHttpServer(new HttpServerOptions().setPort(8080));
  server.requestStream().handler(req -> req.response().setStatusCode(403).end());
  try {
    server.listen(ar -> {
      client = WebClient.wrap(vertx.createHttpClient(new HttpClientOptions()));
      Single<HttpResponse<Buffer>> single = client
              .get(8080, "localhost", "/the_uri")
              .rxSend();
      for (int i = 0; i < times; i++) {
        single.subscribe(resp -> {
          assertEquals(403, resp.statusCode());
          assertNull(resp.body());
          complete();
        }, this::fail);
      }
    });
    await();
  } finally {
    server.close();
  }
}
 

import io.vertx.reactivex.core.buffer.Buffer; //导入依赖的package包/类
@Override
public void onNext(Buffer buffer) {
    Response response = new ResponseParser().apply(buffer.getDelegate().getBytes());
    if (ResponseType.PING.equals(response.getType())) {
        String gameVersion = ((ResponsePing) response).getGameVersion();
        if (!API_VERSION.equals(gameVersion)) {
            log.warn("Ocraft uses sc2api in version {}. The game server uses {}.", API_VERSION, gameVersion);
        }
        cancel();
        onConnectionVerified.run();
    }
}
 

import io.vertx.reactivex.core.buffer.Buffer; //导入依赖的package包/类
/**
 * Utility method to report the completion/failure from a Single to a Routing Context.
 *
 * @param rc the routing context
 * @return the single observer to pass to {@link Single#subscribe()}
 */
public static SingleObserver<Buffer> toObserver(RoutingContext rc) {
    return new SingleObserver<Buffer>() {
        public void onSubscribe(@NonNull Disposable d) {
        }

        public void onSuccess(@NonNull Buffer payload) {
            rc.response().end(payload);
        }

        public void onError(Throwable error) {
            rc.fail(error);
        }
    };
}
 

import io.vertx.reactivex.core.buffer.Buffer; //导入依赖的package包/类
private void delegateWithCircuitBreaker(RoutingContext rc) {
    HttpEndpoint.rxGetWebClient(discovery, svc -> svc.getName().equals("currency-3rdparty-service"))
        .flatMap(client ->

            // TODO
            // Use the circuit breaker (circuit) to call the service. Use the rxExecuteCommandWithFallback` method.
            // This methods takes 2 parameters: the first one if a function taking a `Future` as parameter and
            // needs to report the success or failure on this future. The second method is a function providing
            // the fallback result. You must provide a JSON object as response. For the fallback use:
            // new JsonObject()
            //      .put("amount", rc.getBodyAsJson().getDouble("amount"))
            //      .put("currency", "USD"))
            // In the first function, use the given client, emit a POST request on / containing the incoming
            // payload (rc.getBodyAsJson()). Extract the response payload as JSON (bodyAsJsonObject). Don't
            // forget to subscribe (you can use subscribe(toObserver(fut)). You can have a look to the `delegate`
            // method as example.
            // -----
            circuit.rxExecuteCommandWithFallback(
                fut ->
                    client.post("/").rxSendJsonObject(rc.getBodyAsJson())
                        .map(HttpResponse::bodyAsJsonObject)
                        .subscribe(toObserver(fut)),
                err -> new JsonObject()
                    .put("amount", rc.getBodyAsJson().getDouble("amount"))
                    .put("currency", "USD")))

        // ----
        .map(JsonObject::toBuffer)
        .map(Buffer::new)

        .subscribe(toObserver(rc));
}
 

import io.vertx.reactivex.core.buffer.Buffer; //导入依赖的package包/类
private Observable<ClientResponse> processResponse(final HttpClientResponse response) {
  return Observable.just(Buffer.buffer())
      .mergeWith(response.toObservable())
      .reduce(Buffer::appendBuffer)
      .toObservable()
      .map(buffer -> toResponse(buffer, response));
}
 

import io.vertx.reactivex.core.buffer.Buffer; //导入依赖的package包/类
private ClientResponse toResponse(Buffer buffer, final HttpClientResponse httpResponse) {
  if (httpResponse.statusCode() >= 300 && httpResponse.statusCode() < 400) { //redirect responses
    LOGGER.info("Repository 3xx response: {}, Headers[{}]", httpResponse.statusCode(),
        DataObjectsUtil.toString(httpResponse.headers()));
  } else if (httpResponse.statusCode() != 200) {
    LOGGER.error("Repository error response: {}, Headers[{}]", httpResponse.statusCode(),
        DataObjectsUtil.toString(httpResponse.headers()));
  }

  return new ClientResponse()
      .setStatusCode(httpResponse.statusCode())
      .setHeaders(httpResponse.headers())
      .setBody(buffer.getDelegate());

}
 

import io.vertx.reactivex.core.buffer.Buffer; //导入依赖的package包/类
private void logResponse(Pair<ClientRequest, ServiceMetadata> request,
    HttpResponse<Buffer> resp) {
  if (resp.statusCode() >= 400 && resp.statusCode() < 600) {
    LOGGER.error("{} {} -> Got response {}, headers[{}]",
        logResponseData(request, resp));
  } else if (LOGGER.isDebugEnabled()) {
    LOGGER.debug("{} {} -> Got response {}, headers[{}]",
        logResponseData(request, resp));
  }
}
 

import io.vertx.reactivex.core.buffer.Buffer; //导入依赖的package包/类
private Object[] logResponseData(Pair<ClientRequest, ServiceMetadata> request,
    HttpResponse<Buffer> resp) {
  Object[] data = {
      request.getLeft().getMethod(),
      toUrl(request),
      resp.statusCode(),
      DataObjectsUtil.toString(resp.headers())};

  return data;
}
 

import io.vertx.reactivex.core.buffer.Buffer; //导入依赖的package包/类
private void overrideRequestHeaders(HttpRequest<Buffer> request, ServiceMetadata metadata) {
  if (metadata.getAdditionalHeaders().isPresent()) {
    metadata.getAdditionalHeaders().get().forEach(entry -> {
      request.putHeader(entry.getKey(), entry.getValue().toString());
    });
  }
}
 

import io.vertx.reactivex.core.buffer.Buffer; //导入依赖的package包/类
private void updateRequestQueryParams(HttpRequest<Buffer> request, ServiceMetadata metadata) {
  if (metadata.getQueryParams().isPresent()) {
    metadata.getQueryParams().get().forEach(entry ->
        request.addQueryParam(entry.getKey(), entry.getValue().toString())
    );
  }
}