Java 类io.vertx.core.http.HttpClientResponse 实例源码

private Handler<RestResponse> syncHandlerEx(CountDownLatch countDownLatch, Holder<ResponseWrapper> holder) {
  return restResponse -> {
    RequestContext requestContext = restResponse.getRequestContext();
    HttpClientResponse response = restResponse.getResponse();
    if (response == null) {
      // 请求失败，触发请求SC的其他实例
      if (!requestContext.isRetry()) {
        retry(requestContext, syncHandlerEx(countDownLatch, holder));
      } else {
        countDownLatch.countDown();
      }

      return;
    }

    response.bodyHandler(bodyBuffer -> {
      ResponseWrapper responseWrapper = new ResponseWrapper();
      responseWrapper.response = response;
      responseWrapper.bodyBuffer = bodyBuffer;
      holder.value = responseWrapper;
      countDownLatch.countDown();
    });
  };
}

@Override
public boolean unregisterMicroserviceInstance(String microserviceId, String microserviceInstanceId) {
  Holder<HttpClientResponse> holder = new Holder<>();
  IpPort ipPort = ipPortManager.getAvailableAddress();

  CountDownLatch countDownLatch = new CountDownLatch(1);
  RestUtils.delete(ipPort,
      String.format(Const.REGISTRY_API.MICROSERVICE_INSTANCE_OPERATION_ONE, microserviceId, microserviceInstanceId),
      new RequestParam(),
      syncHandler(countDownLatch, HttpClientResponse.class, holder));
  try {
    countDownLatch.await();
    if (holder.value != null) {
      if (holder.value.statusCode() == Status.OK.getStatusCode()) {
        return true;
      }
      LOGGER.warn(holder.value.statusMessage());
    }
  } catch (Exception e) {
    LOGGER.error("unregister microservice instance {}/{} failed",
        microserviceId,
        microserviceInstanceId,
        e);
  }
  return false;
}

void handleResponse(Invocation invocation, HttpClientResponse clientResponse,
    AsyncResponse asyncResp) {
  clientResponse.bodyHandler(responseBuf -> {
    // 此时是在网络线程中，不应该就地处理，通过dispatcher转移线程
    invocation.getResponseExecutor().execute(() -> {
      try {
        HttpServletResponseEx responseEx =
            new VertxClientResponseToHttpServletResponse(clientResponse, responseBuf);
        for (HttpClientFilter filter : httpClientFilters) {
          Response response = filter.afterReceiveResponse(invocation, responseEx);
          if (response != null) {
            asyncResp.complete(response);
            return;
          }
        }
      } catch (Throwable e) {
        asyncResp.fail(invocation.getInvocationType(), e);
      }
    });
  });
}

@Test
public void testSetCseContext() {
  boolean status = false;
  try {
    Invocation invocation = mock(Invocation.class);
    HttpClientResponse httpResponse = mock(HttpClientResponse.class);
    OperationMeta operationMeta = mock(OperationMeta.class);
    RestOperationMeta swaggerRestOperation = mock(RestOperationMeta.class);
    HttpClientRequest request = mock(HttpClientRequest.class);

    Endpoint endpoint = mock(Endpoint.class);
    when(invocation.getOperationMeta()).thenReturn(operationMeta);
    URLPathBuilder urlPathBuilder = mock(URLPathBuilder.class);
    when(swaggerRestOperation.getPathBuilder()).thenReturn(urlPathBuilder);
    operationMeta.getExtData(RestConst.SWAGGER_REST_OPERATION);
    when(operationMeta.getExtData(RestConst.SWAGGER_REST_OPERATION)).thenReturn(swaggerRestOperation);
    when(invocation.getEndpoint()).thenReturn(endpoint);
    String contentType = httpResponse.getHeader("Content-Type");
    ProduceProcessor produceProcessor = mock(ProduceProcessor.class);
    when(swaggerRestOperation.findProduceProcessor(contentType)).thenReturn(produceProcessor);
    this.setCseContext(invocation, request);
  } catch (Exception ex) {
    status = true;
  }
  Assert.assertFalse(status);
}

public static Handler<HttpClientResponse> json(
  CompletableFuture<JsonResponse> completed) {

  return response -> {
    response.bodyHandler(buffer -> {
      try {
        int statusCode = response.statusCode();
        String body = BufferHelper.stringFromBuffer(buffer);

        System.out.println(String.format("Response: %s", body));
        log.debug(String.format("Response: %s", body));

        completed.complete(new JsonResponse(statusCode, body));

      } catch(Exception e) {
        completed.completeExceptionally(e);
      }
    });
  };
}

