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

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

/**
 * If no CORS was configured and there's no binding for OPTIONS method, the server's fallback service will
 * be matched and the service with partial binding must not be invoked.
 */
@Test
public void testNoCorsWithPartialBinding() {
    final WebClient client = client();
    AggregatedHttpResponse res;

    // A simple OPTIONS request, which should fall back.
    res = client.options("/cors12/get").aggregate().join();
    assertThat(res.status()).isEqualTo(HttpStatus.METHOD_NOT_ALLOWED);

    // A CORS preflight request, which should fall back as well.
    res = preflightRequest(client, "/cors12/get", "http://example.com", "GET");
    assertThat(res.status()).isEqualTo(HttpStatus.FORBIDDEN);
    // .. but will not contain CORS headers.
    assertThat(res.headers().get(HttpHeaderNames.ACCESS_CONTROL_ALLOW_METHODS)).isNull();
}
 

@ParameterizedTest
@ArgumentsSource(ClientAndProtocolProvider.class)
void testTimeoutAfterInformationals(WebClient client) throws Exception {
    serverRequestTimeoutMillis = 1000L;
    final AggregatedHttpResponse res = client.get("/informed_delay/2000").aggregate().get();
    assertThat(res.informationals()).isNotEmpty();
    res.informationals().forEach(h -> {
        assertThat(h.status()).isEqualTo(HttpStatus.PROCESSING);
        assertThat(h.names()).contains(HttpHeaderNames.STATUS);
    });

    assertThat(res.status()).isEqualTo(HttpStatus.SERVICE_UNAVAILABLE);
    assertThat(res.contentType()).isEqualTo(MediaType.PLAIN_TEXT_UTF_8);
    assertThat(res.contentUtf8()).isEqualTo("503 Service Unavailable");
    assertThat(requestLogs.take().responseHeaders().status()).isEqualTo(HttpStatus.SERVICE_UNAVAILABLE);
}
 

@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 getJsonDiff() {
    final WebClient client = dogma.httpClient();
    editFooFile(client);
    final AggregatedHttpResponse aRes = client
            .get("/api/v1/projects/myPro/repos/myRepo/compare?path=/foo0.json&jsonpath=$.a&from=3&to=4")
            .aggregate().join();

    final String expectedJson =
            '{' +
            "   \"path\": \"/foo0.json\"," +
            "   \"type\": \"APPLY_JSON_PATCH\"," +
            "   \"content\": [{" +
            "       \"op\": \"safeReplace\"," +
            "       \"path\": \"\"," +
            "       \"oldValue\": \"bar0\"," +
            "       \"value\": \"baz0\"" +
            "   }]" +
            '}';
    assertThatJson(aRes.contentUtf8()).isEqualTo(expectedJson);
}
 

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

private static void createProject(CentralDogmaExtension dogma) {
    final WebClient client = dogma.httpClient();
    // the default project used for unit tests
    RequestHeaders headers = RequestHeaders.of(HttpMethod.POST, "/api/v1/projects",
                                               HttpHeaderNames.CONTENT_TYPE, MediaType.JSON);
    String body = "{\"name\": \"myPro\"}";
    client.execute(headers, body).aggregate().join();

    // the default repository used for unit tests
    headers = RequestHeaders.of(HttpMethod.POST, "/api/v1/projects/myPro/repos",
                                HttpHeaderNames.CONTENT_TYPE, MediaType.JSON);
    body = "{\"name\": \"myRepo\"}";
    client.execute(headers, body).aggregate().join();
    // default files used for unit tests
    addFooFile(client);
}
 

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

@Test
public void metaAnnotations() {
    final AggregatedHttpResponse msg =
            WebClient.of(rule.httpUri())
                     .execute(RequestHeaders.of(HttpMethod.POST, "/hello",
                                                HttpHeaderNames.CONTENT_TYPE,
                                                MediaType.PLAIN_TEXT_UTF_8,
                                                HttpHeaderNames.ACCEPT, "text/*"),
                              HttpData.ofUtf8("Armeria"))
                     .aggregate().join();
    assertThat(msg.status()).isEqualTo(HttpStatus.CREATED);
    assertThat(msg.contentType()).isEqualTo(MediaType.PLAIN_TEXT_UTF_8);
    assertThat(msg.headers().get(HttpHeaderNames.of("x-foo"))).isEqualTo("foo");
    assertThat(msg.headers().get(HttpHeaderNames.of("x-bar"))).isEqualTo("bar");
    assertThat(msg.contentUtf8())
            .isEqualTo("Hello, Armeria (decorated-1) (decorated-2) (decorated-3)!");
    assertThat(msg.trailers().get(HttpHeaderNames.of("x-baz"))).isEqualTo("baz");
    assertThat(msg.trailers().get(HttpHeaderNames.of("x-qux"))).isEqualTo("qux");
}
 

