import org.springframework.web.reactive.function.client.ClientRequest; //导入依赖的package包/类
/**
 * Create a new {@link ReactiveVaultTemplate} with a {@link VaultEndpointProvider},
 * {@link ClientHttpConnector} and {@link VaultTokenSupplier}.
 *
 * @param endpointProvider must not be {@literal null}.
 * @param connector must not be {@literal null}.
 * @param vaultTokenSupplier must not be {@literal null}.
 */
public ReactiveVaultTemplate(VaultEndpointProvider endpointProvider,
		ClientHttpConnector connector, VaultTokenSupplier vaultTokenSupplier) {

	Assert.notNull(endpointProvider, "VaultEndpointProvider must not be null");
	Assert.notNull(connector, "ClientHttpConnector must not be null");
	Assert.notNull(vaultTokenSupplier, "AuthenticationSupplier must not be null");

	ExchangeFilterFunction filter = ofRequestProcessor(request -> vaultTokenSupplier
			.getVaultToken().map(token -> {

				return ClientRequest.from(request).headers(headers -> {
					headers.set(VaultHttpHeaders.VAULT_TOKEN, token.getToken());
				}).build();
			}));

	this.statelessClient = ReactiveVaultClients.createWebClient(endpointProvider,
			connector);
	this.sessionClient = ReactiveVaultClients
			.createWebClient(endpointProvider, connector).mutate().filter(filter)
			.build();
}
 

import org.springframework.web.reactive.function.client.ClientRequest; //导入依赖的package包/类
private Mono<Quotation> requestForQuotationWithHystrixWebClient(String bankURL, Double loanAmount) {

        ClientRequest<Void> request = ClientRequest.GET(bankURL.concat(REQUEST_QUOTATION_PATH), loanAmount)
                .accept(MediaType.APPLICATION_JSON)
                .build();

        return hystrixWebClient.exchangeCommandBuilder()
                .withCommandName(bankURL)
                .withErrorThresholdPercentage(50)
                .withRequestVolumeThreshold(5)
                .withSleepWindowInMilliseconds(5000)
                .withExecutionTimeoutInMilliseconds(1500)
                .exchange(request)
                .then(response -> response.bodyToMono(Quotation.class))
                .otherwiseReturn(OFFER_IN_CASE_OF_ERROR);
    }
 

import org.springframework.web.reactive.function.client.ClientRequest; //导入依赖的package包/类
@Test
public void execute_withHeaders() {
    Request request = new MockRequest("http://example.ca", HttpMethod.GET);
    request.headers().add(HttpHeaders.ACCEPT, MediaType.APPLICATION_JSON_VALUE);
    request.headers().add(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_XML_VALUE);
    
    requestExecutor.execute(request);
    ClientRequest clientRequest = verifyExchange();
    assertThat(clientRequest.url())
            .isEqualTo(URI.create("http://example.ca"));
    assertThat(clientRequest.method())
            .isEqualTo(HttpMethod.GET);
    assertThat(clientRequest.headers().getFirst(HttpHeaders.ACCEPT))
            .isEqualTo(MediaType.APPLICATION_JSON_VALUE);
    assertThat(clientRequest.headers().getFirst(HttpHeaders.CONTENT_TYPE))
            .isEqualTo(MediaType.APPLICATION_XML_VALUE);
}
 

import org.springframework.web.reactive.function.client.ClientRequest; //导入依赖的package包/类
/**
 * Enriches the request with proper headers and publishes
 * the client sent event
 */
private Span createNewSpan(ClientRequest request, Span optionalParent) {
	URI uri = request.url();
	String spanName = getName(uri);
	Span newSpan;
	if (optionalParent == null) {
		newSpan = tracer().createSpan(spanName);
	} else {
		newSpan = tracer().createSpan(spanName, optionalParent);
	}
	addRequestTags(request);
	newSpan.logEvent(Span.CLIENT_SEND);
	if (log.isDebugEnabled()) {
		log.debug("Starting new client span [" + newSpan + "]");
	}
	return newSpan;
}
 

