io.netty.handler.codec.socksx.v4.DefaultSocks4CommandResponse #com.linecorp.armeria.client.logging.LoggingClient源码实例Demo

public ClientBuilder newBuilder(String url) {
  return Clients.builder("gproto+" + url)
      .decorator(
          client ->
              new SimpleDecoratingHttpClient(client) {
                @Override
                public HttpResponse execute(ClientRequestContext ctx, HttpRequest req)
                    throws Exception {
                  // Many Google services do not support the standard application/grpc+proto
                  // header...
                  req =
                      req.withHeaders(
                          req.headers().toBuilder()
                              .set(HttpHeaderNames.CONTENT_TYPE, "application/grpc")
                              .build());
                  return delegate().execute(ctx, req);
                }
              })
      .decorator(credentialsDecorator.newAccessTokenDecorator())
      .decorator(BraveClient.newDecorator(tracing))
      .decorator(MetricCollectingClient.newDecorator(MetricLabels.grpcRequestLabeler()))
      .decorator(LoggingClient.builder().newDecorator());
}
 

@Provides
@Singleton
@GoogleApis
public static WebClient googleApisClient(
    Optional<MeterRegistry> meterRegistry, GcloudConfig config) {
  ClientFactory factory =
      meterRegistry
          .map(
              registry -> {
                ClientFactoryBuilder builder = ClientFactory.builder().meterRegistry(registry);
                if (config.getDisableEdns()) {
                  builder.domainNameResolverCustomizer(
                      dnsNameResolverBuilder -> dnsNameResolverBuilder.optResourceEnabled(false));
                }
                return builder.build();
              })
          .orElse(ClientFactory.ofDefault());
  return WebClient.builder("https://www.googleapis.com/")
      .factory(factory)
      .decorator(LoggingClient.builder().newDecorator())
      .build();
}
 

@SuppressWarnings("ConstructorLeaksThis")
public PublicKeysManager(@Provided Clock clock, String publicKeysUrl) {
  this.clock = clock;

  URI uri = URI.create(publicKeysUrl);
  path = uri.getPath();

  httpClient =
      WebClient.builder(uri.getScheme() + "://" + uri.getAuthority())
          .decorator(LoggingClient.builder().newDecorator())
          .decorator(RetryingClient.newDecorator(RetryRule.failsafe()))
          .build();
  keysCache =
      new AsyncRefreshingValue<>(
          this::refresh,
          CachedPublicKeys::expirationTime,
          CommonPools.workerGroup().next(),
          clock);
}
 

Builder computeStorageBuilder() {
  WebClientBuilder builder = WebClient.builder("http://" + hostPort())
    // Elasticsearch 7 never returns a response when receiving an HTTP/2 preface instead of the
    // more valid behavior of returning a bad request response, so we can't use the preface.
    //
    // TODO: find or raise a bug with Elastic
    .factory(ClientFactory.builder().useHttp2Preface(false).build());

  if (Boolean.parseBoolean(System.getenv("ES_DEBUG"))) {
    builder.decorator(c -> LoggingClient.builder()
      .requestLogLevel(LogLevel.INFO)
      .successfulResponseLogLevel(LogLevel.INFO).build(c));
  }
  WebClient client = builder.build();
  return ElasticsearchStorage.newBuilder(() -> client).index("zipkin-test").flushOnWrites(true);
}
 

private static WebClient newLoadBalancingClient() throws ExecutionException, InterruptedException {
    // Send HTTP health check requests to '/internal/l7check' every 10 seconds.
    final HealthCheckedEndpointGroup healthCheckedGroup =
            HealthCheckedEndpointGroup.builder(animationGroup, "/internal/l7check")
                                      .protocol(SessionProtocol.HTTP)
                                      .retryInterval(Duration.ofSeconds(10))
                                      .build();

    // Wait until the initial health check is finished.
    healthCheckedGroup.whenReady().get();

    return WebClient.builder(SessionProtocol.HTTP, healthCheckedGroup)
                    // Disable timeout to serve infinite streaming response.
                    .responseTimeoutMillis(0)
                    .decorator(LoggingClient.newDecorator())
                    .build();
}
 