@Test
public void composeWithOtherHook() throws Exception {
    final AtomicInteger calledFlag = new AtomicInteger();
    RxJavaPlugins.setOnSingleAssembly(single -> {
        calledFlag.incrementAndGet();
        return single;
    });
    final WebClient client = WebClient.of(rule.httpUri());
    client.execute(RequestHeaders.of(HttpMethod.GET, "/single")).aggregate().get();
    assertThat(calledFlag.get()).isEqualTo(3);

    try {
        RequestContextAssembly.enable();
        client.execute(RequestHeaders.of(HttpMethod.GET, "/single")).aggregate().get();
        assertThat(calledFlag.get()).isEqualTo(6);
    } finally {
        RequestContextAssembly.disable();
    }
    client.execute(RequestHeaders.of(HttpMethod.GET, "/single")).aggregate().get();
    assertThat(calledFlag.get()).isEqualTo(9);

    RxJavaPlugins.setOnSingleAssembly(null);
    client.execute(RequestHeaders.of(HttpMethod.GET, "/single")).aggregate().get();
    assertThat(calledFlag.get()).isEqualTo(9);
}
 

@ParameterizedTest
@MethodSource("arguments")
void test(String secret, String projectName, ProjectRole role, String repoName,
          Set<Permission> permission, HttpStatus expectedFailureStatus) {
    final WebClient client = WebClient.builder(server.httpUri())
                                      .addHttpHeader(HttpHeaderNames.AUTHORIZATION, "Bearer " + secret)
                                      .build();

    AggregatedHttpResponse response;

    response = client.get("/projects/" + projectName).aggregate().join();
    assertThat(response.status())
            .isEqualTo(role == ProjectRole.OWNER || role == ProjectRole.MEMBER ? HttpStatus.OK
                                                                               : expectedFailureStatus);

    response = client.post("/projects/" + projectName + "/repos/" + repoName, HttpData.empty())
                     .aggregate().join();
    assertThat(response.status()).isEqualTo(permission.contains(Permission.WRITE) ? HttpStatus.OK
                                                                                  : expectedFailureStatus);

    response = client.get("/projects/" + projectName + "/repos/" + repoName)
                     .aggregate().join();
    assertThat(response.status()).isEqualTo(permission.isEmpty() ? expectedFailureStatus
                                                                 : HttpStatus.OK);
}
 

@ParameterizedTest
@ArgumentsSource(SchemesProvider.class)
void shouldGetHelloFromRouter(String scheme) throws Exception {
    final WebClient client = WebClient.builder(scheme + "://example.com:" + port)
                                      .factory(clientFactory)
                                      .build();

    final AggregatedHttpResponse res = client.get("/route").aggregate().join();
    assertThat(res.contentUtf8()).isEqualTo("route");

    final AggregatedHttpResponse res2 =
            client.execute(RequestHeaders.of(HttpMethod.POST, "/route2",
                                             HttpHeaderNames.CONTENT_TYPE, JSON_UTF_8),
                           HttpData.wrap("{\"a\":1}".getBytes())).aggregate().join();
    assertThatJson(res2.contentUtf8()).isArray()
                                      .ofLength(1)
                                      .thatContains("route");
}
 

@Test
void routeService() throws InterruptedException {
    final WebClient client = WebClient.of(server.httpUri());
    AggregatedHttpResponse res = client.get("/greet/armeria").aggregate().join();
    propertyCheckLatch.await();
    assertThat(res.status()).isSameAs(HttpStatus.OK);
    assertThat(res.contentUtf8()).isEqualTo("armeria");

    res = client.post("/greet", "armeria").aggregate().join();
    assertThat(res.status()).isSameAs(HttpStatus.OK);
    assertThat(res.contentUtf8()).isEqualTo("armeria");

    res = client.put("/greet/armeria", "armeria").aggregate().join();
    assertThat(res.status()).isSameAs(HttpStatus.METHOD_NOT_ALLOWED);

    res = client.put("/greet", "armeria").aggregate().join();
    assertThat(res.status()).isSameAs(HttpStatus.METHOD_NOT_ALLOWED);
}
 