public static Handler<HttpClientResponse> jsonErrors(
  CompletableFuture<JsonErrorResponse> completed) {

  return response -> {
    response.bodyHandler(buffer -> {
      try {
        int statusCode = response.statusCode();
        String body = BufferHelper.stringFromBuffer(buffer);

        completed.complete(new JsonErrorResponse(statusCode, body));

      } catch (Exception e) {
        completed.completeExceptionally(e);
      }
    });
  };
}

public static Handler<HttpClientResponse> text(
  CompletableFuture<TextResponse> completed) {

  return response -> {
      int statusCode = response.statusCode();

      response.bodyHandler(buffer -> {
        try {
          String body = BufferHelper.stringFromBuffer(buffer);

          completed.complete(new TextResponse(statusCode, body));

        } catch (Exception e) {
          completed.completeExceptionally(e);
        }
      });
  };
}

/**
 * Proxies the specified HTTP request, enriching its headers with authentication information.
 *
 * @param userId  the ID of the user making the request.
 * @param origReq the original request (i.e., {@link RoutingContext#request()}.
 * @param origRes the original response (i.e., {@link RoutingContext#request()}.
 */
public void proxyUserRequest(final String userId,
                             final HttpServerRequest origReq,
                             final HttpServerResponse origRes) {
    final Handler<HttpClientResponse> proxiedResHandler = proxiedRes -> {
        origRes.setChunked(true);
        origRes.setStatusCode(proxiedRes.statusCode());
        origRes.headers().setAll(proxiedRes.headers());
        proxiedRes.handler(origRes::write);
        proxiedRes.endHandler(v -> origRes.end());
    };

    final HttpClientRequest proxiedReq;
    proxiedReq = httpClient.request(origReq.method(), port, host, origReq.uri(), proxiedResHandler);
    proxiedReq.setChunked(true);
    proxiedReq.headers().add(X_FORWARDED_PROTO, getHeader(origReq, X_FORWARDED_PROTO, origReq.scheme()));
    proxiedReq.headers().add(X_FORWARDED_FOR, getHeader(origReq, X_FORWARDED_FOR, origReq.remoteAddress().host()));
    proxiedReq.headers().addAll(origReq.headers());
    injectRutHeader(proxiedReq, userId);
    origReq.handler(proxiedReq::write);
    origReq.endHandler(v -> proxiedReq.end());
}

public void put(URL url,
                Object body,
                String tenantId,
                Handler<HttpClientResponse> responseHandler) {

  HttpClientRequest request = client.putAbs(url.toString(), responseHandler);

  request.headers().add("Accept","application/json, text/plain");
  request.headers().add("Content-type","application/json");

  if(tenantId != null) {
    request.headers().add(TENANT_HEADER, tenantId);
  }

  request.end(Json.encodePrettily(body));
}

protected Observable<Void> prepareContainer(TestContext context) {
    return just((Void) null)
            .flatMap(aVoid -> vertxContext.verticle().getNodeStats().forceUpdate(vertxContext))
            .flatMap(aVoid -> vertxContext.verticle().getClusterInfo().forceRefresh(vertxContext))
            .flatMap(new WaitForCluster(vertxContext))
            .flatMap(new PostAccount(httpClient, accountName, authAdmin))
            .map(new HttpClientResponseHeaderLogger())
            .map(new AssertHttpClientResponseStatusCode(context, HTTP_NO_CONTENT))
            .map(new ToVoid<HttpClientResponse>())
            .flatMap(new PutContainer(httpClient, accountName, containerName, authNonAdmin))
            .map(new HttpClientResponseHeaderLogger())
            .map(new AssertHttpClientResponseStatusCode(context, HTTP_CREATED))
            .map(new ToVoid<HttpClientResponse>())
            .count()
            .map(new ToVoid<Integer>());
}

@Test
public void testHeadContainerImplicitAllow(TestContext context) {

    Producer authNonAdmin0 = httpBasic("user", "user");

    Async async = context.async();
    just((Void) null)
            .flatMap(new PostAccount(httpClient, accountName, authAdmin))
            .map(new HttpClientResponseHeaderLogger())
            .map(new AssertHttpClientResponseStatusCode(context, HTTP_NO_CONTENT))
            .map(new ToVoid<HttpClientResponse>())
            .flatMap(new PutContainer(httpClient, accountName, containerName, authNonAdmin0))
            .map(new HttpClientResponseHeaderLogger())
            .map(new AssertHttpClientResponseStatusCode(context, HTTP_CREATED))
            .map(new ToVoid<HttpClientResponse>())
            .flatMap(new HeadContainer(httpClient, accountName, containerName, authNonAdmin0))
            .map(new HttpClientResponseHeaderLogger())
            .map(new AssertHttpClientResponseStatusCode(context, HTTP_NO_CONTENT))
            .map(new ToVoid<HttpClientResponse>())
            .subscribe(new TestSubscriber(context, async));
}

