Added support for reactive service discovery

spring-cloud-consul-discovery/pom.xml

@@ -133,6 +133,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-actuator</artifactId>
@@ -147,6 +152,11 @@
 			<artifactId>spring-cloud-test-support</artifactId>
 			<scope>test</scope>
 		</dependency>
+		<dependency>
+			<groupId>io.projectreactor</groupId>
+			<artifactId>reactor-test</artifactId>
+			<scope>test</scope>
+		</dependency>
 	</dependencies>
 
 </project>

spring-cloud-consul-discovery/src/main/java/org/springframework/cloud/consul/discovery/ConditionalOnConsulDiscoveryEnabled.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.consul.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.consul.discovery.enabled</code>.
+ *
+ * @author Tim Ysewyn
+ * @since 2.2.0
+ */
+@Target(ElementType.TYPE)
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+@Inherited
+@ConditionalOnProperty(value = "spring.cloud.consul.discovery.enabled",
+		matchIfMissing = true)
+public @interface ConditionalOnConsulDiscoveryEnabled {
+
+}

spring-cloud-consul-discovery/src/main/java/org/springframework/cloud/consul/discovery/ConsulDiscoveryClientConfiguration.java

@@ -18,15 +18,18 @@ package org.springframework.cloud.consul.discovery;
 
 import com.ecwid.consul.v1.ConsulClient;
 
+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.boot.context.properties.EnableConfigurationProperties;
 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.commons.util.InetUtils;
+import org.springframework.cloud.commons.util.UtilAutoConfiguration;
 import org.springframework.cloud.consul.ConditionalOnConsulEnabled;
+import org.springframework.cloud.consul.ConsulAutoConfiguration;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
@@ -36,13 +39,14 @@ import org.springframework.context.annotation.Configuration;
  * @author Tim Ysewyn
  */
 @Configuration
-@ConditionalOnConsulEnabled
-@ConditionalOnProperty(value = "spring.cloud.consul.discovery.enabled",
-		matchIfMissing = true)
 @ConditionalOnDiscoveryEnabled
+@ConditionalOnBlockingDiscoveryEnabled
+@ConditionalOnConsulEnabled
+@ConditionalOnConsulDiscoveryEnabled
 @EnableConfigurationProperties
 @AutoConfigureBefore({ SimpleDiscoveryClientAutoConfiguration.class,
 		CommonsClientAutoConfiguration.class })
+@AutoConfigureAfter({ UtilAutoConfiguration.class, ConsulAutoConfiguration.class })
 public class ConsulDiscoveryClientConfiguration {
 
 	/**

spring-cloud-consul-discovery/src/main/java/org/springframework/cloud/consul/discovery/configclient/ConsulDiscoveryClientConfigServiceBootstrapConfiguration.java

@@ -22,19 +22,21 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.cloud.config.client.ConfigServicePropertySourceLocator;
 import org.springframework.cloud.consul.ConsulAutoConfiguration;
 import org.springframework.cloud.consul.discovery.ConsulDiscoveryClientConfiguration;
+import org.springframework.cloud.consul.discovery.reactive.ConsulReactiveDiscoveryClientConfiguration;
 import org.springframework.context.annotation.Configuration;
 
 /**
  * Helper for config client that wants to lookup the config server via discovery.
  *
  * @author Spencer Gibb
+ * @author Tim Ysewyn
  */
 @ConditionalOnClass(ConfigServicePropertySourceLocator.class)
-@ConditionalOnProperty(value = "spring.cloud.config.discovery.enabled",
-		matchIfMissing = false)
+@ConditionalOnProperty("spring.cloud.config.discovery.enabled")
 @Configuration
 @ImportAutoConfiguration({ ConsulAutoConfiguration.class,
-		ConsulDiscoveryClientConfiguration.class })
+		ConsulDiscoveryClientConfiguration.class,
+		ConsulReactiveDiscoveryClientConfiguration.class })
 public class ConsulDiscoveryClientConfigServiceBootstrapConfiguration {
 
 }

spring-cloud-consul-discovery/src/main/java/org/springframework/cloud/consul/discovery/reactive/ConsulReactiveDiscoveryClient.java