@Test
public void testRoute_order() {
    final WebClient client = client();
    AggregatedHttpResponse res;

    res = preflightRequest(client, "/cors9/movies", "http://example.com", "GET");
    assertThat(res.status()).isEqualTo(HttpStatus.OK);
    assertThat(res.headers().get(HttpHeaderNames.ACCESS_CONTROL_ALLOW_METHODS)).isEqualTo("GET");

    res = preflightRequest(client, "/cors9/movies/InfinityWar", "http://example.com", "POST");
    assertThat(res.status()).isEqualTo(HttpStatus.OK);
    assertThat(res.headers().get(HttpHeaderNames.ACCESS_CONTROL_ALLOW_METHODS)).isEqualTo("GET");

    res = preflightRequest(client, "/cors9/movies/InfinityWar/actors", "http://example.com", "GET");
    assertThat(res.status()).isEqualTo(HttpStatus.OK);
    assertThat(res.headers().get(HttpHeaderNames.ACCESS_CONTROL_ALLOW_METHODS)).isEqualTo("GET");
}
 

@ParameterizedTest
@CsvSource({
        "/api/users/1, " + ACCESS_TOKEN + ", 200, ",
        "/api/users/2, " + ACCESS_TOKEN + ", 200, ",
        "/api/users/1, , 401, ",
        "/api/users/2, , 401, ",
        "/api/admin/1, " + ACCESS_TOKEN + ", 200, ",
        "/api/admin/1, , 401, ",
        "/assets/index.html, , 200, ",
        "/assets/resources/index.html, , 200, public",
        "/assets/resources/private/profile.jpg, , 200, private",
})
void secured(String path, @Nullable String authorization, int status, String cacheControl) {
    final WebClient client = WebClient.of(authServer.httpUri());
    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);
    assertThat(res.headers().get(HttpHeaderNames.CACHE_CONTROL)).isEqualTo(cacheControl);
}
 

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

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

@ParameterizedTest
@ArgumentsSource(ClientAndProtocolProvider.class)
void testTimeoutAfterPartialContent(WebClient client) throws Exception {
    serverRequestTimeoutMillis = 1000L;
    final CompletableFuture<AggregatedHttpResponse> f = client.get("/content_delay/2000").aggregate();

    // Because the service has written out the content partially, there's no way for the service
    // to reply with '503 Service Unavailable', so it will just close the stream.

    final Class<? extends Throwable> expectedCauseType =
            client.scheme().sessionProtocol().isMultiplex() ?
            ClosedStreamException.class : ClosedSessionException.class;

    assertThatThrownBy(f::get).isInstanceOf(ExecutionException.class)
                              .hasCauseInstanceOf(expectedCauseType);
}
 

@Test
void getFile() {
    final WebClient client = dogma.httpClient();
    addFooJson(client);
    final AggregatedHttpResponse aRes = client.get(CONTENTS_PREFIX + "/foo.json").aggregate().join();

    final String expectedJson =
            '{' +
            "   \"revision\": 2," +
            "   \"path\": \"/foo.json\"," +
            "   \"type\": \"JSON\"," +
            "   \"content\" : {\"a\":\"bar\"}," +
            "   \"url\": \"/api/v1/projects/myPro/repos/myRepo/contents/foo.json\"" +
            '}';
    final String actualJson = aRes.contentUtf8();
    assertThatJson(actualJson).isEqualTo(expectedJson);
}
 

@Test
void delay() {
    server.enqueue(HttpResponse.delayed(AggregatedHttpResponse.of(HttpStatus.OK).toHttpResponse(),
                                        Duration.ofSeconds(1)));
    server.enqueue(HttpResponse.delayed(AggregatedHttpResponse.of(HttpStatus.OK), Duration.ofSeconds(1)));

    final WebClient client =
            WebClient.builder(server.httpUri())
                     .option(ClientOption.RESPONSE_TIMEOUT_MILLIS.newValue(50L))
                     .build();

    assertThatThrownBy(() -> client.get("/").aggregate().join())
            .hasCauseInstanceOf(ResponseTimeoutException.class);
    assertThatThrownBy(() -> client.get("/").aggregate().join())
            .hasCauseInstanceOf(ResponseTimeoutException.class);
}
 