private void handle204(HttpClientResponse res, String msg,
  Handler<AsyncResult<Void>> future) {
  Buffer body = Buffer.buffer();
  res.handler(body::appendBuffer);
  res.endHandler(d -> {
    if (res.statusCode() == 204) {
      future.handle(Future.succeededFuture());
    } else {
      String m = msg + " HTTP error "
        + Integer.toString(res.statusCode()) + "\n"
        + body.toString();
      logger.error(m);
      future.handle(Future.failedFuture(m));
    }
  });

}

public DigestBlob(HttpClientResponse httpClientResponse) {
    super(httpClientResponse);
    digests = new HashMap<>();
    BaseEncoding baseEncoding = base64();
    MultiMap headers = httpClientResponse.headers();
    for (String headerName : headers.names()) {
        Matcher matcher = COMPUTED_DIGEST.matcher(headerName);
        if (matcher.find()) {
            String digestName = matcher.group(1);
            Optional<MessageDigestFactory> oMessageDigestFactory = fromValueIfExists(digestName);
            if (oMessageDigestFactory.isPresent()) {
                MessageDigestFactory messageDigestFactory = oMessageDigestFactory.get();
                withDigest(messageDigestFactory, baseEncoding.decode(headers.get(headerName)));
            }
        }
    }
}

private static final Future<HttpClientResponse> updateCredentials(final String authId, final String type, final JsonObject requestPayload, final int expectedResult) {

        final Future<HttpClientResponse> result = Future.future();
        final String uri = String.format(TEMPLATE_URI_CREDENTIALS_INSTANCE, authId, type);

        final HttpClientRequest req = vertx.createHttpClient().put(getPort(), HOST, uri)
                .putHeader(HttpHeaders.CONTENT_TYPE, HttpUtils.CONTENT_TYPE_JSON)
                .handler(response -> {
                    if (response.statusCode() == expectedResult) {
                        result.complete(response);
                    } else {
                        result.fail("update credentials failed, expected status code " + expectedResult + " but got " + response.statusCode());
                    }
                })
                .exceptionHandler(result::fail);

        if (requestPayload == null) {
            req.end();
        } else {
            req.end(requestPayload.encodePrettily());
        }
        return result;
    }

@Test
public void testListContainerExplicitAllow(TestContext context) {
    Producer authNonAdmin0 = httpBasic("user", "user");

    Async async = context.async();
    just((Void) null)
            .flatMap(new PostAccount(httpClient, accountName, authAdmin))
            .map(new HttpClientResponseHeaderLogger())
            .map(new AssertHttpClientResponseStatusCode(context, HTTP_NO_CONTENT))
            .map(new ToVoid<HttpClientResponse>())
            .flatMap(new PutContainer(httpClient, accountName, containerName, authNonAdmin0))
            .map(new HttpClientResponseHeaderLogger())
            .map(new AssertHttpClientResponseStatusCode(context, HTTP_CREATED))
            .map(new ToVoid<HttpClientResponse>())
            .flatMap(new RefreshIndex(httpClient, authAdmin))
            .flatMap(new GetContainer(httpClient, accountName, containerName, authNonAdmin0))
            .map(new HttpClientResponseHeaderLogger())
            .map(new AssertHttpClientResponseStatusCode(context, HTTP_OK))
            .flatMap(new HttpClientResponseBodyBuffer())
            .map(new HttpBodyLogger())
            .map(new ToVoid<Buffer>())
            .subscribe(new TestSubscriber(context, async));
}

@Override
public Observable<Buffer> call(HttpClientResponse httpClientResponse) {
    BufferWriteEndableWriteStream bufferWriteEndableWriteStream = new BufferWriteEndableWriteStream();
    return AsyncIO.pump(httpClientResponse, bufferWriteEndableWriteStream)
            .map(aVoid -> bufferWriteEndableWriteStream.toBuffer())
            .doOnNext(buffer -> {
                int statusCode = httpClientResponse.statusCode();
                boolean ok = false;
                for (int expectedStatusCode : expectedStatusCodes) {
                    if (statusCode == expectedStatusCode) {
                        ok = true;
                        break;
                    }
                }
                if (!ok && expectedStatusCodes.length <= 0) {
                    ok = true;
                }
                if (!ok) {
                    throw new HttpClientResponseException(httpClientResponse, buffer);
                }
            });
}

@Override
public Observable<HttpClientResponse> call(Void aVoid) {
    return auth.toHttpAuthorization()
            .flatMap(new Func1<String, Observable<HttpClientResponse>>() {
                @Override
                public Observable<HttpClientResponse> call(String s) {
                    StringBuilder urlBuilder = new StringBuilder();
                    urlBuilder = urlBuilder.append("/openstackswift001/" + accountName + "/" + containerName + "?destroy=1");
                    ObservableFuture<HttpClientResponse> handler = RxHelper.observableFuture();
                    HttpClientRequest httpClientRequest =
                            httpClient.delete(urlBuilder.toString(), handler::complete)
                                    .exceptionHandler(handler::fail)
                                    .setTimeout(5000)
                                    .putHeader(AUTHORIZATION, s);
                    httpClientRequest.end();
                    return handler
                            .single();
                }
            });

}