@@ -0,0 +1,127 @@
+/*
+ * 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.consul.discovery.reactive;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import com.ecwid.consul.v1.ConsulClient;
+import com.ecwid.consul.v1.QueryParams;
+import com.ecwid.consul.v1.Response;
+import com.ecwid.consul.v1.health.model.HealthService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import reactor.core.publisher.Flux;
+import reactor.core.scheduler.Schedulers;
+
+import org.springframework.cloud.client.DefaultServiceInstance;
+import org.springframework.cloud.client.ServiceInstance;
+import org.springframework.cloud.client.discovery.ReactiveDiscoveryClient;
+import org.springframework.cloud.consul.discovery.ConsulDiscoveryProperties;
+import org.springframework.util.StringUtils;
+
+import static org.springframework.cloud.consul.discovery.ConsulServerUtils.findHost;
+import static org.springframework.cloud.consul.discovery.ConsulServerUtils.getMetadata;
+
+/**
+ * Consul version of {@link ReactiveDiscoveryClient}.
+ *
+ * @author Tim Ysewyn
+ */
+public class ConsulReactiveDiscoveryClient implements ReactiveDiscoveryClient {
+
+	private static final Logger logger = LoggerFactory
+			.getLogger(ConsulReactiveDiscoveryClient.class);
+
+	private final ConsulClient client;
+
+	private final ConsulDiscoveryProperties properties;
+
+	public ConsulReactiveDiscoveryClient(ConsulClient client,
+			ConsulDiscoveryProperties properties) {
+		this.client = client;
+		this.properties = properties;
+	}
+
+	@Override
+	public String description() {
+		return "Spring Cloud Consul Reactive Discovery Client";
+	}
+
+	@Override
+	public Flux<ServiceInstance> getInstances(String serviceId) {
+		return Flux.defer(() -> {
+			List<ServiceInstance> instances = new ArrayList<>();
+			for (HealthService healthService : getHealthServices(serviceId)) {
+				instances.add(mapToServiceInstance(healthService, serviceId));
+			}
+			return Flux.fromIterable(instances);
+		}).onErrorResume(exception -> {
+			logger.error("Error getting instances from Consul.", exception);
+			return Flux.empty();
+		}).subscribeOn(Schedulers.boundedElastic());
+	}
+
+	private List<HealthService> getHealthServices(String serviceId) {
+		Response<List<HealthService>> services = StringUtils
+				.hasText(properties.getAclToken())
+						? client.getHealthServices(serviceId,
+								properties.getDefaultQueryTag(),
+								properties.isQueryPassing(), QueryParams.DEFAULT,
+								properties.getAclToken())
+						: client.getHealthServices(serviceId,
+								properties.getDefaultQueryTag(),
+								properties.isQueryPassing(), QueryParams.DEFAULT);
+		return services == null ? Collections.emptyList() : services.getValue();
+	}
+
+	private ServiceInstance mapToServiceInstance(HealthService service,
+			String serviceId) {
+		String host = findHost(service);
+		Map<String, String> metadata = getMetadata(service);
+		boolean secure = false;
+		if (metadata.containsKey("secure")) {
+			secure = Boolean.parseBoolean(metadata.get("secure"));
+		}
+		return new DefaultServiceInstance(service.getService().getId(), serviceId, host,
+				service.getService().getPort(), secure, metadata);
+	}
+
+	@Override
+	public Flux<String> getServices() {
+		return Flux.defer(() -> {
+			Response<Map<String, List<String>>> services = StringUtils
+					.hasText(properties.getAclToken())
+							? client.getCatalogServices(QueryParams.DEFAULT,
+									properties.getAclToken())
+							: client.getCatalogServices(QueryParams.DEFAULT);
+			return services == null ? Flux.empty()
+					: Flux.fromIterable(services.getValue().keySet());
+		}).onErrorResume(exception -> {
+			logger.error("Error getting services from Consul.", exception);
+			return Flux.empty();
+		}).subscribeOn(Schedulers.boundedElastic());
+	}
+
+	@Override
+	public int getOrder() {
+		return properties.getOrder();
+	}
+
+}

