Gh 576 use reactive load balancer (#584)

* Add `ReactorLoadBalancerClient` interface and its default implementation.

* Add initial `ReactorLoadBalancerExchangeFilterFunction` implementation.
Add `ReactorLoadBalancerClientAutoConfiguration`.
Refactor `ReactorLoadBalancerClient` interface and default implementation.

* Implement configuration changes to make default `ReactorLoadBalancer` and
`ReactorLoadBalancerClient` work out of the box with `@LoadBalanced
WebClient.Builder`.

* Add more tests for ReactorLoadBalancerExchangeFilterFunction and
DefaultReactorLoadBalancer.

* Fix configuration. Add tests. Add documentation.

* Add information on caching to the documentation.

* Add fixes after code review.

* Small refactoring after code review.

* Switch from handle(response, sink) to map(response).

* Remove redundant cast.

* Add link to caching in Springboot reference to the docs.

* Add more information on working with spring-cloud-loadbalancer vs. spring-cloud-starter-netflix-ribbon to the docs.

* Fix after code review.

docs/src/main/asciidoc/spring-cloud-commons.adoc

@@ -371,7 +371,7 @@ See {githubroot}/spring-cloud-netflix/blob/master/spring-cloud-netflix-ribbon/sr
 
 === Spring WebClient as a Load Balancer Client
 
-`WebClient` can be automatically configured to use the `LoadBalancerClient`.
+`WebClient` can be automatically configured to use a load-balancer client.
 To create a load-balanced `WebClient`, create a `WebClient.Builder` `@Bean` and use the `@LoadBalanced` qualifier, as shown in the following example:
 
 [source,java,indent=0]
@@ -400,6 +400,20 @@ public class MyClass {
 The URI needs to use a virtual host name (that is, a service name, not a host name).
 The Ribbon client is used to create a full physical address.
 
+IMPORTANT: If you want to use a `@LoadBalanced WebClient.Builder`, you need to have a loadbalancer
+implementation in the classpath. It is recommended that you add the
+`org.springframework.cloud:spring-cloud-loadbalancer` dependency to your project.
+Then, `ReactiveLoadBalancer` will be used underneath.
+Alternatively, this functionality will also work with spring-cloud-starter-netflix-ribbon, but the request
+will be handled by a non-reactive `LoadBalancerClient` under the hood. Additionally,
+spring-cloud-starter-netflix-ribbon is already in maintenance mode, so we do not recommned
+adding it to new projects.
+
+TIP: The `ReactorLoadBalancer` used underneath supports caching. If `cacheManager` is detected,
+cached version of `ServiceInstanceSupplier` will be used. If not, we will retrieve instances
+from discovery service without caching them. We recommend https://docs.spring.io/spring-boot/docs/current/reference/html/boot-features-caching.html[enabling caching] in your project
+if you use `ReactiveLoadBalancer`.
+
 ==== Retrying Failed Requests
 
 A load-balanced `RestTemplate` can be configured to retry failed requests.
@@ -514,7 +528,42 @@ TIP: If you see errors such as `java.lang.IllegalArgumentException: Can not set
 [[loadbalanced-webclient]]
 === Spring WebFlux WebClient as a Load Balancer Client
 
-`WebClient` can be configured to use the `LoadBalancerClient`. `LoadBalancerExchangeFilterFunction` is auto-configured if `spring-webflux` is on the classpath. The following example shows how to configure a `WebClient` to use load balancer:
+[[webflux-with-reactive-loadbalancer]]
+==== Spring WebFlux WebClient with Reactive Load Balancer
+
+`WebClient` can be configured to use the `ReactiveLoadBalancer`.
+If you add `org.springframework.cloud:spring-cloud-loadbalancer` to your project,
+ `ReactorLoadBalancerExchangeFilterFunction` is auto-configured if `spring-webflux` is on the classpath.
+The following example shows how to configure a `WebClient` to use reactive load balancer under the hood:
+
+[source,java,indent=0]
+----
+public class MyClass {
+    @Autowired
+    private ReactorLoadBalancerExchangeFilterFunction lbFunction;
+
+    public Mono<String> doOtherStuff() {
+        return WebClient.builder().baseUrl("http://stores")
+            .filter(lbFunction)
+            .build()
+            .get()
+            .uri("/stores")
+            .retrieve()
+            .bodyToMono(String.class);
+    }
+}
+----
+
+The URI needs to use a virtual host name (that is, a service name, not a host name).
+The `ReactorLoadBalancerClient` is used to create a full physical address.
+
+==== Spring WebFlux WebClient with non-reactive Load Balancer Client
+
+If you you don't have `org.springframework.cloud:spring-cloud-loadbalancer` in your project,
+but you do have spring-cloud-starter-netflix-ribbon, you can still use `WebClient` with `LoadBalancerClient`. `LoadBalancerExchangeFilterFunction`
+will be auto-configured if `spring-webflux` is on the classpath. Please note, however, that this is
+uses a non-reactive client under the hood.
+The following example shows how to configure a `WebClient` to use load balancer:
 
 [source,java,indent=0]
 ----
@@ -537,6 +586,45 @@ public class MyClass {
 The URI needs to use a virtual host name (that is, a service name, not a host name).
 The `LoadBalancerClient` is used to create a full physical address.
 
+WARN:
+This approach is now deprecated.
+We suggest you use <<webflux-with-reactive-loadbalancer,WebFlux with reactive Load-Balancer>>
+instead.
+
+==== Passing your own Load-Balancer Client configuration
+
+You can also use the `@LoadBalancerClient` annotation to pass your own load-balancer client configuration, passing the name of the load-balancer client and the configuration class, like so:
+
+[source,java,indent=0]
+----
+@Configuration
+@LoadBalancerClient(value = "stores", configuration = StoresLoadBalancerClientConfiguration.class)
+public class MyConfiguration {
+
+	@Bean
+	@LoadBalanced
+	public WebClient.Builder loadBalancedWebClientBuilder() {
+		return WebClient.builder();
+	}
+}
+----
+
+It is also possible to pass together multiple configurations (for more than one load-balancer client) via the `@LoadBalancerClients` annotation, as shown below:
+
+[source,java,indent=0]
+----
+@Configuration
+@LoadBalancerClients({@LoadBalancerClient(value = "stores", configuration = StoresLoadBalancerClientConfiguration.class), @LoadBalancerClient(value = "customers", configuration = CustomersLoadBalancerClientConfiguration.class)})
+public class MyConfiguration {
+
+	@Bean
+	@LoadBalanced
+	public WebClient.Builder loadBalancedWebClientBuilder() {
+		return WebClient.builder();
+	}
+}
+----
+
 [[ignore-network-interfaces]]
 === Ignore Network Interfaces
 

spring-cloud-commons/src/main/java/org/springframework/cloud/client/loadbalancer/reactive/LoadBalancerUriTools.java

@@ -0,0 +1,118 @@
+/*
+ * Copyright 2012-2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.client.loadbalancer.reactive;
+
+import java.net.URI;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+
+import org.springframework.cloud.client.ServiceInstance;
+import org.springframework.web.util.UriComponentsBuilder;
+
+/**
+ * @author Olga Maciaszek-Sharma
+ * @since 2.2.0
+ */
+final class LoadBalancerUriTools {
+
+	private LoadBalancerUriTools() {
+		throw new IllegalStateException("Can't instantiate a utility class");
+	}
+
+	private static final String PERCENTAGE_SIGN = "%";
+
+	private static final String DEFAULT_SCHEME = "http";
+
+	private static final String DEFAULT_SECURE_SCHEME = "https";
+
+	private static final Map<String, String> INSECURE_SCHEME_MAPPINGS;
+
+	static {
+		INSECURE_SCHEME_MAPPINGS = new HashMap<>();
+		INSECURE_SCHEME_MAPPINGS.put(DEFAULT_SCHEME, DEFAULT_SECURE_SCHEME);
+		INSECURE_SCHEME_MAPPINGS.put("ws", "wss");
+	}
+
+	// see original
+	// https://github.com/spring-cloud/spring-cloud-gateway/blob/master/spring-cloud-gateway-core/
+	// src/main/java/org/springframework/cloud/gateway/support/ServerWebExchangeUtils.java
+	private static boolean containsEncodedParts(URI uri) {
+		boolean encoded = (uri.getRawQuery() != null
+				&& uri.getRawQuery().contains(PERCENTAGE_SIGN))
+				|| (uri.getRawPath() != null
+						&& uri.getRawPath().contains(PERCENTAGE_SIGN))
+				|| (uri.getRawFragment() != null
+						&& uri.getRawFragment().contains(PERCENTAGE_SIGN));
+		// Verify if it is really fully encoded. Treat partial encoded as unencoded.
+		if (encoded) {
+			try {
+				UriComponentsBuilder.fromUri(uri).build(true);
+				return true;
+			}
+			catch (IllegalArgumentException ignore) {
+			}
+			return false;
+		}
+		return false;
+	}
+
+	private static int computePort(int port, String scheme) {
+		if (port >= 0) {
+			return port;
+		}
+		if (Objects.equals(scheme, DEFAULT_SECURE_SCHEME)) {
+			return 443;
+		}
+		return 80;
+	}
+
+	static URI reconstructURI(ServiceInstance serviceInstance, URI original) {
+		if (serviceInstance == null) {
+			throw new IllegalArgumentException("Service Instance cannot be null.");
+		}
+		return doReconstructURI(serviceInstance, original);
+	}
+
+	private static URI doReconstructURI(ServiceInstance serviceInstance, URI original) {
+		String host = serviceInstance.getHost();
+		String scheme = Optional.ofNullable(serviceInstance.getScheme())
+				.orElse(computeScheme(original, serviceInstance));
+		int port = computePort(serviceInstance.getPort(), scheme);
+
+		if (Objects.equals(host, original.getHost()) && port == original.getPort()
+				&& Objects.equals(scheme, original.getScheme())) {
+			return original;
+		}
+
+		boolean encoded = containsEncodedParts(original);
+		return UriComponentsBuilder.fromUri(original).scheme(scheme).host(host).port(port)
+				.build(encoded).toUri();
+	}
+
+	private static String computeScheme(URI original, ServiceInstance serviceInstance) {
+		String originalOrDefault = Optional.ofNullable(original.getScheme())
+				.orElse(DEFAULT_SCHEME);
+		if (serviceInstance.isSecure()
+				&& INSECURE_SCHEME_MAPPINGS.containsKey(originalOrDefault)) {
+			return INSECURE_SCHEME_MAPPINGS.get(originalOrDefault);
+		}
+		return originalOrDefault;
+	}
+
+}

spring-cloud-commons/src/main/java/org/springframework/cloud/client/loadbalancer/reactive/ReactiveLoadBalancer.java

@@ -23,6 +23,7 @@ import org.reactivestreams.Publisher;
  *
  * @param <T> type of the response
  * @author Spencer Gibb
+ * @author Olga Maciaszek-Sharma
  */
 public interface ReactiveLoadBalancer<T> {
 
@@ -42,4 +43,11 @@ public interface ReactiveLoadBalancer<T> {
 		return choose(REQUEST);
 	}
 
+	@FunctionalInterface
+	interface Factory<T> {
+
+		ReactiveLoadBalancer<T> getInstance(String serviceId);
+
+	}
+
 }

spring-cloud-commons/src/main/java/org/springframework/cloud/client/loadbalancer/reactive/ReactiveLoadBalancerAutoConfiguration.java

@@ -23,6 +23,7 @@ import org.springframework.beans.factory.SmartInitializingSingleton;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.cloud.client.loadbalancer.LoadBalanced;
 import org.springframework.cloud.client.loadbalancer.LoadBalancerClient;
 import org.springframework.context.annotation.Bean;
@@ -30,11 +31,14 @@ import org.springframework.context.annotation.Configuration;
 import org.springframework.web.reactive.function.client.WebClient;
 
 /**
+ * @deprecated in favour of {@link ReactorLoadBalancerClientAutoConfiguration}
  * @author Spencer Gibb
+ * @author Olga Maciaszek-Sharma
  */
 @Configuration
 @ConditionalOnClass(WebClient.class)
 @ConditionalOnBean(LoadBalancerClient.class)
+@ConditionalOnMissingBean(ReactiveLoadBalancer.Factory.class)
 public class ReactiveLoadBalancerAutoConfiguration {
 
 	@LoadBalanced
@@ -58,7 +62,7 @@ public class ReactiveLoadBalancerAutoConfiguration {
 	}
 
 	@Bean
-	public WebClientCustomizer loadbalanceClientWebClientCustomizer(
+	public WebClientCustomizer loadBalancerClientWebClientCustomizer(
 			LoadBalancerExchangeFilterFunction filterFunction) {
 		return builder -> builder.filter(filterFunction);
 	}

spring-cloud-commons/src/main/java/org/springframework/cloud/client/loadbalancer/reactive/ReactorLoadBalancerClientAutoConfiguration.java

@@ -0,0 +1,80 @@
+/*
+ * Copyright 2012-2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.client.loadbalancer.reactive;
+
+import java.util.Collections;
+import java.util.List;
+
+import org.springframework.beans.factory.SmartInitializingSingleton;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.cloud.client.loadbalancer.LoadBalanced;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.web.reactive.function.client.WebClient;
+
+/**
+ * An auto-configuration that allows the use of a {@link LoadBalanced}
+ * {@link WebClient.Builder} with {@link ReactorLoadBalancerExchangeFilterFunction} and
+ * {@link ReactiveLoadBalancer} used under the hood.
+ *
+ * @author Olga Maciaszek-Sharma
+ * @since 2.2.0
+ */
+@Configuration
+@ConditionalOnClass(WebClient.class)
+@ConditionalOnBean(ReactiveLoadBalancer.Factory.class)
+public class ReactorLoadBalancerClientAutoConfiguration {
+
+	private List<WebClient.Builder> webClientBuilders = Collections.emptyList();
+
+	List<WebClient.Builder> getBuilders() {
+		return this.webClientBuilders;
+	}
+
+	@Bean
+	public SmartInitializingSingleton loadBalancedWebClientInitializer(
+			final List<WebClientCustomizer> customizers) {
+		return () -> {
+			for (WebClient.Builder webClientBuilder : getBuilders()) {
+				for (WebClientCustomizer customizer : customizers) {
+					customizer.customize(webClientBuilder);
+				}
+			}
+		};
+	}
+
+	@Bean
+	public WebClientCustomizer loadBalancerClientWebClientCustomizer(
+			ReactorLoadBalancerExchangeFilterFunction filterFunction) {
+		return builder -> builder.filter(filterFunction);
+	}
+
+	@Bean
+	public ReactorLoadBalancerExchangeFilterFunction loadBalancerExchangeFilterFunction(
+			ReactiveLoadBalancer.Factory loadBalancerFactory) {
+		return new ReactorLoadBalancerExchangeFilterFunction(loadBalancerFactory);
+	}
+
+	@LoadBalanced
+	@Autowired(required = false)
+	void setWebClientBuilders(List<WebClient.Builder> webClientBuilders) {
+		this.webClientBuilders = webClientBuilders;
+	}
+
+}

spring-cloud-commons/src/main/java/org/springframework/cloud/client/loadbalancer/reactive/ReactorLoadBalancerExchangeFilterFunction.java

@@ -0,0 +1,109 @@
+/*
+ * Copyright 2012-2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.client.loadbalancer.reactive;
+
+import java.net.URI;
+import java.util.Objects;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import reactor.core.publisher.Mono;
+
+import org.springframework.cloud.client.ServiceInstance;
+import org.springframework.http.HttpStatus;
+import org.springframework.web.reactive.function.client.ClientRequest;
+import org.springframework.web.reactive.function.client.ClientResponse;
+import org.springframework.web.reactive.function.client.ExchangeFilterFunction;
+import org.springframework.web.reactive.function.client.ExchangeFunction;
+
+/**
+ * An {@link ExchangeFilterFunction} that uses {@link ReactiveLoadBalancer} to execute
+ * requests against a correct {@link ServiceInstance}.
+ *
+ * @author Olga Maciaszek-Sharma
+ * @since 2.2.0
+ */
+public class ReactorLoadBalancerExchangeFilterFunction implements ExchangeFilterFunction {
+
+	private static final Log LOG = LogFactory
+			.getLog(LoadBalancerExchangeFilterFunction.class);
+
+	private final ReactiveLoadBalancer.Factory<ServiceInstance> loadBalancerFactory;
+
+	public ReactorLoadBalancerExchangeFilterFunction(
+			ReactiveLoadBalancer.Factory<ServiceInstance> loadBalancerFactory) {
+		this.loadBalancerFactory = loadBalancerFactory;
+	}
+
+	@Override
+	public Mono<ClientResponse> filter(ClientRequest request, ExchangeFunction next) {
+		URI originalUrl = request.url();
+		String serviceId = originalUrl.getHost();
+		if (serviceId == null) {
+			String message = String.format(
+					"Request URI does not contain a valid hostname: %s",
+					originalUrl.toString());
+			if (LOG.isWarnEnabled()) {
+				LOG.warn(message);
+			}
+			return Mono.just(
+					ClientResponse.create(HttpStatus.BAD_REQUEST).body(message).build());
+		}
+		return choose(serviceId).flatMap(response -> {
+			ServiceInstance instance = response.getServer();
+			if (instance == null) {
+				String message = serviceInstanceUnavailableMessage(serviceId);
+				if (LOG.isWarnEnabled()) {
+					LOG.warn(message);
+				}
+				return Mono.just(ClientResponse.create(HttpStatus.SERVICE_UNAVAILABLE)
+						.body(serviceInstanceUnavailableMessage(serviceId)).build());
+			}
+
+			if (LOG.isDebugEnabled()) {
+				LOG.debug(String.format(
+						"Load balancer has retrieved the instance for service %s: %s",
+						serviceId, Objects.requireNonNull(instance).getUri()));
+			}
+			ClientRequest newRequest = buildClientRequest(request,
+					LoadBalancerUriTools.reconstructURI(instance, originalUrl));
+			return next.exchange(newRequest);
+		});
+	}
+
+	private Mono<Response<ServiceInstance>> choose(String serviceId) {
+		ReactiveLoadBalancer<ServiceInstance> loadBalancer = loadBalancerFactory
+				.getInstance(serviceId);
+		if (loadBalancer == null) {
+			return Mono.just(new EmptyResponse());
+		}
+		return Mono.from(loadBalancer.choose());
+	}
+
+	private String serviceInstanceUnavailableMessage(String serviceId) {
+		return "Load balancer does not contain an instance for the service " + serviceId;
+	}
+
+	private ClientRequest buildClientRequest(ClientRequest request, URI uri) {
+		return ClientRequest.create(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();
+	}
+
+}

spring-cloud-commons/src/main/resources/META-INF/spring.factories

@@ -7,6 +7,7 @@ org.springframework.cloud.client.discovery.simple.SimpleDiscoveryClientAutoConfi
 org.springframework.cloud.client.hypermedia.CloudHypermediaAutoConfiguration,\
 org.springframework.cloud.client.loadbalancer.AsyncLoadBalancerAutoConfiguration,\
 org.springframework.cloud.client.loadbalancer.LoadBalancerAutoConfiguration,\
+org.springframework.cloud.client.loadbalancer.reactive.ReactorLoadBalancerClientAutoConfiguration,\
 org.springframework.cloud.client.loadbalancer.reactive.ReactiveLoadBalancerAutoConfiguration,\
 org.springframework.cloud.client.serviceregistry.ServiceRegistryAutoConfiguration,\
 org.springframework.cloud.commons.httpclient.HttpClientConfiguration,\

spring-cloud-commons/src/test/java/org/springframework/cloud/client/loadbalancer/reactive/LoadBalancerTestUtils.java

@@ -0,0 +1,61 @@
+/*
+ * Copyright 2012-2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.client.loadbalancer.reactive;
+
+import java.util.List;
+
+import org.springframework.boot.WebApplicationType;
+import org.springframework.boot.autoconfigure.web.reactive.function.client.WebClientAutoConfiguration;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.test.util.ReflectionTestUtils;
+import org.springframework.web.reactive.function.client.ExchangeFilterFunction;
+import org.springframework.web.reactive.function.client.WebClient;
+
+import static org.assertj.core.api.BDDAssertions.then;
+
+/**
+ * Utility class for testing reactive load-balancer clients.
+ *
+ * @author Olga Maciaszek-Sharma
+ */
+final class LoadBalancerTestUtils {
+
+	private LoadBalancerTestUtils() {
+		throw new IllegalStateException("Can't instantiate a utility class");
+	}
+
+	static ConfigurableApplicationContext init(Class<?> config, Class<?> clientClass) {
+		return new SpringApplicationBuilder().web(WebApplicationType.NONE)
+				.sources(config, WebClientAutoConfiguration.class, clientClass).run();
+	}
+
+	@SuppressWarnings("unchecked")
+	static List<ExchangeFilterFunction> getFilters(WebClient.Builder builder) {
+		return (List<ExchangeFilterFunction>) ReflectionTestUtils.getField(builder,
+				"filters");
+	}
+
+	static void assertLoadBalanced(WebClient.Builder webClientBuilder,
+			Class<?> exchangeFilterFunctionClass) {
+		List<ExchangeFilterFunction> filters = getFilters(webClientBuilder);
+		then(filters).hasSize(1);
+		ExchangeFilterFunction interceptor = filters.get(0);
+		then(interceptor).isInstanceOf(exchangeFilterFunctionClass);
+	}
+
+}

spring-cloud-commons/src/test/java/org/springframework/cloud/client/loadbalancer/reactive/LoadBalancerUriToolsTests.java

@@ -0,0 +1,276 @@
+/*
+ * Copyright 2012-2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.client.loadbalancer.reactive;
+
+import java.net.URI;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.cloud.client.ServiceInstance;
+import org.springframework.web.util.UriComponentsBuilder;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for {@link LoadBalancerUriTools}.
+ *
+ * @author Olga Maciaszek-Sharma
+ */
+class LoadBalancerUriToolsTests {
+
+	@Test
+	void originalURIReturnedIfDataMatches() {
+		TestServiceInstance serviceInstance = new TestServiceInstance();
+		URI original = UriComponentsBuilder.fromUriString("http://test.example:8080/xxx")
+				.build().toUri();
+
+		URI reconstructed = LoadBalancerUriTools.reconstructURI(serviceInstance,
+				original);
+
+		assertThat(reconstructed).isEqualTo(original);
+	}
+
+	@Test
+	void serviceInstanceHostSet() {
+		TestServiceInstance serviceInstance = new TestServiceInstance();
+		URI original = UriComponentsBuilder
+				.fromUriString("http://testHost.example:8080/xxx").build().toUri();
+
+		URI reconstructed = LoadBalancerUriTools.reconstructURI(serviceInstance,
+				original);
+
+		assertThat(reconstructed).isNotNull();
+		assertThat(reconstructed.getHost()).isEqualTo(serviceInstance.getHost());
+	}
+
+	@Test
+	void serviceInstanceSchemeSet() {
+		TestServiceInstance serviceInstance = new TestServiceInstance()
+				.withScheme("https");
+		URI original = UriComponentsBuilder.fromUriString("http://test.example/xxx")
+				.build().toUri();
+
+		URI reconstructed = LoadBalancerUriTools.reconstructURI(serviceInstance,
+				original);
+
+		assertThat(reconstructed).isNotNull();
+		assertThat(reconstructed.getScheme()).isEqualTo(serviceInstance.getScheme());
+	}
+
+	@Test
+	void originalSchemeSetIfServiceInstanceSchemeMissing() {
+		TestServiceInstance serviceInstance = new TestServiceInstance().withScheme(null);
+		URI original = UriComponentsBuilder.fromUriString("https://test.example/xxx")
+				.build().toUri();
+
+		URI reconstructed = LoadBalancerUriTools.reconstructURI(serviceInstance,
+				original);
+
+		assertThat(reconstructed).isNotNull();
+		assertThat(reconstructed.getScheme()).isEqualTo(original.getScheme());
+	}
+
+	@Test
+	void secureSchemeSetIfServiceInstanceSchemeMissingAndServiceInstanceSecure() {
+		TestServiceInstance serviceInstance = new TestServiceInstance().withScheme(null)
+				.withSecure(true);
+		URI original = UriComponentsBuilder.fromUriString("http://test.example/xxx")
+				.build().toUri();
+
+		URI reconstructed = LoadBalancerUriTools.reconstructURI(serviceInstance,
+				original);
+
+		assertThat(reconstructed).isNotNull();
+		assertThat(reconstructed.getScheme()).isEqualTo("https");
+	}
+
+	@Test
+	void secureWsSchemeSetIfServiceInstanceSchemeMissingAndServiceInstanceSecure() {
+		TestServiceInstance serviceInstance = new TestServiceInstance().withScheme(null)
+				.withSecure(true);
+		URI original = UriComponentsBuilder.fromUriString("ws://test.example/xxx").build()
+				.toUri();
+
+		URI reconstructed = LoadBalancerUriTools.reconstructURI(serviceInstance,
+				original);
+
+		assertThat(reconstructed).isNotNull();
+		assertThat(reconstructed.getScheme()).isEqualTo("wss");
+	}
+
+	@Test
+	void defaultSchemeSetIfMissing() {
+		TestServiceInstance serviceInstance = new TestServiceInstance().withScheme(null);
+		URI original = UriComponentsBuilder.fromUriString("//test.example/xxx").build()
+				.toUri();
+
+		URI reconstructed = LoadBalancerUriTools.reconstructURI(serviceInstance,
+				original);
+
+		assertThat(reconstructed).isNotNull();
+		assertThat(reconstructed.getScheme()).isEqualTo("http");
+	}
+
+	@Test
+	void serviceInstancePortSet() {
+		TestServiceInstance serviceInstance = new TestServiceInstance().withPort(0);
+		URI original = UriComponentsBuilder.fromUriString("http://test.example:8080/xxx")
+				.build().toUri();
+
+		URI reconstructed = LoadBalancerUriTools.reconstructURI(serviceInstance,
+				original);
+
+		assertThat(reconstructed).isNotNull();
+		assertThat(reconstructed.getPort()).isEqualTo(serviceInstance.getPort());
+	}
+
+	@Test
+	void defaultHttpPortSetIfServiceInstancePortIncorrect() {
+		TestServiceInstance serviceInstance = new TestServiceInstance().withPort(-1);
+		URI original = UriComponentsBuilder.fromUriString("http://test.example:8888/xxx")
+				.build().toUri();
+
+		URI reconstructed = LoadBalancerUriTools.reconstructURI(serviceInstance,
+				original);
+
+		assertThat(reconstructed).isNotNull();
+		assertThat(reconstructed.getPort()).isEqualTo(80);
+	}
+
+	@Test
+	void defaultHttpsPortSetIfServiceInstancePortIncorrect() {
+		TestServiceInstance serviceInstance = new TestServiceInstance()
+				.withScheme("https").withPort(-1);
+		URI original = UriComponentsBuilder.fromUriString("http://test.example:8888/xxx")
+				.build().toUri();
+
+		URI reconstructed = LoadBalancerUriTools.reconstructURI(serviceInstance,
+				original);
+
+		assertThat(reconstructed).isNotNull();
+		assertThat(reconstructed.getPort()).isEqualTo(443);
+	}
+
+	@Test
+	void originalUserInfoSet() {
+		TestServiceInstance serviceInstance = new TestServiceInstance();
+		URI original = UriComponentsBuilder.fromUriString(
+				"http://testUser@testHost.example/path?query1=test1&query2=test2#fragment")
+				.build().toUri();
+
+		URI reconstructed = LoadBalancerUriTools.reconstructURI(serviceInstance,
+				original);
+
+		assertThat(reconstructed).isNotNull();
+		assertThat(reconstructed.getRawUserInfo()).isEqualTo(original.getRawUserInfo());
+		assertThat(reconstructed.getRawQuery()).isEqualTo(original.getRawQuery());
+		assertThat(reconstructed.getRawPath()).isEqualTo(original.getRawPath());
+		assertThat(reconstructed.getRawQuery()).isEqualTo(original.getRawQuery());
+		assertThat(reconstructed.getRawFragment()).isEqualTo(original.getRawFragment());
+		assertThat(reconstructed.getHost()).isEqualTo(serviceInstance.getHost());
+		assertThat(reconstructed.getPort()).isEqualTo(serviceInstance.getPort());
+	}
+
+	@Test
+	void reconstructedURIEncodedCorrectly() {
+		TestServiceInstance serviceInstance = new TestServiceInstance();
+		URI original = UriComponentsBuilder.fromUriString(
+				"http://test.example/path%40%21%242?query=val%40%21%242#frag%40%21%242")
+				.build().toUri();
+
+		URI reconstructed = LoadBalancerUriTools.reconstructURI(serviceInstance,
+				original);
+
+		assertThat(reconstructed).isNotNull();
+		assertThat(reconstructed.getRawUserInfo()).isEqualTo(original.getRawUserInfo());
+		assertThat(reconstructed.getRawQuery()).isEqualTo(original.getRawQuery());
+		assertThat(reconstructed.getRawPath()).isEqualTo(original.getRawPath());
+		assertThat(reconstructed.getRawQuery()).isEqualTo(original.getRawQuery());
+		assertThat(reconstructed.getRawFragment()).isEqualTo(original.getRawFragment());
+		assertThat(reconstructed.getHost()).isEqualTo(serviceInstance.getHost());
+		assertThat(reconstructed.getPort()).isEqualTo(serviceInstance.getPort());
+	}
+
+}
+
+class TestServiceInstance implements ServiceInstance {
+
+	private URI uri;
+
+	private String scheme = "http";
+
+	private String host = "test.example";
+
+	private int port = 8080;
+
+	private boolean secure;
+
+	private Map<String, String> metadata = new LinkedHashMap<>();
+
+	TestServiceInstance withScheme(String scheme) {
+		this.scheme = scheme;
+		return this;
+	}
+
+	TestServiceInstance withPort(int port) {
+		this.port = port;
+		return this;
+	}
+
+	TestServiceInstance withSecure(boolean secure) {
+		this.secure = secure;
+		return this;
+	}
+
+	@Override
+	public String getServiceId() {
+		return "test-service";
+	}
+
+	@Override
+	public String getHost() {
+		return host;
+	}
+
+	@Override
+	public int getPort() {
+		return port;
+	}
+
+	@Override
+	public boolean isSecure() {
+		return secure;
+	}
+
+	@Override
+	public URI getUri() {
+		return uri;
+	}
+
+	@Override
+	public Map<String, String> getMetadata() {
+		return metadata;
+	}
+
+	@Override
+	public String getScheme() {
+		return scheme;
+	}
+
+}

spring-cloud-commons/src/test/java/org/springframework/cloud/client/loadbalancer/reactive/ReactiveLoadBalancerAutoConfigurationTests.java

@@ -18,16 +18,12 @@ package org.springframework.cloud.client.loadbalancer.reactive;
 
 import java.io.IOException;
 import java.net.URI;
-import java.util.List;
 import java.util.Map;
 import java.util.Random;
 
 import org.junit.Test;
 
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.WebApplicationType;
-import org.springframework.boot.autoconfigure.web.reactive.function.client.WebClientAutoConfiguration;
-import org.springframework.boot.builder.SpringApplicationBuilder;
 import org.springframework.cloud.client.DefaultServiceInstance;
 import org.springframework.cloud.client.ServiceInstance;
 import org.springframework.cloud.client.loadbalancer.LoadBalanced;
@@ -38,15 +34,15 @@ import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Primary;
-import org.springframework.test.util.ReflectionTestUtils;
-import org.springframework.web.reactive.function.client.ExchangeFilterFunction;
 import org.springframework.web.reactive.function.client.WebClient;
 
 import static org.assertj.core.api.BDDAssertions.then;
+import static org.springframework.cloud.client.loadbalancer.reactive.LoadBalancerTestUtils.getFilters;
 
 /**
  * @author Spencer Gibb
  * @author Tim Ysewyn
+ * @author Olga Maciaszek-Sharma
  */
 public class ReactiveLoadBalancerAutoConfigurationTests {
 
@@ -63,19 +59,6 @@ public class ReactiveLoadBalancerAutoConfigurationTests {
 		assertLoadBalanced(webClientBuilder);
 	}
 
-	private void assertLoadBalanced(WebClient.Builder webClientBuilder) {
-		List<ExchangeFilterFunction> filters = getFilters(webClientBuilder);
-		then(filters).hasSize(1);
-		ExchangeFilterFunction interceptor = filters.get(0);
-		then(interceptor).isInstanceOf(LoadBalancerExchangeFilterFunction.class);
-	}
-
-	@SuppressWarnings("unchecked")
-	private List<ExchangeFilterFunction> getFilters(WebClient.Builder builder) {
-		return (List<ExchangeFilterFunction>) ReflectionTestUtils.getField(builder,
-				"filters");
-	}
-
 	@Test
 	public void multipleWebClientBuilders() {
 		ConfigurableApplicationContext context = init(TwoWebClientBuilders.class);
@@ -107,12 +90,29 @@ public class ReactiveLoadBalancerAutoConfigurationTests {
 		then(getFilters(builder)).isNullOrEmpty();
 	}
 
-	protected ConfigurableApplicationContext init(Class<?> config) {
-		return new SpringApplicationBuilder().web(WebApplicationType.NONE)
-				// .properties("spring.aop.proxyTargetClass=true")
-				.sources(config, WebClientAutoConfiguration.class,
-						ReactiveLoadBalancerAutoConfiguration.class)
-				.run();
+	@Test
+	public void autoConfigurationNotLoadedWhenReactorLoadBalancerClientPresent() {
+		ConfigurableApplicationContext context = init(
+				ReactorLoadBalancerClientPresent.class);
+		final Map<String, WebClient.Builder> webClientBuilders = context
+				.getBeansOfType(WebClient.Builder.class);
+
+		then(webClientBuilders).hasSize(1);
+
+		WebClient.Builder builder = context.getBean(WebClient.Builder.class);
+
+		then(builder).isNotNull();
+		then(getFilters(builder)).isNullOrEmpty();
+	}
+
+	private ConfigurableApplicationContext init(Class<?> config) {
+		return LoadBalancerTestUtils.init(config,
+				ReactiveLoadBalancerAutoConfiguration.class);
+	}
+
+	private void assertLoadBalanced(WebClient.Builder builder) {
+		LoadBalancerTestUtils.assertLoadBalanced(builder,
+				LoadBalancerExchangeFilterFunction.class);
 	}
 
 	@Configuration
@@ -131,6 +131,16 @@ public class ReactiveLoadBalancerAutoConfigurationTests {
 
 	}
 
+	@Configuration
+	protected static class ReactorLoadBalancerClientPresent extends OneWebClientBuilder {
+
+		@Bean
+		ReactiveLoadBalancer.Factory<ServiceInstance> reactiveLoadBalancerFactory() {
+			return serviceId -> new TestReactiveLoadBalancer();
+		}
+
+	}
+
 	@Configuration
 	protected static class OneWebClientBuilder extends NoWebClientBuilder {
 

spring-cloud-commons/src/test/java/org/springframework/cloud/client/loadbalancer/reactive/ReactorLoadBalancerClientAutoConfigurationTests.java

@@ -0,0 +1,147 @@
+/*
+ * Copyright 2012-2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.client.loadbalancer.reactive;
+
+import java.util.Map;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.cloud.client.ServiceInstance;
+import org.springframework.cloud.client.loadbalancer.LoadBalanced;
+import org.springframework.cloud.client.loadbalancer.LoadBalancedRetryFactory;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Primary;
+import org.springframework.web.reactive.function.client.WebClient;
+
+import static org.assertj.core.api.BDDAssertions.then;
+import static org.springframework.cloud.client.loadbalancer.reactive.LoadBalancerTestUtils.getFilters;
+
+/**
+ * Tests for {@link ReactorLoadBalancerClientAutoConfiguration}.
+ *
+ * @author Olga Maciaszek-Sharma
+ */
+public class ReactorLoadBalancerClientAutoConfigurationTests {
+
+	@Test
+	void loadBalancerFilterAddedToWebClientBuilder() {
+		ConfigurableApplicationContext context = init(OneWebClientBuilder.class);
+		final Map<String, WebClient.Builder> webClientBuilders = context
+				.getBeansOfType(WebClient.Builder.class);
+
+		then(webClientBuilders).isNotNull().hasSize(1);
+		WebClient.Builder webClientBuilder = webClientBuilders.values().iterator().next();
+		then(webClientBuilder).isNotNull();
+
+		assertLoadBalanced(webClientBuilder);
+	}
+
+	@Test
+	void loadBalancerFilterAddedOnlyToLoadBalancedWebClientBuilder() {
+		ConfigurableApplicationContext context = init(TwoWebClientBuilders.class);
+		final Map<String, WebClient.Builder> webClientBuilders = context
+				.getBeansOfType(WebClient.Builder.class);
+
+		then(webClientBuilders).hasSize(2);
+
+		TwoWebClientBuilders.Two two = context.getBean(TwoWebClientBuilders.Two.class);
+
+		then(two.loadBalanced).isNotNull();
+		assertLoadBalanced(two.loadBalanced);
+
+		then(two.nonLoadBalanced).isNotNull();
+		then(getFilters(two.nonLoadBalanced)).isNullOrEmpty();
+	}
+
+	@Test
+	void noCustomWebClientBuilders() {
+		ConfigurableApplicationContext context = init(NoWebClientBuilder.class);
+		final Map<String, WebClient.Builder> webClientBuilders = context
+				.getBeansOfType(WebClient.Builder.class);
+
+		then(webClientBuilders).hasSize(1);
+
+		WebClient.Builder builder = context.getBean(WebClient.Builder.class);
+
+		then(builder).isNotNull();
+		then(getFilters(builder)).isNullOrEmpty();
+	}
+
+	private ConfigurableApplicationContext init(Class<?> config) {
+		return LoadBalancerTestUtils.init(config,
+				ReactorLoadBalancerClientAutoConfiguration.class);
+	}
+
+	private void assertLoadBalanced(WebClient.Builder webClientBuilder) {
+		LoadBalancerTestUtils.assertLoadBalanced(webClientBuilder,
+				ReactorLoadBalancerExchangeFilterFunction.class);
+	}
+
+	@Configuration
+	protected static class NoWebClientBuilder {
+
+		@Bean
+		ReactiveLoadBalancer.Factory<ServiceInstance> reactiveLoadBalancerFactory() {
+			return serviceId -> new TestReactiveLoadBalancer();
+		}
+
+		@Bean
+		LoadBalancedRetryFactory loadBalancedRetryFactory() {
+			return new LoadBalancedRetryFactory() {
+			};
+		}
+
+	}
+
+	@Configuration
+	protected static class OneWebClientBuilder extends NoWebClientBuilder {
+
+		@Bean
+		@LoadBalanced
+		WebClient.Builder loadBalancedWebClientBuilder() {
+			return WebClient.builder();
+		}
+
+	}
+
+	@Configuration
+	protected static class TwoWebClientBuilders extends OneWebClientBuilder {
+
+		@Primary
+		@Bean
+		WebClient.Builder webClientBuilder() {
+			return WebClient.builder();
+		}
+
+		@Configuration
+		protected static class Two {
+
+			@Autowired
+			WebClient.Builder nonLoadBalanced;
+
+			@Autowired
+			@LoadBalanced
+			WebClient.Builder loadBalanced;
+
+		}
+
+	}
+
+}

spring-cloud-commons/src/test/java/org/springframework/cloud/client/loadbalancer/reactive/ReactorLoadBalancerExchangeFilterFunctionTests.java

@@ -0,0 +1,152 @@
+/*
+ * Copyright 2012-2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.client.loadbalancer.reactive;
+
+import java.net.URI;
+import java.util.Collections;
+import java.util.List;
+import java.util.Random;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Mono;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.SpringBootConfiguration;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.web.server.LocalServerPort;
+import org.springframework.cloud.client.ServiceInstance;
+import org.springframework.cloud.client.discovery.DiscoveryClient;
+import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
+import org.springframework.cloud.client.discovery.simple.SimpleDiscoveryProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.http.HttpStatus;
+import org.springframework.test.context.junit.jupiter.SpringExtension;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.reactive.function.client.ClientResponse;
+import org.springframework.web.reactive.function.client.WebClient;
+
+import static org.assertj.core.api.BDDAssertions.then;
+import static org.springframework.boot.test.context.SpringBootTest.WebEnvironment.RANDOM_PORT;
+
+/**
+ * Tests for {@link ReactorLoadBalancerExchangeFilterFunction}.
+ *
+ * @author Olga Maciaszek-Sharma
+ */
+@ExtendWith(SpringExtension.class)
+@SpringBootTest(webEnvironment = RANDOM_PORT)
+class ReactorLoadBalancerExchangeFilterFunctionTests {
+
+	@Autowired
+	private ReactorLoadBalancerExchangeFilterFunction loadBalancerFunction;
+
+	@Autowired
+	private SimpleDiscoveryProperties properties;
+
+	@LocalServerPort
+	private int port;
+
+	@BeforeEach
+	void setUp() {
+		SimpleDiscoveryProperties.SimpleServiceInstance instance = new SimpleDiscoveryProperties.SimpleServiceInstance();
+		instance.setServiceId("testservice");
+		instance.setUri(URI.create("http://localhost:" + this.port));
+		this.properties.getInstances().put("testservice",
+				Collections.singletonList(instance));
+	}
+
+	@Test
+	void correctResponseReturnedForExistingHostAndInstancePresent() {
+		ClientResponse clientResponse = WebClient.builder().baseUrl("http://testservice")
+				.filter(this.loadBalancerFunction).build().get().uri("/hello").exchange()
+				.block();
+		then(clientResponse.statusCode()).isEqualTo(HttpStatus.OK);
+		then(clientResponse.bodyToMono(String.class).block()).isEqualTo("Hello World");
+	}
+
+	@Test
+	void serviceUnavailableReturnedWhenNoInstancePresent() {
+		ClientResponse clientResponse = WebClient.builder().baseUrl("http://xxx")
+				.filter(this.loadBalancerFunction).build().get().exchange().block();
+		then(clientResponse.statusCode()).isEqualTo(HttpStatus.SERVICE_UNAVAILABLE);
+	}
+
+	@Test
+	void badRequestReturnedForIncorrectHost() {
+		ClientResponse clientResponse = WebClient.builder().baseUrl("http:///xxx")
+				.filter(this.loadBalancerFunction).build().get().exchange().block();
+		then(clientResponse.statusCode()).isEqualTo(HttpStatus.BAD_REQUEST);
+	}
+
+	@EnableDiscoveryClient
+	@EnableAutoConfiguration
+	@SpringBootConfiguration
+	@RestController
+	static class Config {
+
+		@RequestMapping("/hello")
+		public String hello() {
+			return "Hello World";
+		}
+
+		@Bean
+		ReactiveLoadBalancer.Factory<ServiceInstance> reactiveLoadBalancerFactory(
+				DiscoveryClient discoveryClient) {
+			return serviceId -> new DiscoveryClientBasedReactiveLoadBalancer(serviceId,
+					discoveryClient);
+		}
+
+	}
+
+}
+
+class DiscoveryClientBasedReactiveLoadBalancer
+		implements ReactiveLoadBalancer<ServiceInstance> {
+
+	private final Random random = new Random();
+
+	private final String serviceId;
+
+	private final DiscoveryClient discoveryClient;
+
+	DiscoveryClientBasedReactiveLoadBalancer(String serviceId,
+			DiscoveryClient discoveryClient) {
+		this.serviceId = serviceId;
+		this.discoveryClient = discoveryClient;
+	}
+
+	@Override
+	public Publisher<Response<ServiceInstance>> choose() {
+		List<ServiceInstance> instances = discoveryClient.getInstances(serviceId);
+		if (instances.size() == 0) {
+			return Mono.just(new EmptyResponse());
+		}
+		int instanceIdx = this.random.nextInt(instances.size());
+		return Mono.just(new DefaultResponse(instances.get(instanceIdx)));
+	}
+
+	@Override
+	public Publisher<Response<ServiceInstance>> choose(Request request) {
+		return choose();
+	}
+
+}

spring-cloud-commons/src/test/java/org/springframework/cloud/client/loadbalancer/reactive/TestReactiveLoadBalancer.java

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2012-2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.client.loadbalancer.reactive;
+
+import java.util.Random;
+
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Mono;
+
+import org.springframework.cloud.client.DefaultServiceInstance;
+import org.springframework.cloud.client.ServiceInstance;
+
+/**
+ * A sample implementation of {@link ReactiveLoadBalancer} used for tests.
+ *
+ * @author Olga Maciaszek-Sharma
+ */
+class TestReactiveLoadBalancer implements ReactiveLoadBalancer<ServiceInstance> {
+
+	private static final String TEST_SERVICE_ID = "testServiceId";
+
+	private final Random random = new Random();
+
+	@Override
+	public Publisher<Response<ServiceInstance>> choose() {
+		return Mono.just(new DefaultResponse(new DefaultServiceInstance(TEST_SERVICE_ID,
+				TEST_SERVICE_ID, TEST_SERVICE_ID, random.nextInt(40000), false)));
+	}
+
+	@Override
+	public Publisher<Response<ServiceInstance>> choose(Request request) {
+		return choose();
+	}
+
+}

spring-cloud-loadbalancer/pom.xml

@@ -56,7 +56,7 @@
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-webflux</artifactId>
-			<scope>test</scope>
+			<optional>true</optional>
 		</dependency>
 		<dependency>
 			<groupId>org.springframework.boot</groupId>

spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/annotation/LoadBalancerClientConfiguration.java

@@ -17,25 +17,34 @@
 package org.springframework.cloud.loadbalancer.annotation;
 
 import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.cache.CacheManager;
+import org.springframework.cloud.client.ConditionalOnDiscoveryEnabled;
+import org.springframework.cloud.client.ServiceInstance;
 import org.springframework.cloud.client.discovery.DiscoveryClient;
 import org.springframework.cloud.loadbalancer.core.CachingServiceInstanceSupplier;
 import org.springframework.cloud.loadbalancer.core.DiscoveryClientServiceInstanceSupplier;
+import org.springframework.cloud.loadbalancer.core.ReactorLoadBalancer;
+import org.springframework.cloud.loadbalancer.core.RoundRobinLoadBalancer;
 import org.springframework.cloud.loadbalancer.core.ServiceInstanceSupplier;
+import org.springframework.cloud.loadbalancer.support.LoadBalancerClientFactory;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.core.env.Environment;
 
 /**
  * @author Spencer Gibb
+ * @author Olga Maciaszek-Sharma
  */
 @Configuration
 @EnableConfigurationProperties
+@ConditionalOnDiscoveryEnabled
 public class LoadBalancerClientConfiguration {
 
 	@Bean
+	@ConditionalOnBean(DiscoveryClient.class)
 	@ConditionalOnMissingBean
 	public ServiceInstanceSupplier discoveryClientServiceInstanceSupplier(
 			DiscoveryClient discoveryClient, Environment env,
@@ -50,4 +59,14 @@ public class LoadBalancerClientConfiguration {
 		return delegate;
 	}
 
+	@Bean
+	@ConditionalOnMissingBean
+	public ReactorLoadBalancer<ServiceInstance> reactorServiceInstanceLoadBalancer(
+			Environment environment,
+			LoadBalancerClientFactory loadBalancerClientFactory) {
+		String name = environment.getProperty(LoadBalancerClientFactory.PROPERTY_NAME);
+		return new RoundRobinLoadBalancer(name, loadBalancerClientFactory
+				.getLazyProvider(name, ServiceInstanceSupplier.class));
+	}
+
 }

spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/config/LoadBalancerAutoConfiguration.java

@@ -20,6 +20,9 @@ import java.util.Collections;
 import java.util.List;
 
 import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.boot.autoconfigure.AutoConfigureBefore;
+import org.springframework.cloud.client.loadbalancer.reactive.ReactiveLoadBalancerAutoConfiguration;
+import org.springframework.cloud.client.loadbalancer.reactive.ReactorLoadBalancerClientAutoConfiguration;
 import org.springframework.cloud.loadbalancer.annotation.LoadBalancerClientSpecification;
 import org.springframework.cloud.loadbalancer.annotation.LoadBalancerClients;
 import org.springframework.cloud.loadbalancer.support.LoadBalancerClientFactory;
@@ -28,9 +31,12 @@ import org.springframework.context.annotation.Configuration;
 
 /**
  * @author Spencer Gibb
+ * @author Olga Maciaszek-Sharma
  */
 @Configuration
 @LoadBalancerClients
+@AutoConfigureBefore({ ReactorLoadBalancerClientAutoConfiguration.class,
+		ReactiveLoadBalancerAutoConfiguration.class })
 // @EnableCaching //TODO: how to enforce, or check conditions?
 // @AutoConfigureBefore(CacheAutoConfiguration.class)
 public class LoadBalancerAutoConfiguration {

spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/core/ReactorServiceInstanceLoadBalancer.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2012-2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.loadbalancer.core;
+
+import org.springframework.cloud.client.ServiceInstance;
+
+/**
+ * A marker interface for {@link ReactorLoadBalancer} that allows selecting
+ * {@link ServiceInstance} objects.
+ *
+ * @author Olga Maciaszek-Sharma
+ * @since 2.2.0
+ */
+public interface ReactorServiceInstanceLoadBalancer
+		extends ReactorLoadBalancer<ServiceInstance> {
+
+}

spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/core/RoundRobinLoadBalancer.java

@@ -33,7 +33,7 @@ import org.springframework.cloud.client.loadbalancer.reactive.Response;
 /**
  * @author Spencer Gibb
  */
-public class RoundRobinLoadBalancer implements ReactorLoadBalancer<ServiceInstance> {
+public class RoundRobinLoadBalancer implements ReactorServiceInstanceLoadBalancer {
 
 	private static final Log log = LogFactory.getLog(RoundRobinLoadBalancer.class);
 

spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/support/LoadBalancerClientFactory.java

@@ -16,9 +16,12 @@
 
 package org.springframework.cloud.loadbalancer.support;
 
+import org.springframework.cloud.client.ServiceInstance;
+import org.springframework.cloud.client.loadbalancer.reactive.ReactiveLoadBalancer;
 import org.springframework.cloud.context.named.NamedContextFactory;
 import org.springframework.cloud.loadbalancer.annotation.LoadBalancerClientConfiguration;
 import org.springframework.cloud.loadbalancer.annotation.LoadBalancerClientSpecification;
+import org.springframework.cloud.loadbalancer.core.ReactorServiceInstanceLoadBalancer;
 import org.springframework.core.env.Environment;
 
 /**
@@ -28,9 +31,11 @@ import org.springframework.core.env.Environment;
  *
  * @author Spencer Gibb
  * @author Dave Syer
+ * @author Olga Maciaszek-Sharma
  */
 public class LoadBalancerClientFactory
-		extends NamedContextFactory<LoadBalancerClientSpecification> {
+		extends NamedContextFactory<LoadBalancerClientSpecification>
+		implements ReactiveLoadBalancer.Factory<ServiceInstance> {
 
 	/**
 	 * Property source name for load balancer.
@@ -50,4 +55,9 @@ public class LoadBalancerClientFactory
 		return environment.getProperty(PROPERTY_NAME);
 	}
 
+	@Override
+	public ReactiveLoadBalancer<ServiceInstance> getInstance(String serviceId) {
+		return getInstance(serviceId, ReactorServiceInstanceLoadBalancer.class);
+	}
+
 }