@Override
public Observable<HttpClientResponse> call(Void aVoid) {
    return auth.toHttpAuthorization()
            .flatMap(new Func1<String, Observable<HttpClientResponse>>() {
                @Override
                public Observable<HttpClientResponse> call(String s) {
                    ObservableFuture<HttpClientResponse> handler = RxHelper.observableFuture();
                    httpClient.head("/openstackswift001/" + accountName, handler::complete)
                            .exceptionHandler(handler::fail)
                            .setTimeout(10000)
                            .putHeader(AUTHORIZATION, s)
                            .end();
                    return handler
                            .single();
                }
            });
}

@Test
public void testHeadContainerAsAdmin(TestContext context) {
    Async async = context.async();
    just((Void) null)
            .flatMap(new PostAccount(httpClient, accountName, authAdmin))
            .map(new HttpClientResponseHeaderLogger())
            .map(new AssertHttpClientResponseStatusCode(context, HTTP_NO_CONTENT))
            .map(new ToVoid<HttpClientResponse>())
            .flatMap(new PutContainer(httpClient, accountName, containerName, authAdmin))
            .map(new HttpClientResponseHeaderLogger())
            .map(new AssertHttpClientResponseStatusCode(context, HTTP_CREATED))
            .map(new ToVoid<HttpClientResponse>())
            .flatMap(new HeadContainer(httpClient, accountName, containerName, authAdmin))
            .map(new HttpClientResponseHeaderLogger())
            .map(new AssertHttpClientResponseStatusCode(context, HTTP_NO_CONTENT))
            .map(new ToVoid<HttpClientResponse>())
            .subscribe(new TestSubscriber(context, async));
}

@Override
public Observable<HttpClientResponse> call(Void aVoid) {
    return auth.toHttpAuthorization()
            .flatMap(new Func1<String, Observable<HttpClientResponse>>() {
                @Override
                public Observable<HttpClientResponse> call(String s) {
                    ObservableFuture<HttpClientResponse> handler = RxHelper.observableFuture();
                    HttpClientRequest httpClientRequest =
                            httpClient.post("/openstackswift001/" + accountName + "/" + containerName + "/" + objectName, handler::complete)
                                    .exceptionHandler(handler::fail)
                                    .setTimeout(5000)
                                    .putHeader(AUTHORIZATION, s);
                    for (String entry : headers.keySet()) {
                        httpClientRequest = httpClientRequest.putHeader(entry, headers.get(entry));
                    }
                    httpClientRequest.end();
                    return handler
                            .single();
                }
            });

}

@Override
public Observable<HttpClientResponse> call(Void aVoid) {
    return auth.toHttpAuthorization()
            .flatMap(new Func1<String, Observable<HttpClientResponse>>() {
                @Override
                public Observable<HttpClientResponse> call(String s) {
                    ObservableFuture<HttpClientResponse> handler = RxHelper.observableFuture();
                    HttpClientRequest httpClientRequest =
                            httpClient.put("/openstackswift001/" + accountName + "/" + containerName, handler::complete)
                                    .exceptionHandler(handler::fail)
                                    .setTimeout(10000)
                                    .putHeader(AUTHORIZATION, s);
                    for (String entry : headers.keySet()) {
                        httpClientRequest = httpClientRequest.putHeader(entry, headers.get(entry));
                    }
                    httpClientRequest.end();
                    return handler
                            .single();
                }
            });

}

@Override
public Observable<HttpClientResponse> call(Void aVoid) {
    return auth.toHttpAuthorization()
            .flatMap(s -> {
                ObservableFuture<HttpClientResponse> handler = RxHelper.observableFuture();
                HttpClientRequest httpClientRequest =
                        httpClient.post("/admin/001/run_jobs", handler::complete)
                                .exceptionHandler(handler::fail)
                                .setTimeout(1000)
                                .putHeader(AUTHORIZATION, s);
                for (String entry : headers.keySet()) {
                    httpClientRequest = httpClientRequest.putHeader(entry, headers.get(entry));
                }
                httpClientRequest.end();
                return handler
                        .single();
            });

}

