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

@Override
public boolean matches(ConditionContext context, AnnotatedTypeMetadata metadata) {
    final ConfigurableListableBeanFactory beanFactory = context.getBeanFactory();
    if (beanFactory == null) {
        return true;
    }

    final String[] beanNames =
            BeanFactoryUtils.beanNamesForTypeIncludingAncestors(beanFactory, ClientFactory.class);

    for (String beanName : beanNames) {
        if (hasQualifier(beanFactory, beanName)) {
            return false;
        }
    }

    return true;
}
 

/**
 * When there are no `ClientFactory`s with `ForCentralDogma` qualifier,
 * the default `ClientFactory` must be used.
 */
@Test
void centralDogmaClient() throws Exception {
    assertThat(client).isNotNull();

    if (SpringBootVersion.getVersion().startsWith("1.")) {
        // JUnit 5 extension for Spring Boot 1.x has a bug which pulls in a bean from other tests,
        // so we can't test this properly.
        final ClientFactory expectedClientFactory =
                new CentralDogmaClientAutoConfigurationWithClientFactoryTest.TestConfiguration()
                        .dogmaClientFactory();
        assertThat(clientFactory).isSameAs(expectedClientFactory);
    } else {
        assertThat(clientFactory).isSameAs(ClientFactory.ofDefault());
    }
}
 

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

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

/**
 * Makes sure the specified certificate is selected.
 */
@Test
void test() throws Exception {
    final AtomicReference<String> actualKeyName = new AtomicReference<>();

    // Create a new ClientFactory with a TrustManager that records the received certificate.
    try (ClientFactory clientFactory =
                 ClientFactory.builder()
                              .tlsCustomizer(b -> {
                                  b.trustManager(new TrustManagerFactoryImpl(actualKeyName));
                              })
                              .build()) {

        // Send a request to make the TrustManager record the certificate.
        final WebClient client = WebClient.builder("h2://127.0.0.1:" + port)
                                          .factory(clientFactory)
                                          .build();
        client.get("/").aggregate().join();

        assertThat(actualKeyName).hasValue(expectedKeyName);
    }
}
 

/**
 * A user can configure an {@link HttpClient} by providing an {@link ArmeriaClientConfigurator} bean.
 */
@Bean
public ArmeriaClientConfigurator armeriaClientConfigurator(ClientFactory clientFactory) {
    // Customize the client using the given WebClientBuilder. For example:
    return builder -> {
        // Use a circuit breaker for each remote host.
        final CircuitBreakerRule rule = CircuitBreakerRule.builder()
                                                          .onServerErrorStatus()
                                                          .onException()
                                                          .thenFailure();
        builder.decorator(CircuitBreakerClient.builder(rule)
                                              .newDecorator());

        // Set a custom client factory.
        builder.factory(clientFactory);
    };
}
 

@Test
public void https() throws Exception {
    final WebClient client = WebClient.builder(server().uri(SessionProtocol.HTTPS))
                                      .factory(ClientFactory.insecure())
                                      .build();
    final AggregatedHttpResponse response = client.get("/jsp/index.jsp").aggregate().get();
    final String actualContent = CR_OR_LF.matcher(response.contentUtf8())
                                         .replaceAll("");
    assertThat(actualContent).isEqualTo(
            "<html><body>" +
            "<p>Hello, Armerian World!</p>" +
            "<p>Have you heard about the class 'org.slf4j.Logger'?</p>" +
            "<p>Context path: </p>" + // ROOT context path
            "<p>Request URI: /index.jsp</p>" +
            "<p>Scheme: https</p>" +
            "</body></html>");
}
 

/**
 * Schedules next retry.
 */
protected static void scheduleNextRetry(ClientRequestContext ctx,
                                        Consumer<? super Throwable> actionOnException,
                                        Runnable retryTask, long nextDelayMillis) {
    try {
        if (nextDelayMillis == 0) {
            ctx.contextAwareEventLoop().execute(retryTask);
        } else {
            @SuppressWarnings("unchecked")
            final ScheduledFuture<Void> scheduledFuture = (ScheduledFuture<Void>) ctx
                    .contextAwareEventLoop().schedule(retryTask, nextDelayMillis, TimeUnit.MILLISECONDS);
            scheduledFuture.addListener(future -> {
                if (future.isCancelled()) {
                    // future is cancelled when the client factory is closed.
                    actionOnException.accept(new IllegalStateException(
                            ClientFactory.class.getSimpleName() + " has been closed."));
                } else if (future.cause() != null) {
                    // Other unexpected exceptions.
                    actionOnException.accept(future.cause());
                }
            });
        }
    } catch (Throwable t) {
        actionOnException.accept(t);
    }
}
 

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