import org.springframework.web.reactive.function.client.ClientRequest; //导入依赖的package包/类
@Override
public Mono<ClientResponse> filter(ClientRequest request, ExchangeFunction next) {
	URI originalUrl = request.url();
	String serviceId = originalUrl.getHost();
	Assert.state(serviceId != null, "Request URI does not contain a valid hostname: " + originalUrl);
	//TODO: reactive lb client

	ServiceInstance instance = this.loadBalancerClient.choose(serviceId);
	URI uri = this.loadBalancerClient.reconstructURI(instance, originalUrl);
	ClientRequest newRequest = ClientRequest.method(request.method(), uri)
			.headers(headers -> headers.addAll(request.headers()))
			.cookies(cookies -> cookies.addAll(request.cookies()))
			.attributes(attributes -> attributes.putAll(request.attributes()))
			.body(request.body())
			.build();
	return next.exchange(newRequest);
}
 

import org.springframework.web.reactive.function.client.ClientRequest; //导入依赖的package包/类
public static ExchangeFilterFunction rewriteEndpointUrl() {
    return withInstance((instance, request, next) -> {
        if (request.url().isAbsolute()) {
            return next.exchange(request);
        }

        UriComponents oldUrl = UriComponentsBuilder.fromUri(request.url()).build();
        if (oldUrl.getPathSegments().isEmpty()) {
            return Mono.error(new InstanceWebClientException("No endpoint specified"));
        }

        String endpointId = oldUrl.getPathSegments().get(0);
        Optional<Endpoint> endpoint = instance.getEndpoints().get(endpointId);

        if (!endpoint.isPresent()) {
            return Mono.error(new InstanceWebClientException("Endpoint '" + endpointId + "' not found"));
        }

        URI newUrl = rewriteUrl(oldUrl, endpoint.get().getUrl());
        ClientRequest newRequest = ClientRequest.from(request)
                                                .attribute(ATTRIBUTE_ENDPOINT, endpoint.get().getId())
                                                .url(newUrl)
                                                .build();
        return next.exchange(newRequest);
    });
}
 

import org.springframework.web.reactive.function.client.ClientRequest; //导入依赖的package包/类
@Override
protected Mono<Quotation> requestForQuotation(String bankURL, Double loanAmount) {
    ClientRequest<Void> request = ClientRequest.GET(bankURL.concat(REQUEST_QUOTATION_PATH), loanAmount)
            .accept(MediaType.APPLICATION_JSON)
            .build();

    return hystrixWebClient.exchangeCommandBuilder()
            .withCommandName(bankURL)
            .withErrorThresholdPercentage(50)
            .withRequestVolumeThreshold(5)
            .withSleepWindowInMilliseconds(5000)
            .withExecutionTimeoutInMilliseconds(Long.valueOf(REQUEST_TIMEOUT.toMillis()).intValue())
            .exchange(request)
            .then(response -> response.bodyToMono(Quotation.class));
}
 

import org.springframework.web.reactive.function.client.ClientRequest; //导入依赖的package包/类
private Mono<Quotation> requestForQuotationWithoutCB(String bankURL, Double loanAmount) {

        ClientRequest<Void> request = ClientRequest.GET(bankURL.concat(REQUEST_QUOTATION_PATH), loanAmount)
                .accept(MediaType.APPLICATION_JSON)
                .build();

        Mono<Quotation> offer = webClient.exchange(request)
                .timeout(REQUEST_TIMEOUT) // Response should come back within REQUEST_TIMEOUT
                .then(response -> response.bodyToMono(Quotation.class))
                .otherwiseReturn(OFFER_IN_CASE_OF_ERROR); // Fallback strategy

        return offer;
    }
 