protected Observable<Void> prepareContainer(TestContext context) {

        return just((Void) null)
                .flatMap(new PostAccount(httpClient, accountName, authAdmin))
                .map(new HttpClientResponseHeaderLogger())
                .map(new AssertHttpClientResponseStatusCode(context, HTTP_NO_CONTENT))
                .map(new ToVoid<HttpClientResponse>())
                .flatMap(new PutContainer(httpClient, accountName, containerName, authNonAdmin))
                .map(new HttpClientResponseHeaderLogger())
                .map(new AssertHttpClientResponseStatusCode(context, HTTP_CREATED))
                .map(new ToVoid<HttpClientResponse>())
                .flatMap(new PostContainer(httpClient, accountName, containerName, authNonAdmin)
                        .setHeader(X_ADD_CONTAINER_META_PREFIX + X_MAX_OBJECT_REVISIONS, valueOf(3)))
                .map(new HttpClientResponseHeaderLogger())
                .map(new AssertHttpClientResponseStatusCode(context, HTTP_NO_CONTENT))
                .map(new ToVoid<HttpClientResponse>());
    }

@Override
public Observable<HttpClientResponse> call(Void aVoid) {
    return auth.toHttpAuthorization()
            .flatMap(new Func1<String, Observable<HttpClientResponse>>() {
                @Override
                public Observable<HttpClientResponse> call(String s) {
                    ObservableFuture<HttpClientResponse> handler = RxHelper.observableFuture();
                    final HttpClientRequest httpClientRequest =
                            httpClient.put("/openstackswift001/" + accountName + "/" + containerName + "/" + objectName, handler::complete)
                                    .exceptionHandler(handler::fail)
                                    .putHeader(AUTHORIZATION, s);
                    for (String entry : headers.keySet()) {
                        httpClientRequest.putHeader(entry, headers.get(entry));
                    }

                    httpClientRequest.setChunked(isChunked());

                    Observable<HttpClientResponse> clientResponse = handler;

                    HttpClientRequestEndableWriteStream adapter = new HttpClientRequestEndableWriteStream(httpClientRequest);

                    return combineSinglesDelayError(pump(data, adapter), clientResponse, (aVoid1, httpClientResponse) -> httpClientResponse);
                }
            });

}

@Override
public Observable<HttpClientResponse> call(Void aVoid) {
    return auth.toHttpAuthorization()
            .flatMap(new Func1<String, Observable<HttpClientResponse>>() {
                @Override
                public Observable<HttpClientResponse> call(String s) {
                    ObservableFuture<HttpClientResponse> handler = RxHelper.observableFuture();
                    HttpClientRequest httpClientRequest =
                            httpClient.head("/openstackswift001/" + accountName + "/" + containerName, handler::complete)
                                    .exceptionHandler(handler::fail)
                                    .setTimeout(10000)
                                    .putHeader(AUTHORIZATION, s);
                    httpClientRequest.end();
                    return handler
                            .single();
                }
            });
}

@Override
public Observable<HttpClientResponse> call(Void aVoid) {
    return auth.toHttpAuthorization()
            .flatMap(new Func1<String, Observable<HttpClientResponse>>() {
                @Override
                public Observable<HttpClientResponse> call(String s) {
                    ObservableFuture<HttpClientResponse> handler = RxHelper.observableFuture();
                    HttpClientRequest httpClientRequest =
                            httpClient.put("/openstackswift001/" + accountName + "/" + containerName + "/" + objectName, handler::complete)
                                    .exceptionHandler(handler::fail)
                                    .setTimeout(20000)
                                    .putHeader(AUTHORIZATION, s);
                    for (String entry : headers.keySet()) {
                        httpClientRequest = httpClientRequest.putHeader(entry, headers.get(entry));
                    }
                    httpClientRequest.setChunked(isChunked());
                    httpClientRequest.end(buffer(data));
                    return handler
                            .single();
                }
            });

}

@Override
public Observable<HttpClientResponse> call(Void aVoid) {
    return auth.toHttpAuthorization()
            .flatMap(new Func1<String, Observable<HttpClientResponse>>() {
                @Override
                public Observable<HttpClientResponse> call(String s) {
                    ObservableFuture<HttpClientResponse> handler = RxHelper.observableFuture();
                    HttpClientRequest httpClientRequest =
                            httpClient.post("/openstackswift001/" + accountName, handler::complete)
                                    .exceptionHandler(handler::fail)
                                    .setTimeout(10000)
                                    .putHeader(AUTHORIZATION, s);
                    for (String entry : headers.keySet()) {
                        httpClientRequest = httpClientRequest.putHeader(entry, headers.get(entry));
                    }
                    httpClientRequest.end();
                    return handler
                            .single();
                }
            });

}