@BeforeEach
void setUp() {
    requestLogQueue.clear();
    final DecoratingHttpClientFunction requestLogRecorder = (delegate, ctx, req) -> {
        ctx.log().whenComplete().thenAccept(requestLogQueue::add);
        return delegate.execute(ctx, req);
    };

    final URI uri = server.httpUri(GrpcSerializationFormats.PROTO);
    blockingStub = Clients.builder(uri)
                          .maxResponseLength(MAX_MESSAGE_SIZE)
                          .decorator(LoggingClient.builder().newDecorator())
                          .decorator(requestLogRecorder)
                          .build(TestServiceBlockingStub.class);
    asyncStub = Clients.builder(uri.getScheme(), server.httpEndpoint())
                       .decorator(LoggingClient.builder().newDecorator())
                       .decorator(requestLogRecorder)
                       .build(TestServiceStub.class);
}
 

@Test
void test() {
    final TestServiceBlockingStub client =
            Clients.builder("gproto+http://127.0.0.1:1/")
                   .decorator(LoggingClient.newDecorator())
                   .decorator(RetryingClient.newDecorator(
                           (ctx, cause) -> CompletableFuture.completedFuture(RetryDecision.noRetry())))
                   .build(TestServiceBlockingStub.class);

    assertThat(Clients.unwrap(client, TestServiceBlockingStub.class)).isSameAs(client);

    assertThat(Clients.unwrap(client, RetryingClient.class)).isInstanceOf(RetryingClient.class);
    assertThat(Clients.unwrap(client, LoggingClient.class)).isInstanceOf(LoggingClient.class);

    // The outermost decorator of the client must be returned,
    // because the search begins from outside to inside.
    // In the current setup, the outermost `Unwrappable` and `Client` are
    // `ArmeriaChannel` and `PooledHttpClient` respectively.
    assertThat(Clients.unwrap(client, Unwrappable.class)).isInstanceOf(ArmeriaChannel.class);
    assertThat(Clients.unwrap(client, Client.class)).isInstanceOf(PooledHttpClient.class);

    assertThat(Clients.unwrap(client, DecodingClient.class)).isNull();
}
 

@Test
void test() {
    final HelloService.Iface client =
            Clients.builder("tbinary+http://127.0.0.1:1/")
                   .decorator(LoggingClient.newDecorator())
                   .rpcDecorator(RetryingRpcClient.newDecorator(
                           RetryRuleWithContent.<RpcResponse>builder().thenNoRetry()))
                   .build(HelloService.Iface.class);

    assertThat(Clients.unwrap(client, HelloService.Iface.class)).isSameAs(client);

    assertThat(Clients.unwrap(client, RetryingRpcClient.class)).isInstanceOf(RetryingRpcClient.class);
    assertThat(Clients.unwrap(client, LoggingClient.class)).isInstanceOf(LoggingClient.class);

    // The outermost decorator of the client must be returned,
    // because the search begins from outside to inside.
    // In the current setup, the outermost `Unwrappable` and `Client` are
    // `THttpClientInvocationHandler` and `RetryingRpcClient` respectively.
    assertThat(Clients.unwrap(client, Unwrappable.class)).isInstanceOf(THttpClientInvocationHandler.class);
    assertThat(Clients.unwrap(client, Client.class)).isInstanceOf(RetryingRpcClient.class);

    assertThat(Clients.unwrap(client, CircuitBreakerRpcClient.class)).isNull();
}
 