import org.springframework.web.reactive.function.client.ClientRequest; //导入依赖的package包/类
@Override
protected Observable<Quotation> construct() {

    ClientRequest<Void> request = ClientRequest.GET(getBankURL().concat(REQUEST_QUOTATION_PATH), getLoanAmount())
            .accept(MediaType.APPLICATION_JSON)
            .build();

    Mono<Quotation> offer = webClient
            .exchange(request)
            .timeout(REQUEST_TIMEOUT) // Response should come back within REQUEST_TIMEOUT
            .then(response -> response.bodyToMono(Quotation.class))
            .otherwiseReturn(throwable -> !(throwable instanceof TimeoutException), OFFER_IN_CASE_OF_ERROR); // Fallback strategy

    return toSingle(offer.toFuture()).toObservable();
}
 

import org.springframework.web.reactive.function.client.ClientRequest; //导入依赖的package包/类
private static Mono<ClientResponse> logClientRequest(ClientRequest clientRequest, ExchangeFunction exchangeFunction, Logger logger, LogLevel logLevel) {
    boolean logHeaders = LogLevel.HEADERS.equals(logLevel);
    boolean logBasic = logHeaders || LogLevel.BASIC.equals(logLevel);

    long startNs = logBasic? System.nanoTime() : 0;
    if (logBasic) {
        logger.log(()-> "--> " + clientRequest.method() + ' ' + clientRequest.url());
        logger.log(()-> "Content-Type: " + clientRequest.headers().getContentType());
        if (logHeaders) {
            logger.log("Headers:");
            clientRequest.headers().forEach((name, values) -> {
                if (!HttpHeaders.CONTENT_TYPE.equals(name))
                    logger.log(()-> " * " + name + ": " + values);
            });
        }
        logger.log(() -> "--> END " + clientRequest.method());
    }
    return exchangeFunction
            .exchange(clientRequest)
            .doOnEach(clientResponseSignal -> {
                ClientResponse response;
                if (logBasic && (response = clientResponseSignal.get()) != null) {
                    long tookMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNs);
                    logger.log("<-- " + response.statusCode() + ' ' + clientRequest.method() + ' ' + clientRequest.url() + " (" + tookMs + "ms)");
                    if (logHeaders) {
                        logger.log("Headers:");
                        response.headers().asHttpHeaders().forEach((name, values) -> logger.log(()-> " - " + name + ": " + values));
                    }
                    logger.log(() -> "<-- END HTTP " + clientRequest.method());
                }
            });
}
 

import org.springframework.web.reactive.function.client.ClientRequest; //导入依赖的package包/类
@Test
public void execute() {
    Request request = new MockRequest("http://example.ca", HttpMethod.GET);
    request.headers().add(HttpHeaders.ACCEPT, MediaType.APPLICATION_JSON_VALUE);
    
    requestExecutor.execute(request);
    ClientRequest clientRequest = verifyExchange();
    assertThat(clientRequest.url())
            .isEqualTo(URI.create("http://example.ca"));
    assertThat(clientRequest.method())
            .isEqualTo(HttpMethod.GET);
}
 

import org.springframework.web.reactive.function.client.ClientRequest; //导入依赖的package包/类
@Bean
public CommandLineRunner tweetBot(TwitterOAuth twitterOAuth, ReactiveTweetRepository tweetRepo) {

    return args -> {
        MultiValueMap<String, String> body = new LinkedMultiValueMap<>();

        String tracks = "#cltjug,#FathersDay";
        if (args.length > 0) {
            log.info("Using arguments as tracks");
            tracks = String.join(",", args);
        }

        log.info("Filtering tracks [{}]", tracks);
        body.add("track", tracks);

        WebClient webClient = WebClient.create()
                .filter((currentRequest, next) ->
                        next.exchange(ClientRequest.from(currentRequest)
                                .header(HttpHeaders.AUTHORIZATION, twitterOAuth.oAuth1Header(
                                        currentRequest.url(), currentRequest.method(), body.toSingleValueMap()))
                                .build()));

        Flux<Tweet> tweets = webClient
                .post()
                .uri(TwitterApiEndpoint.TWITTER_STREAM_API_STATUS_FILTER_URL)
                .contentType(MediaType.APPLICATION_FORM_URLENCODED)
                .body(BodyInserters.fromFormData(body))
                .exchange()
                .flatMapMany(clientResponse -> clientResponse.bodyToFlux(Tweet.class));

        tweetRepo.saveAll(tweets).subscribe(System.out::println);

    };
}
 