@Test
void editFileWithJsonPatch() throws IOException {
    final WebClient client = dogma.httpClient();
    addFooJson(client);
    final AggregatedHttpResponse res1 = editFooJson(client);
    final String expectedJson =
            '{' +
            "   \"revision\": 3," +
            "   \"pushedAt\": \"${json-unit.ignore}\"" +
            '}';
    final String actualJson = res1.contentUtf8();
    assertThatJson(actualJson).isEqualTo(expectedJson);

    final AggregatedHttpResponse res2 = client.get(CONTENTS_PREFIX + "/foo.json").aggregate().join();
    assertThat(Jackson.readTree(res2.contentUtf8()).get("content").get("a").textValue())
            .isEqualToIgnoringCase("baz");
}
 

EurekaEndpointGroup(EndpointSelectionStrategy selectionStrategy,
                    WebClient webClient, long registryFetchIntervalSeconds, @Nullable String appName,
                    @Nullable String instanceId, @Nullable String vipAddress,
                    @Nullable String secureVipAddress, @Nullable List<String> regions) {
    super(selectionStrategy);
    this.webClient = PooledWebClient.of(webClient);
    this.registryFetchIntervalSeconds = registryFetchIntervalSeconds;

    final RequestHeadersBuilder headersBuilder = RequestHeaders.builder();
    headersBuilder.method(HttpMethod.GET);
    headersBuilder.add(HttpHeaderNames.ACCEPT, MediaTypeNames.JSON_UTF_8);
    responseConverter = responseConverter(headersBuilder, appName, instanceId,
                                          vipAddress, secureVipAddress, regions);
    requestHeaders = headersBuilder.build();

    webClient.options().factory().whenClosed().thenRun(this::closeAsync);
    fetchRegistry();
}
 

@Test
void watchRepository() {
    final WebClient client = dogma.httpClient();
    addFooJson(client);
    final RequestHeaders headers = RequestHeaders.of(HttpMethod.GET, CONTENTS_PREFIX,
                                                     HttpHeaderNames.IF_NONE_MATCH, "-1");
    final CompletableFuture<AggregatedHttpResponse> future = client.execute(headers).aggregate();

    assertThatThrownBy(() -> future.get(500, TimeUnit.MILLISECONDS))
            .isExactlyInstanceOf(TimeoutException.class);

    editFooJson(client);

    await().atMost(3, TimeUnit.SECONDS).untilAsserted(future::isDone);
    final AggregatedHttpResponse res = future.join();

    final String expectedJson =
            '{' +
            "   \"revision\" : 3" +
            '}';
    final String actualJson = res.contentUtf8();
    assertThatJson(actualJson).isEqualTo(expectedJson);
}
 

@Produces
static ListenableFuture<List<AggregatedHttpResponse>> fetchFromBackend(
        AggregatedHttpRequest request, List<Long> dbNums, WebClient backendClient,
        ServiceRequestContext context) {
    // The context is mounted in a thread-local, meaning it is available to all logic such as tracing.
    checkState(ServiceRequestContext.current() == context);
    checkState(context.eventLoop().inEventLoop());

    final Stream.Builder<Long> nums = Stream.builder();
    for (String token : Iterables.concat(
            NUM_SPLITTER.split(request.path().substring(1)),
            NUM_SPLITTER.split(request.contentUtf8()))) {
        nums.add(Long.parseLong(token));
    }
    dbNums.forEach(nums::add);

    return Futures.allAsList(
            nums.build()
                .map(num -> toListenableFuture(backendClient.get("/square/" + num).aggregate()))
            .collect(toImmutableList()));
}
 

@Test
void retryWithContentOnResponseTimeout() {
    final Backoff backoff = Backoff.fixed(100);
    final RetryRuleWithContent<HttpResponse> strategy =
            RetryRuleWithContent.<HttpResponse>onResponse((unused, response) -> {
                return response.aggregate().thenApply(unused0 -> false);
            }).orElse(RetryRuleWithContent.onResponse((unused, response) -> {
                return response.aggregate().thenApply(unused0 -> false);
            })).orElse(RetryRuleWithContent.<HttpResponse>onResponse((unused, response) -> {
                return response.aggregate().thenApply(unused0 -> false);
            }).orElse(RetryRule.builder()
                               .onException(ResponseTimeoutException.class)
                               .thenBackoff(backoff)));
    final WebClient client = client(strategy, 0, 500, 100);
    final AggregatedHttpResponse res = client.get("/1sleep-then-success").aggregate().join();
    assertThat(res.contentUtf8()).isEqualTo("Succeeded after retry");
}
 