@Test
void testSocks4BasicCase() throws Exception {
    final ClientFactory clientFactory = ClientFactory.builder().proxyConfig(
            ProxyConfig.socks4(socksProxyServer.address())).build();
    final WebClient webClient = WebClient.builder(H1C, backendServer.httpEndpoint())
                                         .factory(clientFactory)
                                         .decorator(LoggingClient.newDecorator())
                                         .build();
    final CompletableFuture<AggregatedHttpResponse> responseFuture =
            webClient.get(PROXY_PATH).aggregate();
    final AggregatedHttpResponse response = responseFuture.join();

    assertThat(response.status()).isEqualTo(OK);
    assertThat(response.contentUtf8()).isEqualTo(SUCCESS_RESPONSE);
    assertThat(numSuccessfulProxyRequests).isEqualTo(1);
    clientFactory.close();
}
 

@Test
void testSocks5BasicCase() throws Exception {
    final ClientFactory clientFactory = ClientFactory.builder().proxyConfig(
            ProxyConfig.socks5(socksProxyServer.address())).build();
    final WebClient webClient = WebClient.builder(H1C, backendServer.httpEndpoint())
                                         .factory(clientFactory)
                                         .decorator(LoggingClient.newDecorator())
                                         .build();
    final CompletableFuture<AggregatedHttpResponse> responseFuture =
            webClient.get(PROXY_PATH).aggregate();
    final AggregatedHttpResponse response = responseFuture.join();
    assertThat(response.status()).isEqualTo(OK);
    assertThat(response.contentUtf8()).isEqualTo(SUCCESS_RESPONSE);
    assertThat(numSuccessfulProxyRequests).isEqualTo(1);
    clientFactory.close();
}
 

@Test
void testH1CProxyBasicCase() throws Exception {
    final ClientFactory clientFactory = ClientFactory.builder().proxyConfig(
            ProxyConfig.connect(httpProxyServer.address())).build();
    final WebClient webClient = WebClient.builder(H1C, backendServer.httpEndpoint())
                                         .factory(clientFactory)
                                         .decorator(LoggingClient.newDecorator())
                                         .build();
    final CompletableFuture<AggregatedHttpResponse> responseFuture =
            webClient.get(PROXY_PATH).aggregate();
    final AggregatedHttpResponse response = responseFuture.join();
    assertThat(response.status()).isEqualTo(OK);
    assertThat(response.contentUtf8()).isEqualTo(SUCCESS_RESPONSE);
    assertThat(numSuccessfulProxyRequests).isEqualTo(1);
    clientFactory.close();
}
 

@Test
void testHttpsProxyBasicCase() throws Exception {
    final ClientFactory clientFactory =
            ClientFactory.builder().tlsNoVerify().proxyConfig(
                    ProxyConfig.connect(httpsProxyServer.address(), true)).build();
    final WebClient webClient = WebClient.builder(H1C, backendServer.httpEndpoint())
                                         .factory(clientFactory)
                                         .decorator(LoggingClient.newDecorator())
                                         .build();
    final CompletableFuture<AggregatedHttpResponse> responseFuture =
            webClient.get(PROXY_PATH).aggregate();
    final AggregatedHttpResponse response = responseFuture.join();
    assertThat(response.status()).isEqualTo(OK);
    assertThat(response.contentUtf8()).isEqualTo(SUCCESS_RESPONSE);
    assertThat(numSuccessfulProxyRequests).isEqualTo(1);
    clientFactory.close();
}
 

@Test
void testProxyWithH2C() throws Exception {
    final int numRequests = 5;
    final ClientFactory clientFactory = ClientFactory.builder().proxyConfig(
            ProxyConfig.socks4(socksProxyServer.address())).build();
    final WebClient webClient = WebClient.builder(SessionProtocol.H2C, backendServer.httpEndpoint())
                                         .factory(clientFactory)
                                         .decorator(LoggingClient.newDecorator())
                                         .build();

    final List<CompletableFuture<AggregatedHttpResponse>> responseFutures = new ArrayList<>();
    for (int i = 0; i < numRequests; i++) {
        responseFutures.add(webClient.get(PROXY_PATH).aggregate());
    }
    await().until(() -> responseFutures.stream().allMatch(CompletableFuture::isDone));
    assertThat(responseFutures.stream().map(CompletableFuture::join))
            .allMatch(response -> response.contentUtf8().equals(SUCCESS_RESPONSE));
    assertThat(numSuccessfulProxyRequests).isGreaterThanOrEqualTo(1);
    clientFactory.close();
}
 