spring-cloud-consul-discovery/src/main/java/org/springframework/cloud/consul/discovery/reactive/ConsulReactiveDiscoveryClientConfiguration.java

@@ -0,0 +1,80 @@
+/*
+ * 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.consul.discovery.reactive;
+
+import com.ecwid.consul.v1.ConsulClient;
+
+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.boot.context.properties.EnableConfigurationProperties;
+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.commons.util.InetUtils;
+import org.springframework.cloud.commons.util.UtilAutoConfiguration;
+import org.springframework.cloud.consul.ConditionalOnConsulEnabled;
+import org.springframework.cloud.consul.ConsulAutoConfiguration;
+import org.springframework.cloud.consul.discovery.ConditionalOnConsulDiscoveryEnabled;
+import org.springframework.cloud.consul.discovery.ConsulDiscoveryProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * @author Tim Ysewyn
+ */
+@Configuration
+@ConditionalOnDiscoveryEnabled
+@ConditionalOnReactiveDiscoveryEnabled
+@ConditionalOnConsulEnabled
+@ConditionalOnConsulDiscoveryEnabled
+@EnableConfigurationProperties(DiscoveryClientHealthIndicatorProperties.class)
+@AutoConfigureBefore(ReactiveCommonsClientAutoConfiguration.class)
+@AutoConfigureAfter({ UtilAutoConfiguration.class,
+		ReactiveCompositeDiscoveryClientAutoConfiguration.class,
+		ConsulAutoConfiguration.class })
+public class ConsulReactiveDiscoveryClientConfiguration {
+
+	@Bean
+	@ConditionalOnMissingBean
+	public ConsulDiscoveryProperties consulDiscoveryProperties(InetUtils inetUtils) {
+		return new ConsulDiscoveryProperties(inetUtils);
+	}
+
+	@Bean
+	@ConditionalOnMissingBean
+	public ConsulReactiveDiscoveryClient consulReactiveDiscoveryClient(
+			ConsulClient client, ConsulDiscoveryProperties discoveryProperties) {
+		return new ConsulReactiveDiscoveryClient(client, discoveryProperties);
+	}
+
+	@Bean
+	@ConditionalOnClass(
+			name = "org.springframework.boot.actuate.health.ReactiveHealthIndicator")
+	@ConditionalOnDiscoveryHealthIndicatorEnabled
+	public ReactiveDiscoveryClientHealthIndicator consulReactiveDiscoveryClientHealthIndicator(
+			ConsulReactiveDiscoveryClient client,
+			DiscoveryClientHealthIndicatorProperties properties) {
+		return new ReactiveDiscoveryClientHealthIndicator(client, properties);
+	}
+
+}

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

@@ -4,6 +4,7 @@ org.springframework.cloud.consul.discovery.configclient.ConsulConfigServerAutoCo
 org.springframework.cloud.consul.serviceregistry.ConsulAutoServiceRegistrationAutoConfiguration,\
 org.springframework.cloud.consul.serviceregistry.ConsulServiceRegistryAutoConfiguration,\
 org.springframework.cloud.consul.discovery.ConsulDiscoveryClientConfiguration,\
+org.springframework.cloud.consul.discovery.reactive.ConsulReactiveDiscoveryClientConfiguration,\
 org.springframework.cloud.consul.discovery.ConsulCatalogWatchAutoConfiguration, \
 org.springframework.cloud.consul.support.ConsulHeartbeatAutoConfiguration
 org.springframework.cloud.bootstrap.BootstrapConfiguration=\

spring-cloud-consul-discovery/src/test/java/org/springframework/cloud/consul/discovery/reactive/ConsulReactiveDiscoveryClientConfigurationTests.java