@Test
void retryWithContentOnUnprocessedException() {
    final Backoff backoff = Backoff.fixed(2000);
    final RetryRuleWithContent<HttpResponse> strategy =
            RetryRuleWithContent.<HttpResponse>onResponse((unused, response) -> {
                return response.aggregate().thenApply(unused0 -> false);
            }).orElse(RetryRuleWithContent.onResponse((unused, response) -> {
                return response.aggregate().thenApply(unused0 -> false);
            })).orElse(RetryRuleWithContent.<HttpResponse>onResponse((unused, response) -> {
                return response.aggregate().thenApply(unused0 -> false);
            }).orElse(RetryRule.builder()
                               .onException(UnprocessedRequestException.class)
                               .thenBackoff(backoff)));
    final Function<? super HttpClient, RetryingClient> retryingDecorator =
            RetryingClient.builder(strategy)
                          .maxTotalAttempts(5)
                          .newDecorator();

    final WebClient client = WebClient.builder("http://127.0.0.1:1")
                                      .factory(ClientFactory.builder()
                                                            .options(clientFactory.options())
                                                            .connectTimeoutMillis(Long.MAX_VALUE)
                                                            .build())
                                      .responseTimeoutMillis(0)
                                      .decorator(LoggingClient.newDecorator())
                                      .decorator(retryingDecorator)
                                      .build();
    final Stopwatch stopwatch = Stopwatch.createStarted();
    assertThatThrownBy(() -> client.get("/unprocessed-exception").aggregate().join())
            .isInstanceOf(CompletionException.class)
            .hasCauseInstanceOf(UnprocessedRequestException.class);
    assertThat(stopwatch.elapsed()).isBetween(Duration.ofSeconds(7), Duration.ofSeconds(20));
}
 

@Produces
@FetchedPostPage
static ListenableFuture<List<@Nullable AggregatedHttpResponse>> fetchPosts(
    @HashtagPage List<@Nullable AggregatedHttpResponse> hashtagPages,
    SharedDataExtractor sharedDataExtractor,
    WebClient instagramClient,
    ServiceRequestContext ctx) {
  return Futures.successfulAsList(
      hashtagPages.stream()
          .filter(Objects::nonNull)
          .map(page -> sharedDataExtractor.extractSharedData(page, TagPage.class))
          .flatMap(
              page ->
                  page
                      .getEntryData()
                      .getTagPage()
                      .get(0)
                      .getGraphql()
                      .getHashtag()
                      .getPosts()
                      .getEdges()
                      .stream())
          .map(
              post ->
                  toListenableFuture(
                      instagramClient
                          .get("/p/" + post.getNode().getShortcode() + '/')
                          .aggregateWithPooledObjects(ctx.eventLoop(), ctx.alloc())))
          .collect(toImmutableList()));
}
 

@Test
void methodNotAllowed() {
    final WebClient client = WebClient.of(server.httpUri());
    final AggregatedHttpResponse res = client.delete("/hello").aggregate().join();
    assertThat(res.status()).isSameAs(HttpStatus.METHOD_NOT_ALLOWED);
    await().untilAsserted(() -> {
        assertThat(logHolder.get().requestHeaders().path()).isEqualTo("/hello");
    });
    assertThat(logHolder.get().requestCause()).isNull();
}
 

@ParameterizedTest
@ArgumentsSource(ClientAndProtocolProvider.class)
void testTimeout_customHandler_deferred(WebClient client) throws Exception {
    serverRequestTimeoutMillis = 100L;
    final AggregatedHttpResponse res = client.get("/delay-custom-deferred/2000").aggregate().get();
    assertThat(res.status()).isEqualTo(HttpStatus.OK);
    assertThat(res.contentType()).isEqualTo(MediaType.PLAIN_TEXT_UTF_8);
    assertThat(res.contentUtf8()).isEqualTo("timed out");
    assertThat(requestLogs.take().responseHeaders().status()).isEqualTo(HttpStatus.OK);
}
 

@ParameterizedTest
@ArgumentsSource(SchemesProvider.class)
void shouldGetHelloFromRestController(String scheme) throws Exception {
    final WebClient client = WebClient.builder(scheme + "://example.com:" + port)
                                      .factory(clientFactory)
                                      .build();
    final AggregatedHttpResponse response = client.get("/hello").aggregate().join();
    assertThat(response.contentUtf8()).isEqualTo("hello");
}