@Test
void testProxy_protocolUpgrade_notSharableExceptionNotThrown() throws Exception {
    DYNAMIC_HANDLER.setWriteCustomizer((ctx, msg, promise) -> {
        ctx.write(new DefaultSocks4CommandResponse(Socks4CommandStatus.REJECTED_OR_FAILED), promise);
    });
    final ClientFactory clientFactory = ClientFactory.builder().proxyConfig(
            ProxyConfig.socks4(socksProxyServer.address())).build();
    final WebClient webClient = WebClient.builder(SessionProtocol.HTTP, backendServer.httpEndpoint())
                                         .factory(clientFactory)
                                         .decorator(LoggingClient.newDecorator())
                                         .build();
    final CompletableFuture<AggregatedHttpResponse> responseFuture =
            webClient.get(PROXY_PATH).aggregate();
    assertThatThrownBy(responseFuture::join).isInstanceOf(CompletionException.class)
                                            .hasCauseInstanceOf(UnprocessedRequestException.class)
                                            .hasRootCauseInstanceOf(ProxyConnectException.class);
    clientFactory.close();
}
 

@Test
void testProxy_connectionFailure_throwsException() throws Exception {
    final int unusedPort;
    try (ServerSocket ss = new ServerSocket(0)) {
        unusedPort = ss.getLocalPort();
    }

    final ClientFactory clientFactory = ClientFactory.builder().proxyConfig(
            ProxyConfig.socks4(new InetSocketAddress("127.0.0.1", unusedPort))).build();
    final WebClient webClient = WebClient.builder(H1C, backendServer.httpEndpoint())
                                         .factory(clientFactory)
                                         .decorator(LoggingClient.newDecorator())
                                         .build();
    final CompletableFuture<AggregatedHttpResponse> responseFuture =
            webClient.get(PROXY_PATH).aggregate();

    assertThatThrownBy(responseFuture::join).isInstanceOf(CompletionException.class)
                                            .hasMessageContaining("Connection refused")
                                            .hasCauseInstanceOf(UnprocessedRequestException.class)
                                            .hasRootCauseInstanceOf(ConnectException.class);
    clientFactory.close();
}
 

@Test
void testProxy_connectionTimeoutFailure_throwsException() throws Exception {
    DYNAMIC_HANDLER.setChannelReadCustomizer((ctx, msg) -> {
        if (msg instanceof DefaultSocks4CommandRequest) {
            ctx.channel().eventLoop().schedule(
                    () -> ctx.fireChannelRead(msg), 50, TimeUnit.MILLISECONDS);
        } else {
            ctx.fireChannelRead(msg);
        }
    });

    final ClientFactory clientFactory = ClientFactory.builder().proxyConfig(
            ProxyConfig.socks4(socksProxyServer.address())).connectTimeoutMillis(1).build();

    final WebClient webClient = WebClient.builder(H1C, backendServer.httpEndpoint())
                                         .factory(clientFactory)
                                         .decorator(LoggingClient.newDecorator())
                                         .build();
    final CompletableFuture<AggregatedHttpResponse> responseFuture =
            webClient.get(PROXY_PATH).aggregate();
    assertThatThrownBy(responseFuture::join).isInstanceOf(CompletionException.class)
                                            .hasCauseInstanceOf(UnprocessedRequestException.class)
                                            .hasRootCauseInstanceOf(ProxyConnectException.class);
    clientFactory.close();
}
 