import org.springframework.web.reactive.function.client.ClientRequest; //导入依赖的package包/类
/**
 * Adds HTTP tags to the client side span
 */
private void addRequestTags(ClientRequest request) {
	keysInjector().addRequestTags(request.url().toString(),
			request.url().getHost(),
			request.url().getPath(),
			request.method().name(),
			request.headers());
}
 

import org.springframework.web.reactive.function.client.ClientRequest; //导入依赖的package包/类
public static ExchangeFilterFunction setInstance(Mono<Instance> instance) {
    return (request, next) -> instance.map(
            i -> ClientRequest.from(request).attribute(ATTRIBUTE_INSTANCE, i).build())
                                      .switchIfEmpty(request.url().isAbsolute() ?
                                              Mono.just(request) :
                                              Mono.error(new InstanceWebClientException("Instance not found")))
                                      .flatMap(next::exchange);
}
 

import org.springframework.web.reactive.function.client.ClientRequest; //导入依赖的package包/类
public static ExchangeFilterFunction addHeaders(HttpHeadersProvider httpHeadersProvider) {
    return withInstance((instance, request, next) -> {
        ClientRequest newRequest = ClientRequest.from(request)
                                                .headers(headers -> headers.addAll(
                                                        httpHeadersProvider.getHeaders(instance)))
                                                .build();
        return next.exchange(newRequest);
    });
}
 

import org.springframework.web.reactive.function.client.ClientRequest; //导入依赖的package包/类
public static ExchangeFilterFunction convertLegacyEndpoint(LegacyEndpointConverter converter) {
    return (ClientRequest request, ExchangeFunction next) -> {
        Mono<ClientResponse> clientResponse = next.exchange(request);
        if (request.attribute(ATTRIBUTE_ENDPOINT).map(converter::canConvert).orElse(false)) {
            return clientResponse.flatMap(response -> {
                if (response.headers().contentType().map(ACTUATOR_V1_MEDIATYPE::isCompatibleWith).orElse(false)) {
                    return convertClientResponse(converter::convert, ACTUATOR_V2_MEDIATYPE).apply(response);
                }
                return Mono.just(response);
            });
        }
        return clientResponse;
    };
}
 

import org.springframework.web.reactive.function.client.ClientRequest; //导入依赖的package包/类
private ExchangeFilterFunction oAuthToken(String token) {
    return (clientRequest, exchangeFunction) ->
            exchangeFunction
                    .exchange(ClientRequest.from(clientRequest).header("Authorization", "Bearer " + token).build());
}
 

import org.springframework.web.reactive.function.client.ClientRequest; //导入依赖的package包/类
private ClientRequest verifyExchange() {
    ClientRequest request = this.captor.getValue();
    verify(this.exchangeFunction).exchange(request);
    verifyNoMoreInteractions(this.exchangeFunction);
    return request;
}
 

import org.springframework.web.reactive.function.client.ClientRequest; //导入依赖的package包/类
ClientRequestTextMap(ClientRequest readDelegate,
		ClientRequest.Builder writeDelegate) {
	this.readDelegate = readDelegate;
	this.writeDelegate = writeDelegate;
}
 

import org.springframework.web.reactive.function.client.ClientRequest; //导入依赖的package包/类
ClientRequest process(ClientRequest request);