Initial Load Balancer support for WebClient.

fixes gh-237

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

@@ -405,7 +405,7 @@ for a description of what there properties do.
 NOTE: `client` in the above examples should be replaced with your Ribbon client's
 name.
 
-=== Multiple RestTemplate objects
+==== Multiple RestTemplate objects
 
 If you want a `RestTemplate` that is not load balanced, create a `RestTemplate`
 bean and inject it as normal.  To access the load balanced `RestTemplate` use
@@ -451,6 +451,42 @@ public class MyClass {
 
 TIP: If you see errors like `java.lang.IllegalArgumentException: Can not set org.springframework.web.client.RestTemplate field com.my.app.Foo.restTemplate to com.sun.proxy.$Proxy89` try injecting `RestOperations` instead or setting `spring.aop.proxyTargetClass=true`.
 
+[[loadbalanced-webclient]]
+=== Spring WebFlux WebClient as a Load Balancer Client
+
+`WebClient` can be configured to use the `LoadBalancerClient. A `LoadBalancerExchangeFilterFunction` is auto-configured if spring-webflux is on the classpath.
+
+[source,java,indent=0]
+----
+@Configuration
+public class MyConfiguration {
+
+    @LoadBalanced
+    @Bean
+    RestTemplate restTemplate() {
+        return new RestTemplate();
+    }
+}
+
+public class MyClass {
+    @Autowired
+    private LoadBalancerExchangeFilterFunction 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 (ie. service name, not a host name).
+The `LoadBalancerClient` is used to create a full physical address.
+
 [[ignore-network-interfaces]]
 === Ignore Network Interfaces
 

spring-cloud-commons/pom.xml

@@ -33,6 +33,11 @@
 			<artifactId>spring-boot-starter-web</artifactId>
 			<optional>true</optional>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-webflux</artifactId>
+			<optional>true</optional>
+		</dependency>
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-aop</artifactId>

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

@@ -0,0 +1,44 @@
+package org.springframework.cloud.client.loadbalancer.reactive;
+
+import java.net.URI;
+
+import org.springframework.cloud.client.ServiceInstance;
+import org.springframework.cloud.client.loadbalancer.LoadBalancerClient;
+import org.springframework.util.Assert;
+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;
+
+import reactor.core.publisher.Mono;
+
+/**
+ * @author Spencer Gibb
+ */
+public class LoadBalancerExchangeFilterFunction implements ExchangeFilterFunction {
+
+	private final LoadBalancerClient loadBalancerClient;
+
+	public LoadBalancerExchangeFilterFunction(LoadBalancerClient loadBalancerClient) {
+		this.loadBalancerClient = loadBalancerClient;
+	}
+
+	@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);
+	}
+
+}

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

@@ -0,0 +1,22 @@
+package org.springframework.cloud.client.loadbalancer.reactive;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.cloud.client.loadbalancer.LoadBalancerClient;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.web.reactive.function.client.WebClient;
+
+/**
+ * @author Spencer Gibb
+ */
+@Configuration
+@ConditionalOnClass(WebClient.class)
+@ConditionalOnBean(LoadBalancerClient.class)
+public class ReactiveLoadBalancerAutoConfiguration {
+
+	@Bean
+	public LoadBalancerExchangeFilterFunction loadBalancerExchangeFilterFunction(LoadBalancerClient client) {
+		return new LoadBalancerExchangeFilterFunction(client);
+	}
+}

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

@@ -1,15 +1,16 @@
 # AutoConfiguration
 org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
 org.springframework.cloud.client.CommonsClientAutoConfiguration,\
+org.springframework.cloud.client.discovery.composite.CompositeDiscoveryClientAutoConfiguration,\
 org.springframework.cloud.client.discovery.noop.NoopDiscoveryClientAutoConfiguration,\
+org.springframework.cloud.client.discovery.simple.SimpleDiscoveryClientAutoConfiguration,\
 org.springframework.cloud.client.hypermedia.CloudHypermediaAutoConfiguration,\
 org.springframework.cloud.client.loadbalancer.AsyncLoadBalancerAutoConfiguration,\
 org.springframework.cloud.client.loadbalancer.LoadBalancerAutoConfiguration,\
+org.springframework.cloud.client.loadbalancer.reactive.ReactiveLoadBalancerAutoConfiguration,\
 org.springframework.cloud.client.serviceregistry.ServiceRegistryAutoConfiguration,\
-org.springframework.cloud.commons.util.UtilAutoConfiguration,\
-org.springframework.cloud.client.discovery.composite.CompositeDiscoveryClientAutoConfiguration,\
-org.springframework.cloud.client.discovery.simple.SimpleDiscoveryClientAutoConfiguration,\
-org.springframework.cloud.commons.httpclient.HttpClientConfiguration
+org.springframework.cloud.commons.httpclient.HttpClientConfiguration,\
+org.springframework.cloud.commons.util.UtilAutoConfiguration
 
 
 # Environment Post Processors

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

@@ -0,0 +1,116 @@
+package org.springframework.cloud.client.loadbalancer.reactive;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Random;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+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.cloud.client.discovery.simple.SimpleDiscoveryProperties.SimpleServiceInstance;
+import org.springframework.cloud.client.loadbalancer.LoadBalancerClient;
+import org.springframework.cloud.client.loadbalancer.LoadBalancerRequest;
+import org.springframework.context.annotation.Bean;
+import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.reactive.function.client.WebClient;
+import org.springframework.web.util.UriComponentsBuilder;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.springframework.boot.test.context.SpringBootTest.WebEnvironment.RANDOM_PORT;
+
+/**
+ * @author Spencer Gibb
+ */
+@RunWith(SpringRunner.class)
+@SpringBootTest(webEnvironment = RANDOM_PORT)
+public class LoadBalancerExchangeFilterFunctionTests {
+
+	@Autowired
+	private LoadBalancerExchangeFilterFunction lbFunction;
+
+	@Autowired
+	private SimpleDiscoveryProperties properties;
+
+	@LocalServerPort
+	private int port;
+
+	@Before
+	public void before() {
+		SimpleServiceInstance instance = new SimpleServiceInstance();
+		instance.setServiceId("testservice");
+		instance.setUri(URI.create("http://localhost:" + this.port));
+		properties.getInstances().put("testservice", Arrays.asList(instance));
+	}
+
+	@Test
+	public void testFilterFunctionWorks() {
+		String value = WebClient.builder()
+				.baseUrl("http://testservice")
+				.filter(lbFunction)
+				.build()
+				.get()
+				.uri("/hello")
+				.retrieve()
+				.bodyToMono(String.class).block();
+		assertThat(value).isEqualTo("Hello World");
+	}
+
+	@EnableDiscoveryClient
+	@EnableAutoConfiguration
+	@SpringBootConfiguration
+	@RestController
+	static class Config {
+
+		@RequestMapping("/hello")
+		public String hello() {
+			return "Hello World";
+		}
+
+		@Bean
+		LoadBalancerClient loadBalancerClient(DiscoveryClient discoveryClient) {
+			return new LoadBalancerClient() {
+				Random random = new Random();
+
+				@Override
+				public <T> T execute(String serviceId, LoadBalancerRequest<T> request) throws IOException {
+					throw new UnsupportedOperationException();
+				}
+
+				@Override
+				public <T> T execute(String serviceId, ServiceInstance serviceInstance, LoadBalancerRequest<T> request) throws IOException {
+					throw new UnsupportedOperationException();
+				}
+
+				@Override
+				public URI reconstructURI(ServiceInstance instance, URI original) {
+					return UriComponentsBuilder.fromUri(original)
+							.host(instance.getHost())
+							.port(instance.getPort())
+							.build()
+							.toUri();
+				}
+
+				@Override
+				public ServiceInstance choose(String serviceId) {
+					List<ServiceInstance> instances = discoveryClient.getInstances(serviceId);
+					int instanceIdx = random.nextInt(instances.size());
+					return instances.get(instanceIdx);
+				}
+			};
+		}
+
+	}
+}