@Test
void testProxy_responseFailure_throwsException() throws Exception {
    DYNAMIC_HANDLER.setWriteCustomizer((ctx, msg, promise) -> {
        ctx.write(new DefaultSocks4CommandResponse(Socks4CommandStatus.REJECTED_OR_FAILED), promise);
    });
    final ClientFactory clientFactory = ClientFactory.builder().proxyConfig(
            ProxyConfig.socks4(socksProxyServer.address())).build();
    final WebClient webClient = WebClient.builder(H1C, backendServer.httpEndpoint())
                                         .factory(clientFactory)
                                         .decorator(LoggingClient.newDecorator())
                                         .build();
    final CompletableFuture<AggregatedHttpResponse> responseFuture =
            webClient.get(PROXY_PATH).aggregate();

    assertThatThrownBy(responseFuture::join).isInstanceOf(CompletionException.class)
                                            .hasCauseInstanceOf(UnprocessedRequestException.class)
                                            .hasRootCauseInstanceOf(ProxyConnectException.class);
    clientFactory.close();
}
 

@Test
void testProxyServerImmediateClose() throws Exception {
    DYNAMIC_HANDLER.setChannelReadCustomizer((ctx, msg) -> {
        ctx.close();
    });
    try (ClientFactory clientFactory = ClientFactory.builder().proxyConfig(
            ProxyConfig.socks4(socksProxyServer.address())).build()) {
        final WebClient webClient = WebClient.builder(H1C, backendServer.httpEndpoint())
                                             .factory(clientFactory)
                                             .decorator(LoggingClient.newDecorator())
                                             .build();
        final CompletableFuture<AggregatedHttpResponse> responseFuture =
                webClient.get(PROXY_PATH).aggregate();
        await().timeout(Duration.ofSeconds(10)).until(responseFuture::isCompletedExceptionally);
        assertThatThrownBy(responseFuture::join).isInstanceOf(CompletionException.class)
                                                .hasCauseInstanceOf(UnprocessedRequestException.class)
                                                .hasRootCauseInstanceOf(ProxyConnectException.class);
    }
}
 

private WebClient client(RetryRuleWithContent<HttpResponse> retryRuleWithContent,
                         long responseTimeoutMillis,
                         long responseTimeoutForEach, int maxTotalAttempts) {
    final Function<? super HttpClient, RetryingClient> retryingDecorator =
            RetryingClient.builder(retryRuleWithContent)
                          .responseTimeoutMillisForEachAttempt(responseTimeoutForEach)
                          .useRetryAfter(true)
                          .maxTotalAttempts(maxTotalAttempts)
                          .newDecorator();

    return WebClient.builder(server.httpUri())
                    .factory(clientFactory)
                    .responseTimeoutMillis(responseTimeoutMillis)
                    .decorator(LoggingClient.newDecorator())
                    .decorator(retryingDecorator)
                    .build();
}
 

private HealthCheckedEndpointGroup build(Endpoint endpoint,
                                         Consumer<HealthCheckedEndpointGroupBuilder> customizer) {

    final HealthCheckedEndpointGroupBuilder builder =
            HealthCheckedEndpointGroup.builder(endpoint, HEALTH_CHECK_PATH);
    builder.withClientOptions(b -> {
        b.decorator(LoggingClient.newDecorator());
        b.decorator((delegate, ctx, req) -> {
            // Record when health check requests were sent.
            logs.add(req.headers());
            return delegate.execute(ctx, req);
        });
        return b;
    });
    customizer.accept(builder);
    return builder.build();
}
 

private HealthCheckedEndpointGroup build(HealthCheckedEndpointGroupBuilder builder) {
    // Specify backoff explicitly to disable jitter.
    builder.retryBackoff(Backoff.fixed(RETRY_INTERVAL.toMillis()));
    builder.withClientOptions(b -> {
        b.decorator(LoggingClient.newDecorator());
        b.decorator((delegate, ctx, req) -> {
            // Record when health check requests were sent.
            final Queue<RequestLog> healthCheckRequestLogs = this.healthCheckRequestLogs;
            if (healthCheckRequestLogs != null) {
                ctx.log().whenComplete().thenAccept(healthCheckRequestLogs::add);
            }
            return delegate.execute(ctx, req);
        });
        return b;
    });
    return builder.build();
}
 