@Override
public HttpClientResponse call(HttpClientResponse httpClientResponse) {
    out.println("Assert #" + assertIndex);
    MultiMap headers = httpClientResponse.headers();
    String etag = headers.get(ETAG);
    String contentMd5 = headers.get(CONTENT_MD5);
    String contentSha512 = headers.get(X_CONTENT_SHA512);
    String versionId = headers.get(X_CONTENT_VERSION);
    String contentLength = headers.get(CONTENT_LENGTH);
    String acceptRanges = headers.get(ACCEPT_RANGES);
    String lastModified = headers.get(LAST_MODIFIED);
    String date = headers.get(DATE);
    String serverSideEncryption = headers.get(X_SERVER_SIDE_ENCRYPTION);

    assertEquals(context, base16().lowerCase().encode(dataMd5), etag);
    assertEquals(context, base64().encode(dataMd5), contentMd5);
    assertEquals(context, base64().encode(dataSha512), contentSha512);
    assertEquals(context, expectedVersion, parseLong(versionId));
    //VertxAssert.assertEquals(context, expectedContentLength, Long.parseLong(contentLength));
    assertEquals(context, "none", acceptRanges);
    assertNotNull(context, lastModified);
    assertNotNull(context, date);
    assertEquals(context, this.serverSideEncryption, parseBoolean(serverSideEncryption));

    return httpClientResponse;
}

protected Observable<Void> testPermission(TestContext context, final String accountName, String username, String password) {

        Producer auth = httpBasic(username, password);

        return just((Void) null)
                .flatMap(new PostAccount(httpClient, accountName, auth))
                .map(new HttpClientResponseHeaderLogger())
                .map(new AssertHttpClientResponseStatusCode(context, HTTP_FORBIDDEN))
                .map(new ToVoid<HttpClientResponse>())
                .flatMap(new RefreshIndex(httpClient, authAdmin))
                .flatMap(new GetAccount(httpClient, accountName, auth)
                        .setMediaTypes(JSON_UTF_8))
                .map(new HttpClientResponseHeaderLogger())
                .map(new AssertHttpClientResponseStatusCode(context, "".equals(username) ? HTTP_FORBIDDEN : HTTP_NOT_FOUND))
                .map(new ToVoid<HttpClientResponse>())
                .flatMap(new HeadAccount(httpClient, accountName, auth))
                .map(new HttpClientResponseHeaderLogger())
                .map(new AssertHttpClientResponseStatusCode(context, HTTP_FORBIDDEN))
                .map(new ToVoid<HttpClientResponse>());
    }

protected Observable<Void> prepareContainer(TestContext context) {

        return just((Void) null)
                .flatMap(aVoid -> vertxContext.verticle().getNodeStats().forceUpdate(vertxContext))
                .flatMap(aVoid -> vertxContext.verticle().getClusterInfo().forceRefresh(vertxContext))
                .flatMap(new WaitForCluster(vertxContext))
                .flatMap(new RefreshIndex(httpClient, authAdmin))
                .flatMap(new PostAccount(httpClient, accountName, authAdmin))
                .map(new HttpClientResponseHeaderLogger())
                .map(new AssertHttpClientResponseStatusCode(context, HTTP_NO_CONTENT))
                .map(new ToVoid<HttpClientResponse>())
                .flatMap(new PutContainer(httpClient, accountName, containerName, authNonAdmin))
                .map(new HttpClientResponseHeaderLogger())
                .map(new AssertHttpClientResponseStatusCode(context, HTTP_CREATED))
                .map(new ToVoid<HttpClientResponse>())
                .flatMap(new PostContainer(httpClient, accountName, containerName, authNonAdmin)
                        .setHeader(X_ADD_CONTAINER_META_PREFIX + X_MAX_OBJECT_REVISIONS, valueOf(3)))
                .map(new HttpClientResponseHeaderLogger())
                .map(new AssertHttpClientResponseStatusCode(context, HTTP_NO_CONTENT))
                .map(new ToVoid<HttpClientResponse>());
    }

private static Future<HttpClientResponse> addCredentials(final JsonObject requestPayload, final String contentType, final int expectedStatusCode) {

        final Future<HttpClientResponse> result = Future.future();
        final HttpClientRequest req = vertx.createHttpClient().post(getPort(), HOST, URI_ADD_CREDENTIALS)
                .putHeader(HttpHeaders.CONTENT_TYPE, contentType)
                .handler(response -> {
                    if (response.statusCode() == expectedStatusCode) {
                        result.complete(response);
                    } else {
                        result.fail("add credentials failed, expected status code " + expectedStatusCode + " but got " + response.statusCode());
                    }
                }).exceptionHandler(result::fail);

        if (requestPayload == null) {
            req.end();
        } else {
            req.end(requestPayload.encodePrettily());
        }
        return result;
    }