@ParameterizedTest
@CsvSource({
        "foo.com, /foo/1, " + ACCESS_TOKEN + ", 200",
        "foo.com, /foo/1, , 401",
        "bar.com, /bar/1, , 200"
})
void virtualHost(String host, String path, @Nullable String authorization, int status) {
    final ClientFactory factory =
            ClientFactory.builder()
                         .addressResolverGroupFactory(eventLoop -> MockAddressResolverGroup.localhost())
                         .build();
    final WebClient client = WebClient.builder("http://" + host + ':' + virtualHostServer.httpPort())
                                      .factory(factory)
                                      .build();
    final RequestHeaders headers;
    if (authorization != null) {
        headers = RequestHeaders.of(HttpMethod.GET, path, HttpHeaderNames.AUTHORIZATION, authorization);
    } else {
        headers = RequestHeaders.of(HttpMethod.GET, path);
    }
    final AggregatedHttpResponse res = client.execute(headers).aggregate().join();
    assertThat(res.status().code()).isEqualTo(status);
}
 

/**
 * Returns {@link ClientFactory#ofDefault()} which is used as the default {@link ClientFactory} of a
 * {@link CentralDogma} client.
 */
@Bean
@Conditional(MissingCentralDogmaClientFactory.class)
@ForCentralDogma
public ClientFactory dogmaClientFactory() {
    return ClientFactory.ofDefault();
}
 

@Test
void centralDogmaClient() throws Exception {
    assertThat(client).isNotNull();
    assertThat(clientFactoryForCentralDogma).isNotSameAs(ClientFactory.ofDefault());
    assertThat(clientFactoryForCentralDogma).isSameAs(TestConfiguration.dogmaClientFactory);
    assertThat(clientFactoryForTest).isSameAs(TestConfiguration.otherClientFactory);
    assertThat(testBean).isNotNull();
}
 

private static void test(Function<CentralDogma, CompletableFuture<?>> watchAction) throws Exception {
    final ClientFactory clientFactory = ClientFactory.builder().build();
    final CentralDogma client = new ArmeriaCentralDogmaBuilder()
            .clientFactory(clientFactory)
            .host("127.0.0.1", dogma.serverAddress().getPort())
            .build();

    final CompletableFuture<?> future = watchAction.apply(client);

    // Wait until the server receives the watch request.
    await().untilAsserted(() -> {
        assertThat(MoreMeters.measureAll(dogma.dogma().meterRegistry().get()))
                .containsEntry("watches.active#value", 1.0);
    });

    // Close the `ClientFactory` to trigger disconnection.
    clientFactory.close();

    // The watch request should finish without an exception.
    assertThat(future.join()).isNull();

    // Wait until the server detects the watch cancellation.
    await().untilAsserted(() -> {
        assertThat(MoreMeters.measureAll(dogma.dogma().meterRegistry().get()))
                .containsEntry("watches.active#value", 0.0);
    });
}
 

@Test
public void mockGrpcServerServesOverSSL() { // sanity checks the mock server
  TraceServiceGrpc.TraceServiceBlockingStub sslTraceService =
      Clients.builder("gproto+https://" + mockServer.grpcURI() + "/")
          .factory(ClientFactory.builder()
              .tlsCustomizer(tls -> tls.trustManager(InsecureTrustManagerFactory.INSTANCE))
              .build())
          .build(TraceServiceGrpc.TraceServiceBlockingStub.class);

  sslTraceService.batchWriteSpans(BatchWriteSpansRequest.getDefaultInstance());
}
 

@Override
protected SimpleBenchmarkClient newClient() {
    final WebClient webClient = WebClient.builder(baseUrl())
                                         .factory(ClientFactory.insecure())
                                         .build();
    return ArmeriaRetrofit.builder(webClient)
                          .addConverterFactory(JacksonConverterFactory.create())
                          .build()
                          .create(SimpleBenchmarkClient.class);
}
 

@Test
void testTimingAnnotations() {
    // Use separate client factory to make sure connection is created.
    final ClientFactory clientFactory = ClientFactory.builder().build();
    final WebClient client = WebClient.builder(server.httpUri())
                                      .factory(clientFactory)
                                      .decorator(BraveClient.newDecorator(newTracing("timed-client")))
                                      .build();
    assertThat(client.get("/http").aggregate().join().status()).isEqualTo(HttpStatus.OK);
    final MutableSpan[] initialConnectSpans = spanHandler.take(1);
    assertThat(initialConnectSpans[0].annotations())
            .extracting(Map.Entry::getValue).containsExactlyInAnyOrder(
            "connection-acquire.start",
            "socket-connect.start",
            "socket-connect.end",
            "connection-acquire.end",
            "ws",
            "wr");

    // Make another request which will reuse the connection so no connection timing.
    assertThat(client.get("/http").aggregate().join().status()).isEqualTo(HttpStatus.OK);

    final MutableSpan[] secondConnectSpans = spanHandler.take(1);
    assertThat(secondConnectSpans[0].annotations())
            .extracting(Map.Entry::getValue).containsExactlyInAnyOrder(
            "ws",
            "wr");
}
 

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