@Test
void initialHealthCheckCanHaveEndpoints() throws Exception {
    serverOne.start();

    // even localhost usually takes long enough to resolve that this test would never work if the initial
    // health check didn't wait for localhost's DNS resolution.
    final int port = serverOne.httpPort();
    try (HealthCheckedEndpointGroup endpointGroup =
                 HealthCheckedEndpointGroup.builder(DnsAddressEndpointGroup.of("localhost", port),
                                                    HEALTH_CHECK_PATH)
                                           .retryInterval(Duration.ofHours(1))
                                           .withClientOptions(b -> {
                                               return b.decorator(LoggingClient.newDecorator());
                                           })
                                           .build()) {

        assertThat(endpointGroup.whenReady().get(10, TimeUnit.SECONDS)).hasSize(1);
    }
}
 

MyHttpClient(String uri, int maxLength) {
    final WebClientBuilder builder = WebClient.builder(serverExtension.httpUri().resolve(uri));

    final ContentPreviewerFactory factory = contentPreviewerFactory(maxLength);

    client = builder.decorator(ContentPreviewingClient.newDecorator(factory))
                    .decorator(LoggingClient.builder()
                                            .requestLogLevel(LogLevel.INFO)
                                            .successfulResponseLogLevel(LogLevel.INFO)
                                            .newDecorator())
                    .decorator((delegate, ctx, req) -> {
                        if (waitingFuture != null) {
                            ctx.log().whenComplete().thenAccept(waitingFuture::complete);
                        }
                        return delegate.execute(ctx, req);
                    }).build();
}
 

@Provides
static ServiceAccountsClient serviceAccountsClient(
    GcloudConfig config, GoogleCredentialsDecoratingClient.Factory credentialsDecorator) {
  return ArmeriaRetrofit.builder(
          WebClient.builder("https://iam.googleapis.com/v1/projects/" + config.getProject() + "/")
              .decorator(LoggingClient.builder().newDecorator())
              .decorator(credentialsDecorator.newAccessTokenDecorator())
              .build())
      .addConverterFactory(JacksonConverterFactory.create(OBJECT_MAPPER))
      .build()
      .create(ServiceAccountsClient.class);
}
 

@Override
protected CompletableFuture<AggregatedHttpResponse> fetchToken(Type type) {
  URI uri = URI.create(ComputeEngineCredentials.getTokenServerEncodedUrl());

  // In practice, this URL shouldn't change at runtime but it's not infeasible, and since this
  // shouldn't be executed often, just create a client every time.
  WebClient client =
      WebClient.builder("h1c://" + uri.getAuthority() + "/")
          .decorator(LoggingClient.builder().newDecorator())
          .build();
  return client
      .execute(RequestHeaders.of(HttpMethod.GET, uri.getPath(), METADATA_FLAVOR_HEADER, "Google"))
      .aggregate();
}
 

@Provides
@Singleton
static Function<HttpClient, LoggingClient> loggingClient(
    LoggingConfig config,
    @RequestHeaderSanitizer Set<Consumer<HttpHeadersBuilder>> requestHeaderSanitizers,
    @ResponseHeaderSanitizer Set<Consumer<HttpHeadersBuilder>> responseHeaderSanitizers) {
  LoggingClientBuilder builder = LoggingClient.builder();
  configureLoggingDecorator(builder, config, requestHeaderSanitizers, responseHeaderSanitizers);
  if (config.getLogAllClientRequests()) {
    builder.requestLogLevel(LogLevel.INFO);
    builder.successfulResponseLogLevel(LogLevel.INFO);
  }
  return builder::build;
}
 

@Provides
@Singleton
static YummlyApi yummlyApi(YummlyConfig config) {
  return ArmeriaRetrofit.builder(
          WebClient.builder("http://api.yummly.com/v1/api/")
              .addHttpHeader(HttpHeaderNames.of("X-Yummly-App-ID"), config.getApiId())
              .addHttpHeader(HttpHeaderNames.of("X-Yummly-App-Key"), config.getApiKey())
              .decorator(LoggingClient.builder().newDecorator())
              .build())
      .addCallAdapterFactory(GuavaCallAdapterFactory.create())
      .addConverterFactory(JacksonConverterFactory.create(OBJECT_MAPPER))
      .build()
      .create(YummlyApi.class);
}
 