default void handleError(HttpClientResponse response) {
    response.bodyHandler(bh -> {
        String body = bh.toString();
        if (log.isDebugEnabled()) {
            log.debug(body);
        }

        log.error("Request failed with statusCode {" + response.statusCode() + "} statusMessage {" + response.statusMessage() + "} {" + body
                + "} for method {" + getMethod() + "} and uri {" + getUri() + "}");

        // Try to parse the body data and fail using the extracted exception.
        try {
            JsonObject responseObj = new JsonObject(body);
            getFuture().fail(new MeshRestClientJsonObjectException(response.statusCode(), response.statusMessage(), responseObj));
            return;
        } catch (Exception e) {
            if (log.isDebugEnabled()) {
                log.debug("Could not deserialize response {" + body + "}.", e);
            }
        }

        getFuture().fail(new MeshRestClientMessageException(response.statusCode(), response.statusMessage()));
        return;
    });

}

protected Observable<Void> prepareContainer(TestContext context) {

        return just((Void) null)
                .flatMap(aVoid -> vertxContext.verticle().getNodeStats().forceUpdate(vertxContext))
                .flatMap(aVoid -> vertxContext.verticle().getClusterInfo().forceRefresh(vertxContext))
                .flatMap(new WaitForCluster(vertxContext))
                .flatMap(new PostAccount(httpClient, accountName, authAdmin))
                .map(new HttpClientResponseHeaderLogger())
                .map(new AssertHttpClientResponseStatusCode(context, HTTP_NO_CONTENT))
                .map(new ToVoid<HttpClientResponse>())
                .flatMap(new PutContainer(httpClient, accountName, containerName, authNonAdmin))
                .map(new HttpClientResponseHeaderLogger())
                .map(new AssertHttpClientResponseStatusCode(context, HTTP_CREATED))
                .count()
                .map(new ToVoid<Integer>());
    }

@Test
public void testListContainerNoPermissions(TestContext context) {
    Async async = context.async();
    just((Void) null)
            .flatMap(new PostAccount(httpClient, accountName, authAdmin))
            .map(new HttpClientResponseHeaderLogger())
            .map(new AssertHttpClientResponseStatusCode(context, HTTP_NO_CONTENT))
            .map(new ToVoid<HttpClientResponse>())
            .flatMap(new PutContainer(httpClient, accountName, containerName, authAdmin))
            .map(new HttpClientResponseHeaderLogger())
            .map(new AssertHttpClientResponseStatusCode(context, HTTP_CREATED))
            .map(new ToVoid<HttpClientResponse>())
            .flatMap(new RefreshIndex(httpClient, authAdmin))
            .flatMap(new GetContainer(httpClient, accountName, containerName, authWithoutPerms))
            .map(new HttpClientResponseHeaderLogger())
            .map(new AssertHttpClientResponseStatusCode(context, HTTP_FORBIDDEN))
            .map(new ToVoid<HttpClientResponse>())
            .subscribe(new TestSubscriber(context, async));
}

@Test
public void testHeadContainerNoPermissions(TestContext context) {
    Async async = context.async();
    just((Void) null)
            .flatMap(new PostAccount(httpClient, accountName, authAdmin))
            .map(new HttpClientResponseHeaderLogger())
            .map(new AssertHttpClientResponseStatusCode(context, HTTP_NO_CONTENT))
            .map(new ToVoid<HttpClientResponse>())
            .flatMap(new PutContainer(httpClient, accountName, containerName, authAdmin))
            .map(new HttpClientResponseHeaderLogger())
            .map(new AssertHttpClientResponseStatusCode(context, HTTP_CREATED))
            .map(new ToVoid<HttpClientResponse>())
            .flatMap(new HeadContainer(httpClient, accountName, containerName, authWithoutPerms))
            .map(new HttpClientResponseHeaderLogger())
            .map(new AssertHttpClientResponseStatusCode(context, HTTP_FORBIDDEN))
            .map(new ToVoid<HttpClientResponse>())
            .subscribe(new TestSubscriber(context, async));
}

@Test
public void testListContainerAsNonAdmin(TestContext context) {
    Async async = context.async();
    just((Void) null)
            .flatMap(new PostAccount(httpClient, accountName, authAdmin))
            .map(new HttpClientResponseHeaderLogger())
            .map(new AssertHttpClientResponseStatusCode(context, HTTP_NO_CONTENT))
            .map(new ToVoid<HttpClientResponse>())
            .flatMap(new PutContainer(httpClient, accountName, containerName, authAdmin))
            .map(new HttpClientResponseHeaderLogger())
            .map(new AssertHttpClientResponseStatusCode(context, HTTP_CREATED))
            .map(new ToVoid<HttpClientResponse>())
            .flatMap(new RefreshIndex(httpClient, authAdmin))
            .flatMap(new GetContainer(httpClient, accountName, containerName, authWithoutPerms))
            .map(new HttpClientResponseHeaderLogger())
            .map(new AssertHttpClientResponseStatusCode(context, HTTP_FORBIDDEN))
            .map(new ToVoid<HttpClientResponse>())
            .subscribe(new TestSubscriber(context, async));
}