private static void clientSocketClosedBeforeHalfClose(String protocol) throws Exception {
    final ClientFactory factory = ClientFactory.builder().build();
    final UnitTestServiceStub stub =
            Clients.builder("gproto+" + protocol + "://127.0.0.1:" + server.httpPort() + '/')
                   .factory(factory)
                   .build(UnitTestServiceStub.class);
    final AtomicReference<SimpleResponse> response = new AtomicReference<>();
    final StreamObserver<SimpleRequest> stream = stub.streamClientCancels(
            new StreamObserver<SimpleResponse>() {
                @Override
                public void onNext(SimpleResponse value) {
                    response.set(value);
                }

                @Override
                public void onError(Throwable t) {
                }

                @Override
                public void onCompleted() {
                }
            });
    stream.onNext(SimpleRequest.getDefaultInstance());
    await().untilAsserted(() -> assertThat(response).hasValue(SimpleResponse.getDefaultInstance()));
    factory.close();
    await().untilAsserted(() -> assertThat(COMPLETED).hasValue(true));

    checkRequestLog((rpcReq, rpcRes, grpcStatus) -> {
        assertThat(rpcReq.method()).isEqualTo("armeria.grpc.testing.UnitTestService/StreamClientCancels");
        assertThat(rpcReq.params()).containsExactly(SimpleRequest.getDefaultInstance());
        assertThat(grpcStatus).isNotNull();
        assertThat(grpcStatus.getCode()).isEqualTo(protocol.startsWith("h2") ? Code.CANCELLED
                                                                             : Code.UNKNOWN);
    });
}
 

private static RequestLog clientSocketClosedAfterHalfCloseBeforeCloseCancels(SessionProtocol protocol)
        throws Exception {

    final ClientFactory factory = ClientFactory.builder().build();
    final UnitTestServiceStub stub =
            Clients.builder(server.uri(protocol, GrpcSerializationFormats.PROTO))
                   .factory(factory)
                   .build(UnitTestServiceStub.class);
    final AtomicReference<SimpleResponse> response = new AtomicReference<>();
    stub.streamClientCancelsBeforeResponseClosedCancels(
            SimpleRequest.getDefaultInstance(),
            new StreamObserver<SimpleResponse>() {
                @Override
                public void onNext(SimpleResponse value) {
                    response.set(value);
                }

                @Override
                public void onError(Throwable t) {
                }

                @Override
                public void onCompleted() {
                }
            });
    await().untilAsserted(() -> assertThat(response).hasValue(SimpleResponse.getDefaultInstance()));
    factory.close();
    CLIENT_CLOSED.set(true);
    await().untilAsserted(() -> assertThat(COMPLETED).hasValue(true));

    final RequestLog log = requestLogQueue.take();
    assertThat(log.isComplete()).isTrue();
    assertThat(log.requestContent()).isNotNull();
    assertThat(log.responseContent()).isNull();
    final RpcRequest rpcReq = (RpcRequest) log.requestContent();
    assertThat(rpcReq.method()).isEqualTo(
            "armeria.grpc.testing.UnitTestService/StreamClientCancelsBeforeResponseClosedCancels");
    assertThat(rpcReq.params()).containsExactly(SimpleRequest.getDefaultInstance());
    return log;
}
 

@BeforeAll
static void init() throws Exception {
    final ConnectionPoolListener connectionPoolListener =
            ENABLE_CONNECTION_POOL_LOGGING ? new ConnectionPoolLoggingListener()
                                           : ConnectionPoolListener.noop();

    clientFactoryWithUseHttp2Preface = ClientFactory.builder()
                                                    .tlsNoVerify()
                                                    .connectionPoolListener(connectionPoolListener)
                                                    .useHttp2Preface(true)
                                                    .build();

    clientFactoryWithoutUseHttp2Preface = ClientFactory.builder()
                                                       .tlsNoVerify()
                                                       .connectionPoolListener(connectionPoolListener)
                                                       .useHttp2Preface(false)
                                                       .build();

    final ClientDecorationBuilder decoBuilder = ClientDecoration.builder();
    decoBuilder.addRpc((delegate, ctx, req) -> {
        if (recordMessageLogs) {
            ctx.log().whenComplete().thenAccept(requestLogs::add);
        }
        return delegate.execute(ctx, req);
    });

    if (ENABLE_LOGGING_DECORATORS) {
        decoBuilder.addRpc(LoggingRpcClient.newDecorator());
    }

    clientOptions = ClientOptions.of(ClientOption.DECORATION.newValue(decoBuilder.build()));
}