@Test
void credentialsUnaryCall() {
    final TestServiceBlockingStub stub =
            // Explicitly construct URL to better test authority.
            Clients.builder("gproto+http://localhost:" + server.httpPort())
                   .decorator(LoggingClient.builder().newDecorator())
                   .build(TestServiceBlockingStub.class)
                   .withCallCredentials(
                           new CallCredentials() {
                               @Override
                               public void applyRequestMetadata(RequestInfo requestInfo,
                                                                Executor appExecutor,
                                                                MetadataApplier applier) {
                                   assertThat(requestInfo.getMethodDescriptor())
                                           .isEqualTo(TestServiceGrpc.getEmptyCallMethod());
                                   assertThat(requestInfo.getAuthority())
                                           .isEqualTo("localhost:" + server.httpPort());
                                   assertThat(requestInfo.getSecurityLevel())
                                           .isEqualTo(SecurityLevel.NONE);
                                   assertThat(appExecutor).isEqualTo(CommonPools.blockingTaskExecutor());

                                   CommonPools.blockingTaskExecutor().schedule(() -> {
                                       final Metadata metadata = new Metadata();
                                       metadata.put(TestServiceImpl.EXTRA_HEADER_KEY, "token");
                                       applier.apply(metadata);
                                   }, 100, TimeUnit.MILLISECONDS);
                               }

                               @Override
                               public void thisUsesUnstableApi() {
                               }
                           });

    assertThat(stub.emptyCall(EMPTY)).isNotNull();

    final HttpHeaders clientHeaders = CLIENT_HEADERS_CAPTURE.get();
    assertThat(clientHeaders.get(TestServiceImpl.EXTRA_HEADER_NAME)).isEqualTo("token");
}
 

@Test
void credentialsUnaryCall_https() {
    final TestServiceBlockingStub stub =
            // Explicitly construct URL to better test authority.
            Clients.builder("gproto+https://127.0.0.1:" + server.httpsPort())
                   .decorator(LoggingClient.builder().newDecorator())
                   .factory(ClientFactory.insecure())
                   .build(TestServiceBlockingStub.class)
                   .withCallCredentials(
                           new CallCredentials() {
                               @Override
                               public void applyRequestMetadata(RequestInfo requestInfo,
                                                                Executor appExecutor,
                                                                MetadataApplier applier) {
                                   assertThat(requestInfo.getAuthority())
                                           .isEqualTo("127.0.0.1:" + server.httpsPort());
                                   assertThat(requestInfo.getSecurityLevel())
                                           .isEqualTo(SecurityLevel.PRIVACY_AND_INTEGRITY);
                                   applier.apply(new Metadata());
                               }

                               @Override
                               public void thisUsesUnstableApi() {
                               }
                           });

    assertThat(stub.emptyCall(EMPTY)).isNotNull();
}
 

@Test
void credentialsUnaryCall_fail() {
    final TestServiceBlockingStub stub =
            // Explicitly construct URL to better test authority.
            Clients.builder("gproto+https://127.0.0.1:" + server.httpsPort())
                   .decorator(LoggingClient.builder().newDecorator())
                   .factory(ClientFactory.insecure())
                   .build(TestServiceBlockingStub.class)
                   .withCallCredentials(
                           new CallCredentials() {
                               @Override
                               public void applyRequestMetadata(RequestInfo requestInfo,
                                                                Executor appExecutor,
                                                                MetadataApplier applier) {
                                   applier.fail(Status.FAILED_PRECONDITION);
                               }

                               @Override
                               public void thisUsesUnstableApi() {
                               }
                           });

    assertThatThrownBy(() -> stub.emptyCall(EMPTY))
            .isInstanceOfSatisfying(StatusRuntimeException.class,
                                    t -> assertThat(t.getStatus().getCode())
                                            .isEqualTo(Code.FAILED_PRECONDITION));
}