Added support for reactive service discovery

spring-cloud-kubernetes-core/src/main/java/org/springframework/cloud/kubernetes/ConditionalOnKubernetesEnabled.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2019-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.kubernetes;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Inherited;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+
+/**
+ * Provides a more succinct conditional <code>spring.cloud.kubernetes.enabled</code>.
+ *
+ * @author Tim Ysewyn
+ * @since 2.2.0
+ */
+@Target(ElementType.TYPE)
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+@Inherited
+@ConditionalOnProperty(value = "spring.cloud.kubernetes.enabled", matchIfMissing = true)
+public @interface ConditionalOnKubernetesEnabled {
+
+}

spring-cloud-kubernetes-core/src/main/java/org/springframework/cloud/kubernetes/KubernetesAutoConfiguration.java

@@ -29,7 +29,6 @@ import org.springframework.boot.actuate.autoconfigure.health.ConditionalOnEnable
 import org.springframework.boot.actuate.health.HealthIndicator;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -39,9 +38,10 @@ import org.springframework.context.annotation.Configuration;
  *
  * @author Ioannis Canellos
  * @author Eddú Meléndez
+ * @author Tim Ysewyn
  */
 @Configuration
-@ConditionalOnProperty(value = "spring.cloud.kubernetes.enabled", matchIfMissing = true)
+@ConditionalOnKubernetesEnabled
 @EnableConfigurationProperties(KubernetesClientProperties.class)
 public class KubernetesAutoConfiguration {
 

spring-cloud-kubernetes-discovery/pom.xml

@@ -42,6 +42,21 @@
 			<artifactId>spring-boot-autoconfigure</artifactId>
 			<optional>true</optional>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-actuator</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-webflux</artifactId>
+			<optional>true</optional>
+		</dependency>
 		<dependency>
 			<groupId>org.springframework.cloud</groupId>
 			<artifactId>spring-cloud-commons</artifactId>
@@ -125,6 +140,11 @@
 			<version>${spring-cloud-config.version}</version>
 			<scope>test</scope>
 		</dependency>
+		<dependency>
+			<groupId>io.projectreactor</groupId>
+			<artifactId>reactor-test</artifactId>
+			<scope>test</scope>
+		</dependency>
 	</dependencies>
 
 </project>

spring-cloud-kubernetes-discovery/src/main/java/org/springframework/cloud/kubernetes/discovery/ConditionalOnKubernetesDiscoveryEnabled.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2019-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.kubernetes.discovery;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Inherited;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+
+/**
+ * Provides a more succinct conditional
+ * <code>spring.cloud.kubernetes.discovery.enabled</code>.
+ *
+ * @author Tim Ysewyn
+ * @since 2.2.0
+ */
+@Target(ElementType.TYPE)
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+@Inherited
+@ConditionalOnProperty(value = "spring.cloud.kubernetes.discovery.enabled",
+		matchIfMissing = true)
+public @interface ConditionalOnKubernetesDiscoveryEnabled {
+
+}

spring-cloud-kubernetes-discovery/src/main/java/org/springframework/cloud/kubernetes/discovery/KubernetesDiscoveryClientAutoConfiguration.java

@@ -18,12 +18,15 @@ package org.springframework.cloud.kubernetes.discovery;
 
 import io.fabric8.kubernetes.client.KubernetesClient;
 
+import org.springframework.boot.autoconfigure.AutoConfigureAfter;
 import org.springframework.boot.autoconfigure.AutoConfigureBefore;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.cloud.client.CommonsClientAutoConfiguration;
+import org.springframework.cloud.client.ConditionalOnBlockingDiscoveryEnabled;
 import org.springframework.cloud.client.ConditionalOnDiscoveryEnabled;
 import org.springframework.cloud.client.discovery.simple.SimpleDiscoveryClientAutoConfiguration;
+import org.springframework.cloud.kubernetes.ConditionalOnKubernetesEnabled;
+import org.springframework.cloud.kubernetes.KubernetesAutoConfiguration;
 import org.springframework.cloud.kubernetes.registry.KubernetesRegistration;
 import org.springframework.cloud.kubernetes.registry.KubernetesServiceRegistry;
 import org.springframework.context.annotation.Bean;
@@ -37,9 +40,10 @@ import org.springframework.context.annotation.Configuration;
  */
 @Configuration
 @ConditionalOnDiscoveryEnabled
-@ConditionalOnProperty(name = "spring.cloud.kubernetes.enabled", matchIfMissing = true)
+@ConditionalOnKubernetesEnabled
 @AutoConfigureBefore({ SimpleDiscoveryClientAutoConfiguration.class,
 		CommonsClientAutoConfiguration.class })
+@AutoConfigureAfter({ KubernetesAutoConfiguration.class })
 public class KubernetesDiscoveryClientAutoConfiguration {
 
 	@Bean
@@ -72,18 +76,6 @@ public class KubernetesDiscoveryClientAutoConfiguration {
 		}
 	}
 
-	@Bean
-	@ConditionalOnMissingBean
-	@ConditionalOnProperty(name = "spring.cloud.kubernetes.discovery.enabled",
-			matchIfMissing = true)
-	public KubernetesDiscoveryClient kubernetesDiscoveryClient(KubernetesClient client,
-			KubernetesDiscoveryProperties properties,
-			KubernetesClientServicesFunction kubernetesClientServicesFunction,
-			DefaultIsServicePortSecureResolver isServicePortSecureResolver) {
-		return new KubernetesDiscoveryClient(client, properties,
-				kubernetesClientServicesFunction, isServicePortSecureResolver);
-	}
-
 	@Bean
 	public KubernetesServiceRegistry getServiceRegistry() {
 		return new KubernetesServiceRegistry();
@@ -100,4 +92,21 @@ public class KubernetesDiscoveryClientAutoConfiguration {
 		return new KubernetesDiscoveryProperties();
 	}
 
+	@Configuration
+	@ConditionalOnBlockingDiscoveryEnabled
+	@ConditionalOnKubernetesDiscoveryEnabled
+	public static class KubernetesDiscoveryClientConfiguration {
+
+		@Bean
+		@ConditionalOnMissingBean
+		public KubernetesDiscoveryClient kubernetesDiscoveryClient(
+				KubernetesClient client, KubernetesDiscoveryProperties properties,
+				KubernetesClientServicesFunction kubernetesClientServicesFunction,
+				DefaultIsServicePortSecureResolver isServicePortSecureResolver) {
+			return new KubernetesDiscoveryClient(client, properties,
+					kubernetesClientServicesFunction, isServicePortSecureResolver);
+		}
+
+	}
+
 }

spring-cloud-kubernetes-discovery/src/main/java/org/springframework/cloud/kubernetes/discovery/reactive/KubernetesReactiveDiscoveryClient.java

@@ -0,0 +1,69 @@
+/*
+ * Copyright 2019-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.kubernetes.discovery.reactive;
+
+import io.fabric8.kubernetes.client.KubernetesClient;
+import reactor.core.publisher.Flux;
+import reactor.core.scheduler.Schedulers;
+
+import org.springframework.cloud.client.ServiceInstance;
+import org.springframework.cloud.client.discovery.ReactiveDiscoveryClient;
+import org.springframework.cloud.kubernetes.discovery.KubernetesClientServicesFunction;
+import org.springframework.cloud.kubernetes.discovery.KubernetesDiscoveryClient;
+import org.springframework.cloud.kubernetes.discovery.KubernetesDiscoveryProperties;
+import org.springframework.util.Assert;
+
+/**
+ * Kubernetes implementation of {@link ReactiveDiscoveryClient}. Currently relies on the
+ * {@link KubernetesDiscoveryClient} for feature parity.
+ *
+ * @author Tim Ysewyn
+ */
+public class KubernetesReactiveDiscoveryClient implements ReactiveDiscoveryClient {
+
+	private final KubernetesDiscoveryClient kubernetesDiscoveryClient;
+
+	public KubernetesReactiveDiscoveryClient(KubernetesClient client,
+			KubernetesDiscoveryProperties properties,
+			KubernetesClientServicesFunction kubernetesClientServicesFunction) {
+		this.kubernetesDiscoveryClient = new KubernetesDiscoveryClient(client, properties,
+				kubernetesClientServicesFunction);
+	}
+
+	@Override
+	public String description() {
+		return "Kubernetes Reactive Discovery Client";
+	}
+
+	@Override
+	public Flux<ServiceInstance> getInstances(String serviceId) {
+		Assert.notNull(serviceId,
+				"[Assertion failed] - the object argument must not be null");
+		return Flux
+				.defer(() -> Flux
+						.fromIterable(kubernetesDiscoveryClient.getInstances(serviceId)))
+				.subscribeOn(Schedulers.boundedElastic());
+	}
+
+	@Override
+	public Flux<String> getServices() {
+		return Flux
+				.defer(() -> Flux.fromIterable(kubernetesDiscoveryClient.getServices()))
+				.subscribeOn(Schedulers.boundedElastic());
+	}
+
+}

spring-cloud-kubernetes-discovery/src/main/java/org/springframework/cloud/kubernetes/discovery/reactive/KubernetesReactiveDiscoveryClientAutoConfiguration.java

@@ -0,0 +1,76 @@
+/*
+ * Copyright 2013-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.kubernetes.discovery.reactive;
+
+import io.fabric8.kubernetes.client.KubernetesClient;
+
+import org.springframework.boot.autoconfigure.AutoConfigureAfter;
+import org.springframework.boot.autoconfigure.AutoConfigureBefore;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.cloud.client.ConditionalOnDiscoveryEnabled;
+import org.springframework.cloud.client.ConditionalOnDiscoveryHealthIndicatorEnabled;
+import org.springframework.cloud.client.ConditionalOnReactiveDiscoveryEnabled;
+import org.springframework.cloud.client.ReactiveCommonsClientAutoConfiguration;
+import org.springframework.cloud.client.discovery.composite.reactive.ReactiveCompositeDiscoveryClientAutoConfiguration;
+import org.springframework.cloud.client.discovery.health.DiscoveryClientHealthIndicatorProperties;
+import org.springframework.cloud.client.discovery.health.reactive.ReactiveDiscoveryClientHealthIndicator;
+import org.springframework.cloud.client.discovery.simple.reactive.SimpleReactiveDiscoveryClientAutoConfiguration;
+import org.springframework.cloud.kubernetes.ConditionalOnKubernetesEnabled;
+import org.springframework.cloud.kubernetes.discovery.ConditionalOnKubernetesDiscoveryEnabled;
+import org.springframework.cloud.kubernetes.discovery.KubernetesClientServicesFunction;
+import org.springframework.cloud.kubernetes.discovery.KubernetesDiscoveryClientAutoConfiguration;
+import org.springframework.cloud.kubernetes.discovery.KubernetesDiscoveryProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * Auto configuration for reactive discovery client.
+ *
+ * @author Tim Ysewyn
+ */
+@Configuration
+@ConditionalOnDiscoveryEnabled
+@ConditionalOnReactiveDiscoveryEnabled
+@ConditionalOnKubernetesEnabled
+@ConditionalOnKubernetesDiscoveryEnabled
+@AutoConfigureBefore({ SimpleReactiveDiscoveryClientAutoConfiguration.class,
+		ReactiveCommonsClientAutoConfiguration.class })
+@AutoConfigureAfter({ ReactiveCompositeDiscoveryClientAutoConfiguration.class,
+		KubernetesDiscoveryClientAutoConfiguration.class })
+public class KubernetesReactiveDiscoveryClientAutoConfiguration {
+
+	@Bean
+	@ConditionalOnMissingBean
+	public KubernetesReactiveDiscoveryClient kubernetesReactiveDiscoveryClient(
+			KubernetesClient client, KubernetesDiscoveryProperties properties,
+			KubernetesClientServicesFunction kubernetesClientServicesFunction) {
+		return new KubernetesReactiveDiscoveryClient(client, properties,
+				kubernetesClientServicesFunction);
+	}
+
+	@Bean
+	@ConditionalOnClass(
+			name = "org.springframework.boot.actuate.health.ReactiveHealthIndicator")
+	@ConditionalOnDiscoveryHealthIndicatorEnabled
+	public ReactiveDiscoveryClientHealthIndicator kubernetesReactiveDiscoveryClientHealthIndicator(
+			KubernetesReactiveDiscoveryClient client,
+			DiscoveryClientHealthIndicatorProperties properties) {
+		return new ReactiveDiscoveryClientHealthIndicator(client, properties);
+	}
+
+}

spring-cloud-kubernetes-discovery/src/main/resources/META-INF/spring.factories

@@ -1,5 +1,6 @@
 org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
 org.springframework.cloud.kubernetes.discovery.KubernetesCatalogWatchAutoConfiguration, \
-org.springframework.cloud.kubernetes.discovery.KubernetesDiscoveryClientAutoConfiguration
+org.springframework.cloud.kubernetes.discovery.KubernetesDiscoveryClientAutoConfiguration, \
+org.springframework.cloud.kubernetes.discovery.reactive.KubernetesReactiveDiscoveryClientAutoConfiguration
 org.springframework.cloud.bootstrap.BootstrapConfiguration=\
 org.springframework.cloud.kubernetes.discovery.KubernetesDiscoveryClientConfigClientBootstrapConfiguration

spring-cloud-kubernetes-discovery/src/test/java/org/springframework/cloud/kubernetes/discovery/reactive/KubernetesReactiveDiscoveryClientAutoConfigurationTests.java

@@ -0,0 +1,127 @@
+/*
+ * Copyright 2013-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.kubernetes.discovery.reactive;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.test.context.FilteredClassLoader;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.cloud.client.ReactiveCommonsClientAutoConfiguration;
+import org.springframework.cloud.client.discovery.ReactiveDiscoveryClient;
+import org.springframework.cloud.client.discovery.health.reactive.ReactiveDiscoveryClientHealthIndicator;
+import org.springframework.cloud.commons.util.UtilAutoConfiguration;
+import org.springframework.cloud.kubernetes.KubernetesAutoConfiguration;
+import org.springframework.cloud.kubernetes.discovery.KubernetesDiscoveryClientAutoConfiguration;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Tim Ysewyn
+ */
+class KubernetesReactiveDiscoveryClientAutoConfigurationTests {
+
+	private ApplicationContextRunner contextRunner = new ApplicationContextRunner()
+			.withConfiguration(AutoConfigurations.of(UtilAutoConfiguration.class,
+					ReactiveCommonsClientAutoConfiguration.class,
+					KubernetesAutoConfiguration.class,
+					KubernetesDiscoveryClientAutoConfiguration.class,
+					KubernetesReactiveDiscoveryClientAutoConfiguration.class));
+
+	@Test
+	public void shouldWorkWithDefaults() {
+		contextRunner.run(context -> {
+			assertThat(context).hasSingleBean(ReactiveDiscoveryClient.class);
+			assertThat(context)
+					.hasSingleBean(ReactiveDiscoveryClientHealthIndicator.class);
+		});
+	}
+
+	@Test
+	public void shouldNotHaveDiscoveryClientWhenDiscoveryDisabled() {
+		contextRunner.withPropertyValues("spring.cloud.discovery.enabled=false")
+				.run(context -> {
+					assertThat(context)
+							.doesNotHaveBean("kubernetesReactiveDiscoveryClient");
+					assertThat(context).doesNotHaveBean(ReactiveDiscoveryClient.class);
+					assertThat(context).doesNotHaveBean(
+							ReactiveDiscoveryClientHealthIndicator.class);
+				});
+	}
+
+	@Test
+	public void shouldNotHaveDiscoveryClientWhenReactiveDiscoveryDisabled() {
+		contextRunner.withPropertyValues("spring.cloud.discovery.reactive.enabled=false")
+				.run(context -> {
+					assertThat(context)
+							.doesNotHaveBean("kubernetesReactiveDiscoveryClient");
+					assertThat(context).doesNotHaveBean(ReactiveDiscoveryClient.class);
+					assertThat(context).doesNotHaveBean(
+							ReactiveDiscoveryClientHealthIndicator.class);
+				});
+	}
+
+	@Test
+	public void shouldNotHaveDiscoveryClientWhenKubernetesDisabled() {
+		contextRunner.withPropertyValues("spring.cloud.kubernetes.enabled=false")
+				.run(context -> {
+					assertThat(context)
+							.doesNotHaveBean("kubernetesReactiveDiscoveryClient");
+					assertThat(context).doesNotHaveBean(ReactiveDiscoveryClient.class);
+					assertThat(context).doesNotHaveBean(
+							ReactiveDiscoveryClientHealthIndicator.class);
+				});
+	}
+
+	@Test
+	public void shouldNotHaveDiscoveryClientWhenKubernetesDiscoveryDisabled() {
+		contextRunner
+				.withPropertyValues("spring.cloud.kubernetes.discovery.enabled=false")
+				.run(context -> {
+					assertThat(context)
+							.doesNotHaveBean("kubernetesReactiveDiscoveryClient");
+					assertThat(context).doesNotHaveBean(ReactiveDiscoveryClient.class);
+					assertThat(context).doesNotHaveBean(
+							ReactiveDiscoveryClientHealthIndicator.class);
+				});
+	}
+
+	@Test
+	public void worksWithoutWebflux() {
+		contextRunner
+				.withClassLoader(
+						new FilteredClassLoader("org.springframework.web.reactive"))
+				.run(context -> {
+					assertThat(context).doesNotHaveBean(ReactiveDiscoveryClient.class);
+					assertThat(context).doesNotHaveBean(
+							ReactiveDiscoveryClientHealthIndicator.class);
+				});
+	}
+
+	@Test
+	public void worksWithoutActuator() {
+		contextRunner
+				.withClassLoader(
+						new FilteredClassLoader("org.springframework.boot.actuate"))
+				.run(context -> {
+					assertThat(context).hasSingleBean(ReactiveDiscoveryClient.class);
+					assertThat(context).doesNotHaveBean(
+							ReactiveDiscoveryClientHealthIndicator.class);
+				});
+	}
+
+}

spring-cloud-kubernetes-discovery/src/test/java/org/springframework/cloud/kubernetes/discovery/reactive/KubernetesReactiveDiscoveryClientTests.java

@@ -0,0 +1,329 @@
+/*
+ * Copyright 2019-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.kubernetes.discovery.reactive;
+
+import java.util.HashMap;
+
+import io.fabric8.kubernetes.api.model.Endpoints;
+import io.fabric8.kubernetes.api.model.EndpointsBuilder;
+import io.fabric8.kubernetes.api.model.EndpointsList;
+import io.fabric8.kubernetes.api.model.ServiceBuilder;
+import io.fabric8.kubernetes.api.model.ServiceListBuilder;
+import io.fabric8.kubernetes.client.Config;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.server.mock.KubernetesServer;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import reactor.core.publisher.Flux;
+import reactor.test.StepVerifier;
+
+import org.springframework.cloud.client.ServiceInstance;
+import org.springframework.cloud.client.discovery.ReactiveDiscoveryClient;
+import org.springframework.cloud.kubernetes.discovery.KubernetesDiscoveryProperties;
+import org.springframework.cloud.kubernetes.discovery.support.KubernetesExtension;
+import org.springframework.cloud.kubernetes.discovery.support.KubernetesExtension.Client;
+import org.springframework.cloud.kubernetes.discovery.support.KubernetesExtension.Server;
+
+import static java.util.Collections.singletonList;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Tim Ysewyn
+ */
+@ExtendWith(KubernetesExtension.class)
+class KubernetesReactiveDiscoveryClientTests {
+
+	@BeforeEach
+	public void setup(@Client KubernetesClient kubernetesClient) {
+		// Configure the kubernetes master url to point to the mock server
+		System.setProperty(Config.KUBERNETES_MASTER_SYSTEM_PROPERTY,
+				kubernetesClient.getConfiguration().getMasterUrl());
+		System.setProperty(Config.KUBERNETES_TRUST_CERT_SYSTEM_PROPERTY, "true");
+		System.setProperty(Config.KUBERNETES_AUTH_TRYKUBECONFIG_SYSTEM_PROPERTY, "false");
+		System.setProperty(Config.KUBERNETES_AUTH_TRYSERVICEACCOUNT_SYSTEM_PROPERTY,
+				"false");
+		System.setProperty(Config.KUBERNETES_HTTP2_DISABLE, "true");
+	}
+
+	@Test
+	public void verifyDefaults(@Client KubernetesClient kubernetesClient) {
+		KubernetesDiscoveryProperties properties = new KubernetesDiscoveryProperties();
+		ReactiveDiscoveryClient client = new KubernetesReactiveDiscoveryClient(
+				kubernetesClient, properties, KubernetesClient::services);
+		assertThat(client.description())
+				.isEqualTo("Kubernetes Reactive Discovery Client");
+		assertThat(client.getOrder()).isEqualTo(ReactiveDiscoveryClient.DEFAULT_ORDER);
+	}
+
+	@Test
+	public void shouldReturnFluxOfServices(@Client KubernetesClient kubernetesClient,
+			@Server KubernetesServer kubernetesServer) {
+		kubernetesServer.expect().get().withPath("/api/v1/namespaces/test/services")
+				.andReturn(200, new ServiceListBuilder().addNewItem().withNewMetadata()
+						.withName("s1").withLabels(new HashMap<String, String>() {
+							{
+								put("label", "value");
+							}
+						}).endMetadata().endItem().addNewItem().withNewMetadata()
+						.withName("s2").withLabels(new HashMap<String, String>() {
+							{
+								put("label", "value");
+								put("label2", "value2");
+							}
+						}).endMetadata().endItem().addNewItem().withNewMetadata()
+						.withName("s3").endMetadata().endItem().build())
+				.once();
+
+		KubernetesDiscoveryProperties properties = new KubernetesDiscoveryProperties();
+		ReactiveDiscoveryClient client = new KubernetesReactiveDiscoveryClient(
+				kubernetesClient, properties, KubernetesClient::services);
+		Flux<String> services = client.getServices();
+		StepVerifier.create(services).expectNext("s1", "s2", "s3").expectComplete()
+				.verify();
+	}
+
+	@Test
+	public void shouldReturnEmptyFluxOfServicesWhenNoInstancesFound(
+			@Client KubernetesClient kubernetesClient,
+			@Server KubernetesServer kubernetesServer) {
+		kubernetesServer.expect().get().withPath("/api/v1/namespaces/test/services")
+				.andReturn(200, new ServiceListBuilder().build()).once();
+
+		KubernetesDiscoveryProperties properties = new KubernetesDiscoveryProperties();
+		ReactiveDiscoveryClient client = new KubernetesReactiveDiscoveryClient(
+				kubernetesClient, properties, KubernetesClient::services);
+		Flux<String> services = client.getServices();
+		StepVerifier.create(services).expectNextCount(0).expectComplete().verify();
+	}
+
+	@Test
+	public void shouldReturnEmptyFluxForNonExistingService(
+			@Client KubernetesClient kubernetesClient) {
+		KubernetesDiscoveryProperties properties = new KubernetesDiscoveryProperties();
+		ReactiveDiscoveryClient client = new KubernetesReactiveDiscoveryClient(
+				kubernetesClient, properties, KubernetesClient::services);
+		Flux<ServiceInstance> instances = client.getInstances("nonexistent-service");
+		StepVerifier.create(instances).expectNextCount(0).expectComplete().verify();
+	}
+
+	@Test
+	public void shouldReturnEmptyFluxWhenServiceHasNoSubsets(
+			@Client KubernetesClient kubernetesClient,
+			@Server KubernetesServer kubernetesServer) {
+		kubernetesServer.expect().get().withPath("/api/v1/namespaces/test/services")
+				.andReturn(200,
+						new ServiceListBuilder().addNewItem().withNewMetadata()
+								.withName("existing-service")
+								.withLabels(new HashMap<String, String>() {
+									{
+										put("label", "value");
+									}
+								}).endMetadata().endItem().build())
+				.once();
+
+		KubernetesDiscoveryProperties properties = new KubernetesDiscoveryProperties();
+		ReactiveDiscoveryClient client = new KubernetesReactiveDiscoveryClient(
+				kubernetesClient, properties, KubernetesClient::services);
+		Flux<ServiceInstance> instances = client.getInstances("existing-service");
+		StepVerifier.create(instances).expectNextCount(0).expectComplete().verify();
+	}
+
+	@Test
+	public void shouldReturnFlux(@Client KubernetesClient kubernetesClient,
+			@Server KubernetesServer kubernetesServer) {
+		kubernetesServer.expect().get().withPath("/api/v1/namespaces/test/services")
+				.andReturn(200,
+						new ServiceListBuilder().addNewItem().withNewMetadata()
+								.withName("existing-service")
+								.withLabels(new HashMap<String, String>() {
+									{
+										put("label", "value");
+									}
+								}).endMetadata().endItem().build())
+				.once();
+
+		Endpoints endPoints = new EndpointsBuilder().withNewMetadata()
+				.withName("endpoint").withNamespace("test").endMetadata().addNewSubset()
+				.addNewAddress().withIp("ip1").withNewTargetRef().withUid("uid1")
+				.endTargetRef().endAddress().addNewPort("http", 80, "TCP").endSubset()
+				.build();
+
+		kubernetesServer.expect().get()
+				.withPath("/api/v1/namespaces/test/endpoints/existing-service")
+				.andReturn(200, endPoints).once();
+
+		kubernetesServer.expect().get()
+				.withPath("/api/v1/namespaces/test/services/existing-service")
+				.andReturn(200,
+						new ServiceBuilder().withNewMetadata()
+								.withName("existing-service")
+								.withLabels(new HashMap<String, String>() {
+									{
+										put("label", "value");
+									}
+								}).endMetadata().build())
+				.once();
+
+		KubernetesDiscoveryProperties properties = new KubernetesDiscoveryProperties();
+		ReactiveDiscoveryClient client = new KubernetesReactiveDiscoveryClient(
+				kubernetesClient, properties, KubernetesClient::services);
+		Flux<ServiceInstance> instances = client.getInstances("existing-service");
+		StepVerifier.create(instances).expectNextCount(1).expectComplete().verify();
+	}
+
+	@Test
+	public void shouldReturnFluxWithPrefixedMetadata(
+			@Client KubernetesClient kubernetesClient,
+			@Server KubernetesServer kubernetesServer) {
+		kubernetesServer.expect().get().withPath("/api/v1/namespaces/test/services")
+				.andReturn(200,
+						new ServiceListBuilder().addNewItem().withNewMetadata()
+								.withName("existing-service")
+								.withLabels(new HashMap<String, String>() {
+									{
+										put("label", "value");
+									}
+								}).endMetadata().endItem().build())
+				.once();
+
+		Endpoints endPoints = new EndpointsBuilder().withNewMetadata()
+				.withName("endpoint").withNamespace("test").endMetadata().addNewSubset()
+				.addNewAddress().withIp("ip1").withNewTargetRef().withUid("uid1")
+				.endTargetRef().endAddress().addNewPort("http", 80, "TCP").endSubset()
+				.build();
+
+		kubernetesServer.expect().get()
+				.withPath("/api/v1/namespaces/test/endpoints/existing-service")
+				.andReturn(200, endPoints).once();
+
+		kubernetesServer.expect().get()
+				.withPath("/api/v1/namespaces/test/services/existing-service")
+				.andReturn(200,
+						new ServiceBuilder().withNewMetadata()
+								.withName("existing-service")
+								.withLabels(new HashMap<String, String>() {
+									{
+										put("label", "value");
+									}
+								}).endMetadata().build())
+				.once();
+
+		KubernetesDiscoveryProperties properties = new KubernetesDiscoveryProperties();
+		properties.getMetadata().setAnnotationsPrefix("annotation.");
+		properties.getMetadata().setLabelsPrefix("label.");
+		properties.getMetadata().setPortsPrefix("port.");
+		ReactiveDiscoveryClient client = new KubernetesReactiveDiscoveryClient(
+				kubernetesClient, properties, KubernetesClient::services);
+		Flux<ServiceInstance> instances = client.getInstances("existing-service");
+		StepVerifier.create(instances).expectNextCount(1).expectComplete().verify();
+	}
+
+	@Test
+	public void shouldReturnFluxWhenServiceHasMultiplePortsAndPrimaryPortNameIsSet(
+			@Client KubernetesClient kubernetesClient,
+			@Server KubernetesServer kubernetesServer) {
+		kubernetesServer.expect().get().withPath("/api/v1/namespaces/test/services")
+				.andReturn(200,
+						new ServiceListBuilder().addNewItem().withNewMetadata()
+								.withName("existing-service")
+								.withLabels(new HashMap<String, String>() {
+									{
+										put("label", "value");
+									}
+								}).endMetadata().endItem().build())
+				.once();
+
+		Endpoints endPoints = new EndpointsBuilder().withNewMetadata()
+				.withName("endpoint").withNamespace("test").endMetadata().addNewSubset()
+				.addNewAddress().withIp("ip1").withNewTargetRef().withUid("uid1")
+				.endTargetRef().endAddress().addNewPort("http", 80, "TCP")
+				.addNewPort("https", 443, "TCP").endSubset().build();
+
+		kubernetesServer.expect().get()
+				.withPath("/api/v1/namespaces/test/endpoints/existing-service")
+				.andReturn(200, endPoints).once();
+
+		kubernetesServer.expect().get()
+				.withPath("/api/v1/namespaces/test/services/existing-service")
+				.andReturn(200,
+						new ServiceBuilder().withNewMetadata()
+								.withName("existing-service")
+								.withLabels(new HashMap<String, String>() {
+									{
+										put("label", "value");
+									}
+								}).endMetadata().build())
+				.once();
+
+		KubernetesDiscoveryProperties properties = new KubernetesDiscoveryProperties();
+		properties.setPrimaryPortName("https");
+		ReactiveDiscoveryClient client = new KubernetesReactiveDiscoveryClient(
+				kubernetesClient, properties, KubernetesClient::services);
+		Flux<ServiceInstance> instances = client.getInstances("existing-service");
+		StepVerifier.create(instances).expectNextCount(1).expectComplete().verify();
+	}
+
+	@Test
+	public void shouldReturnFluxOfServicesAcrossAllNamespaces(
+			@Client KubernetesClient kubernetesClient,
+			@Server KubernetesServer kubernetesServer) {
+		kubernetesServer.expect().get().withPath("/api/v1/namespaces/test/services")
+				.andReturn(200,
+						new ServiceListBuilder().addNewItem().withNewMetadata()
+								.withName("existing-service")
+								.withLabels(new HashMap<String, String>() {
+									{
+										put("label", "value");
+									}
+								}).endMetadata().endItem().build())
+				.once();
+
+		Endpoints endpoints = new EndpointsBuilder().withNewMetadata()
+				.withName("endpoint").withNamespace("test").endMetadata().addNewSubset()
+				.addNewAddress().withIp("ip1").withNewTargetRef().withUid("uid1")
+				.endTargetRef().endAddress().addNewPort("http", 80, "TCP")
+				.addNewPort("https", 443, "TCP").endSubset().build();
+
+		EndpointsList endpointsList = new EndpointsList();
+		endpointsList.setItems(singletonList(endpoints));
+
+		kubernetesServer.expect().get().withPath(
+				"/api/v1/endpoints?fieldSelector=metadata.name%3Dexisting-service")
+				.andReturn(200, endpointsList).once();
+
+		kubernetesServer.expect().get()
+				.withPath("/api/v1/namespaces/test/services/existing-service")
+				.andReturn(200,
+						new ServiceBuilder().withNewMetadata()
+								.withName("existing-service")
+								.withLabels(new HashMap<String, String>() {
+									{
+										put("label", "value");
+									}
+								}).endMetadata().build())
+				.once();
+
+		KubernetesDiscoveryProperties properties = new KubernetesDiscoveryProperties();
+		properties.setAllNamespaces(true);
+		ReactiveDiscoveryClient client = new KubernetesReactiveDiscoveryClient(
+				kubernetesClient, properties, KubernetesClient::services);
+		Flux<ServiceInstance> instances = client.getInstances("existing-service");
+		StepVerifier.create(instances).expectNextCount(1).expectComplete().verify();
+	}
+
+}

spring-cloud-kubernetes-discovery/src/test/java/org/springframework/cloud/kubernetes/discovery/support/KubernetesExtension.java

@@ -0,0 +1,91 @@
+/*
+ * Copyright 2019-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.kubernetes.discovery.support;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.server.mock.KubernetesServer;
+import org.junit.jupiter.api.extension.AfterEachCallback;
+import org.junit.jupiter.api.extension.BeforeEachCallback;
+import org.junit.jupiter.api.extension.ExtensionContext;
+import org.junit.jupiter.api.extension.ParameterContext;
+import org.junit.jupiter.api.extension.ParameterResolutionException;
+import org.junit.jupiter.api.extension.ParameterResolver;
+
+/**
+ * @author Tim Ysewyn
+ */
+public class KubernetesExtension
+		implements ParameterResolver, BeforeEachCallback, AfterEachCallback {
+
+	private final KubernetesServer mockServer = new KubernetesServer();
+
+	@Override
+	public boolean supportsParameter(ParameterContext parameterContext,
+			ExtensionContext context) {
+		return (parameterContext.getParameter().isAnnotationPresent(Server.class)
+				&& KubernetesServer.class
+						.isAssignableFrom(parameterContext.getParameter().getType()))
+				|| (parameterContext.getParameter().isAnnotationPresent(Client.class)
+						&& KubernetesClient.class.isAssignableFrom(
+								parameterContext.getParameter().getType()));
+	}
+
+	@Override
+	public void beforeEach(ExtensionContext context) throws Exception {
+		mockServer.before();
+	}
+
+	@Override
+	public void afterEach(ExtensionContext context) throws Exception {
+		mockServer.after();
+	}
+
+	@Override
+	public Object resolveParameter(ParameterContext parameterContext,
+			ExtensionContext extensionContext) throws ParameterResolutionException {
+		if (parameterContext.getParameter().isAnnotationPresent(Client.class)) {
+			return mockServer.getClient();
+		}
+		else {
+			return mockServer;
+		}
+	}
+
+	/**
+	 * Enables injection of kubernetes server to test.
+	 */
+	@Target({ ElementType.PARAMETER })
+	@Retention(RetentionPolicy.RUNTIME)
+	public @interface Server {
+
+	}
+
+	/**
+	 * Enables injection of kubernetes client to test.
+	 */
+	@Target({ ElementType.PARAMETER })
+	@Retention(RetentionPolicy.RUNTIME)
+	public @interface Client {
+
+	}
+
+}

spring-cloud-kubernetes-integration-tests/simple-core/src/test/java/org/springframework/cloud/kubernetes/it/GreetingAndHealthIT.java

@@ -46,7 +46,7 @@ public class GreetingAndHealthIT {
 	public void testHealthEndpoint() {
 		given().baseUri(String.format("%s://%s:%d", PROTOCOL, HOST, PORT))
 				.contentType("application/json").get("actuator/health").then()
-				.statusCode(200).body("details.kubernetes.details.inside", is(true));
+				.statusCode(200).body("components.kubernetes.details.inside", is(true));
 	}
 
 }