@Test
public void testListContainerAsAdmin(TestContext context) {
    Async async = context.async();
    just((Void) null)
            .flatMap(new PostAccount(httpClient, accountName, authAdmin))
            .map(new HttpClientResponseHeaderLogger())
            .map(new AssertHttpClientResponseStatusCode(context, HTTP_NO_CONTENT))
            .map(new ToVoid<HttpClientResponse>())
            .flatMap(new PutContainer(httpClient, accountName, containerName, authAdmin))
            .map(new HttpClientResponseHeaderLogger())
            .map(new AssertHttpClientResponseStatusCode(context, HTTP_CREATED))
            .map(new ToVoid<HttpClientResponse>())
            .flatMap(new RefreshIndex(httpClient, authAdmin))
            .flatMap(new GetContainer(httpClient, accountName, containerName, authAdmin))
            .map(new HttpClientResponseHeaderLogger())
            .map(new AssertHttpClientResponseStatusCode(context, HTTP_OK))
            .map(new ToVoid<HttpClientResponse>())
            .subscribe(new TestSubscriber(context, async));
}

@SuppressWarnings("unchecked")
private <T> Handler<RestResponse> syncHandler(CountDownLatch countDownLatch, Class<T> cls,
    Holder<T> holder) {
  return restResponse -> {
    RequestContext requestContext = restResponse.getRequestContext();
    HttpClientResponse response = restResponse.getResponse();
    if (response == null) {
      // 请求失败，触发请求SC的其他实例
      if (!requestContext.isRetry()) {
        retry(requestContext, syncHandler(countDownLatch, cls, holder));
      } else {
        countDownLatch.countDown();
      }
      return;
    }
    response.bodyHandler(
        bodyBuffer -> {
          if (cls.getName().equals(HttpClientResponse.class.getName())) {
            holder.value = (T) response;
            countDownLatch.countDown();
            return;
          }
          try {
            holder.value =
                JsonUtils.readValue(bodyBuffer.getBytes(), cls);
          } catch (Exception e) {
            LOGGER.warn("read value failed and response message is {}",
                bodyBuffer.toString());
          }
          countDownLatch.countDown();
        });
  };
}

private <T> Handler<RestResponse> syncHandlerForInstances(CountDownLatch countDownLatch,
    MicroserviceInstances mInstances) {
  return restResponse -> {
    RequestContext requestContext = restResponse.getRequestContext();
    HttpClientResponse response = restResponse.getResponse();
    if (response == null) {
      // 请求失败，触发请求SC的其他实例
      if (!requestContext.isRetry()) {
        retry(requestContext, syncHandlerForInstances(countDownLatch, mInstances));
      } else {
        countDownLatch.countDown();
      }
      return;
    }
    response.bodyHandler(
        bodyBuffer -> {
          try {
            mInstances.setRevision(response.getHeader("X-Resource-Revision"));
            switch (response.statusCode()) {
              case 304:
                mInstances.setNeedRefresh(false);
                break;
              case 200:
                mInstances
                    .setInstancesResponse(JsonUtils.readValue(bodyBuffer.getBytes(), FindInstancesResponse.class));
                break;
              default:
                LOGGER.warn(bodyBuffer.toString());
                break;
            }
          } catch (Exception e) {
            LOGGER.warn("read value failed and response message is {}", bodyBuffer.toString());
          }
          countDownLatch.countDown();
        });
  };
}

@Override
public HeartbeatResponse heartbeat(String microserviceId, String microserviceInstanceId) {
  Holder<HttpClientResponse> holder = new Holder<>();
  IpPort ipPort = ipPortManager.getAvailableAddress();

  CountDownLatch countDownLatch = new CountDownLatch(1);
  RestUtils.put(ipPort,
      String.format(Const.REGISTRY_API.MICROSERVICE_HEARTBEAT, microserviceId, microserviceInstanceId),
      new RequestParam().setTimeout(ServiceRegistryConfig.INSTANCE.getHeartBeatRequestTimeout()),
      syncHandler(countDownLatch, HttpClientResponse.class, holder));

  try {
    countDownLatch.await();
    if (holder.value != null) {
      HeartbeatResponse response = new HeartbeatResponse();
      response.setMessage(holder.value.statusMessage());
      if (holder.value.statusCode() == Status.OK.getStatusCode()) {
        response.setOk(true);
        return response;
      }
      LOGGER.warn(holder.value.statusMessage());
      return response;
    }
  } catch (Exception e) {
    LOGGER.error("update microservice instance {}/{} heartbeat failed",
        microserviceId,
        microserviceInstanceId,
        e);
  }
  return null;
}