@@ -0,0 +1,120 @@
+/*
+ * 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.consul.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.consul.ConsulAutoConfiguration;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Tim Ysewyn
+ */
+class ConsulReactiveDiscoveryClientConfigurationTests {
+
+	private ApplicationContextRunner contextRunner = new ApplicationContextRunner()
+			.withConfiguration(AutoConfigurations.of(UtilAutoConfiguration.class,
+					ReactiveCommonsClientAutoConfiguration.class,
+					ConsulAutoConfiguration.class,
+					ConsulReactiveDiscoveryClientConfiguration.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("consulReactiveDiscoveryClient");
+					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("consulReactiveDiscoveryClient");
+					assertThat(context).doesNotHaveBean(ReactiveDiscoveryClient.class);
+					assertThat(context).doesNotHaveBean(
+							ReactiveDiscoveryClientHealthIndicator.class);
+				});
+	}
+
+	@Test
+	public void shouldNotHaveDiscoveryClientWhenConsulDisabled() {
+		contextRunner.withPropertyValues("spring.cloud.consul.enabled=false")
+				.run(context -> {
+					assertThat(context).doesNotHaveBean("consulReactiveDiscoveryClient");
+					assertThat(context).doesNotHaveBean(ReactiveDiscoveryClient.class);
+					assertThat(context).doesNotHaveBean(
+							ReactiveDiscoveryClientHealthIndicator.class);
+				});
+	}
+
+	@Test
+	public void shouldNotHaveDiscoveryClientWhenConsulDiscoveryDisabled() {
+		contextRunner.withPropertyValues("spring.cloud.consul.discovery.enabled=false")
+				.run(context -> {
+					assertThat(context).doesNotHaveBean("consulReactiveDiscoveryClient");
+					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-consul-discovery/src/test/java/org/springframework/cloud/consul/discovery/reactive/ConsulReactiveDiscoveryClientTests.java

@@ -0,0 +1,181 @@
+/*
+ * 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.consul.discovery.reactive;
+
+import java.util.List;
+import java.util.Map;
+
+import com.ecwid.consul.v1.ConsulClient;
+import com.ecwid.consul.v1.QueryParams;
+import com.ecwid.consul.v1.Response;
+import com.ecwid.consul.v1.health.model.HealthService;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import reactor.core.publisher.Flux;
+import reactor.test.StepVerifier;
+
+import org.springframework.cloud.client.ServiceInstance;
+import org.springframework.cloud.consul.discovery.ConsulDiscoveryProperties;
+
+import static java.util.Collections.emptyList;
+import static java.util.Collections.singletonList;
+import static java.util.Collections.singletonMap;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * @author Tim Ysewyn
+ */
+@ExtendWith(MockitoExtension.class)
+class ConsulReactiveDiscoveryClientTests {
+
+	@Mock
+	private ConsulClient consulClient;
+
+	@Mock
+	private ConsulDiscoveryProperties properties;
+
+	@InjectMocks
+	private ConsulReactiveDiscoveryClient client;
+
+	@Test
+	public void verifyDefaults() {
+		when(properties.getOrder()).thenReturn(1);
+		assertThat(client.description())
+				.isEqualTo("Spring Cloud Consul Reactive Discovery Client");
+		assertThat(client.getOrder()).isEqualTo(1);
+	}
+
+	@Test
+	public void shouldReturnEmptyFluxOfServicesWhenConsulFails() {
+		Flux<String> services = client.getServices();
+		when(consulClient.getCatalogServices(QueryParams.DEFAULT))
+				.thenThrow(new RuntimeException("Possible runtime exception"));
+		StepVerifier.create(services).expectNextCount(0).expectComplete().verify();
+		verify(consulClient).getCatalogServices(QueryParams.DEFAULT);
+	}
+
+	@Test
+	public void shouldReturnFluxOfServices() {
+		Flux<String> services = client.getServices();
+		when(consulClient.getCatalogServices(QueryParams.DEFAULT))
+				.thenReturn(consulServicesResponse());
+		StepVerifier.create(services).expectNext("my-service").expectComplete().verify();
+		verify(properties).getAclToken();
+		verify(consulClient).getCatalogServices(QueryParams.DEFAULT);
+	}
+
+	@Test
+	public void shouldReturnFluxOfServicesWithAclToken() {
+		when(properties.getAclToken()).thenReturn("aclToken");
+		when(consulClient.getCatalogServices(QueryParams.DEFAULT, "aclToken"))
+				.thenReturn(consulServicesResponse());
+		Flux<String> services = client.getServices();
+		StepVerifier.create(services).expectNext("my-service").expectComplete().verify();
+		verify(properties, times(2)).getAclToken();
+		verify(consulClient).getCatalogServices(QueryParams.DEFAULT, "aclToken");
+	}
+
+	@Test
+	public void shouldReturnEmptyFluxForNonExistingService() {
+		configureCommonProperties();
+		when(consulClient.getHealthServices("nonexistent-service", "queryTag", false,
+				QueryParams.DEFAULT)).thenReturn(emptyConsulInstancesResponse());
+		Flux<ServiceInstance> instances = client.getInstances("nonexistent-service");
+		StepVerifier.create(instances).expectNextCount(0).expectComplete().verify();
+		verify(properties).getAclToken();
+		verify(consulClient).getHealthServices("nonexistent-service", "queryTag", false,
+				QueryParams.DEFAULT);
+	}
+
+	@Test
+	public void shouldReturnEmptyFluxWhenConsulFails() {
+		configureCommonProperties();
+		when(consulClient.getHealthServices("existing-service", "queryTag", false,
+				QueryParams.DEFAULT))
+						.thenThrow(new RuntimeException("Possible runtime exception"));
+		Flux<ServiceInstance> instances = client.getInstances("existing-service");
+		StepVerifier.create(instances).expectNextCount(0).expectComplete().verify();
+		verify(consulClient).getHealthServices("existing-service", "queryTag", false,
+				QueryParams.DEFAULT);
+	}
+
+	@Test
+	public void shouldReturnFluxOfServiceInstances() {
+		configureCommonProperties();
+		Response<List<HealthService>> response = consulInstancesResponse();
+		when(consulClient.getHealthServices("existing-service", "queryTag", false,
+				QueryParams.DEFAULT)).thenReturn(response);
+		Flux<ServiceInstance> instances = client.getInstances("existing-service");
+		StepVerifier.create(instances).expectNextCount(1).expectComplete().verify();
+		verify(properties).getAclToken();
+		verify(properties).getDefaultQueryTag();
+		verify(properties).isQueryPassing();
+		verify(consulClient).getHealthServices("existing-service", "queryTag", false,
+				QueryParams.DEFAULT);
+	}
+
+	@Test
+	public void shouldReturnFluxOfServiceInstancesWithAclToken() {
+		configureCommonProperties();
+		when(properties.getAclToken()).thenReturn("aclToken");
+		Response<List<HealthService>> response = consulInstancesResponse();
+		when(consulClient.getHealthServices("existing-service", "queryTag", false,
+				QueryParams.DEFAULT, "aclToken")).thenReturn(response);
+		Flux<ServiceInstance> instances = client.getInstances("existing-service");
+		StepVerifier.create(instances).expectNextCount(1).expectComplete().verify();
+		verify(properties, times(2)).getAclToken();
+		verify(properties).getDefaultQueryTag();
+		verify(properties).isQueryPassing();
+		verify(consulClient).getHealthServices("existing-service", "queryTag", false,
+				QueryParams.DEFAULT, "aclToken");
+	}
+
+	private Response<Map<String, List<String>>> consulServicesResponse() {
+		return new Response<>(singletonMap("my-service", singletonList("")), 0L, true,
+				System.currentTimeMillis());
+	}
+
+	private void configureCommonProperties() {
+		when(properties.getDefaultQueryTag()).thenReturn("queryTag");
+		when(properties.isQueryPassing()).thenReturn(false);
+	}
+
+	private Response<List<HealthService>> emptyConsulInstancesResponse() {
+		return new Response<>(emptyList(), 0L, true, System.currentTimeMillis());
+	}
+
+	private Response<List<HealthService>> consulInstancesResponse() {
+		HealthService healthService = mock(HealthService.class);
+		HealthService.Service service = mock(HealthService.Service.class);
+
+		when(healthService.getService()).thenReturn(service);
+		when(service.getAddress()).thenReturn("localhost");
+		when(service.getPort()).thenReturn(443);
+		when(service.getTags()).thenReturn(singletonList("secure=true"));
+
+		return new Response<>(singletonList(healthService), 0L, true,
+				System.currentTimeMillis());
+	}
+
+}