Skip to content

Commit ba9a10c

Browse files
wind57wind57
committed
wip
Signed-off-by: wind57 <[email protected]>
1 parent cc511dd commit ba9a10c

File tree

5 files changed

+159
-29
lines changed

5 files changed

+159
-29
lines changed

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

Lines changed: 37 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,9 +16,45 @@
1616

1717
package org.springframework.cloud.kubernetes.discovery;
1818

19+
import reactor.core.publisher.Flux;
20+
21+
import org.springframework.cloud.client.ServiceInstance;
22+
import org.springframework.cloud.client.discovery.ReactiveDiscoveryClient;
23+
import org.springframework.cloud.kubernetes.commons.discovery.DefaultKubernetesServiceInstance;
24+
import org.springframework.cloud.kubernetes.commons.discovery.KubernetesDiscoveryProperties;
25+
import org.springframework.cloud.kubernetes.commons.discovery.Service;
26+
import org.springframework.util.StringUtils;
27+
import org.springframework.web.reactive.function.client.WebClient;
28+
1929
/**
2030
* @author wind57
2131
*/
22-
abstract class KubernetesAbstractReactiveDiscoveryClient {
32+
abstract class KubernetesAbstractReactiveDiscoveryClient implements ReactiveDiscoveryClient {
33+
34+
private final WebClient webClient;
35+
36+
KubernetesAbstractReactiveDiscoveryClient(WebClient.Builder webClientBuilder, KubernetesDiscoveryProperties properties) {
37+
if (!StringUtils.hasText(properties.discoveryServerUrl())) {
38+
throw new DiscoveryServerUrlInvalidException();
39+
}
40+
webClient = webClientBuilder.baseUrl(properties.discoveryServerUrl()).build();
41+
}
42+
43+
@Override
44+
public Flux<String> getServices() {
45+
return webClient.get()
46+
.uri("/apps")
47+
.exchangeToFlux(clientResponse -> clientResponse.bodyToFlux(Service.class).map(Service::name));
48+
}
49+
50+
@Override
51+
public Flux<ServiceInstance> getInstances(String serviceId) {
52+
return webClient.get()
53+
.uri("/apps/" + serviceId)
54+
.exchangeToFlux(clientResponse -> clientResponse.bodyToFlux(DefaultKubernetesServiceInstance.class));
55+
}
56+
57+
@Override
58+
public abstract String description();
2359

2460
}

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

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,52 @@
1+
/*
2+
* Copyright 2013-present the original author or authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package org.springframework.cloud.kubernetes.discovery;
18+
19+
import org.springframework.cache.annotation.Cacheable;
20+
import reactor.core.publisher.Flux;
21+
22+
import org.springframework.cloud.client.ServiceInstance;
23+
import org.springframework.cloud.kubernetes.commons.discovery.KubernetesDiscoveryProperties;
24+
import org.springframework.web.reactive.function.client.WebClient;
25+
26+
/**
27+
* @author wind57
28+
*/
29+
class KubernetesCacheableReactiveDiscoveryClient extends KubernetesAbstractReactiveDiscoveryClient {
30+
31+
KubernetesCacheableReactiveDiscoveryClient(WebClient.Builder webClientBuilder,
32+
KubernetesDiscoveryProperties properties) {
33+
super(webClientBuilder, properties);
34+
}
35+
36+
@Override
37+
@Cacheable("k8s-reactive-discovery-services")
38+
public Flux<String> getServices() {
39+
return super.getServices();
40+
}
41+
42+
@Override
43+
@Cacheable("k8s-reactive-discovery-instances")
44+
public Flux<ServiceInstance> getInstances(String serviceId) {
45+
return super.getInstances(serviceId);
46+
}
47+
48+
@Override
49+
public String description() {
50+
return "Reactive Cacheable Kubernetes Discovery Client";
51+
}
52+
}

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

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,8 @@
2323
import org.springframework.cloud.kubernetes.commons.PodUtils;
2424
import org.springframework.cloud.kubernetes.commons.discovery.KubernetesDiscoveryClientHealthIndicatorInitializer;
2525
import org.springframework.cloud.kubernetes.commons.discovery.KubernetesDiscoveryProperties;
26+
import org.springframework.cloud.kubernetes.commons.discovery.conditionals.ConditionalOnDiscoveryCacheableReactiveDisabled;
27+
import org.springframework.cloud.kubernetes.commons.discovery.conditionals.ConditionalOnDiscoveryCacheableReactiveEnabled;
2628
import org.springframework.cloud.kubernetes.commons.discovery.conditionals.ConditionalOnSpringCloudKubernetesReactiveDiscovery;
2729
import org.springframework.cloud.kubernetes.commons.discovery.conditionals.ConditionalOnSpringCloudKubernetesReactiveDiscoveryHealthInitializer;
2830
import org.springframework.context.ApplicationEventPublisher;
@@ -46,11 +48,21 @@ WebClient.Builder webClientBuilder() {
4648

4749
@Bean
4850
@ConditionalOnMissingBean
51+
@ConditionalOnDiscoveryCacheableReactiveDisabled
4952
KubernetesReactiveDiscoveryClient kubernetesReactiveDiscoveryClient(WebClient.Builder webClientBuilder,
5053
KubernetesDiscoveryProperties properties) {
5154
return new KubernetesReactiveDiscoveryClient(webClientBuilder, properties);
5255
}
5356

57+
@Bean
58+
@ConditionalOnMissingBean
59+
@ConditionalOnDiscoveryCacheableReactiveEnabled
60+
KubernetesCacheableReactiveDiscoveryClient kubernetesCacheableReactiveDiscoveryClient(
61+
WebClient.Builder webClientBuilder,
62+
KubernetesDiscoveryProperties properties) {
63+
return new KubernetesCacheableReactiveDiscoveryClient(webClientBuilder, properties);
64+
}
65+
5466
@Bean
5567
@ConditionalOnMissingBean
5668
PodUtils<?> kubernetesDiscoveryPodUtils() {

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

Lines changed: 2 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -16,47 +16,21 @@
1616

1717
package org.springframework.cloud.kubernetes.discovery;
1818

19-
import reactor.core.publisher.Flux;
20-
21-
import org.springframework.cloud.client.ServiceInstance;
22-
import org.springframework.cloud.client.discovery.ReactiveDiscoveryClient;
23-
import org.springframework.cloud.kubernetes.commons.discovery.DefaultKubernetesServiceInstance;
2419
import org.springframework.cloud.kubernetes.commons.discovery.KubernetesDiscoveryProperties;
25-
import org.springframework.cloud.kubernetes.commons.discovery.Service;
26-
import org.springframework.util.StringUtils;
2720
import org.springframework.web.reactive.function.client.WebClient;
2821

2922
/**
3023
* @author Ryan Baxter
3124
*/
32-
public class KubernetesReactiveDiscoveryClient implements ReactiveDiscoveryClient {
33-
34-
private final WebClient webClient;
25+
final class KubernetesReactiveDiscoveryClient extends KubernetesAbstractReactiveDiscoveryClient {
3526

3627
KubernetesReactiveDiscoveryClient(WebClient.Builder webClientBuilder, KubernetesDiscoveryProperties properties) {
37-
if (!StringUtils.hasText(properties.discoveryServerUrl())) {
38-
throw new DiscoveryServerUrlInvalidException();
39-
}
40-
webClient = webClientBuilder.baseUrl(properties.discoveryServerUrl()).build();
28+
super(webClientBuilder, properties);
4129
}
4230

4331
@Override
4432
public String description() {
4533
return "Reactive Kubernetes Discovery Client";
4634
}
4735

48-
@Override
49-
public Flux<ServiceInstance> getInstances(String serviceId) {
50-
return webClient.get()
51-
.uri("/apps/" + serviceId)
52-
.exchangeToFlux(clientResponse -> clientResponse.bodyToFlux(DefaultKubernetesServiceInstance.class));
53-
}
54-
55-
@Override
56-
public Flux<String> getServices() {
57-
return webClient.get()
58-
.uri("/apps")
59-
.exchangeToFlux(clientResponse -> clientResponse.bodyToFlux(Service.class).map(Service::name));
60-
}
61-
6236
}

‎spring-cloud-kubernetes-discovery/src/test/java/org/springframework/cloud/kubernetes/discovery/KubernetesDiscoveryAutoConfigurationTests.java‎

Lines changed: 56 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -322,6 +322,62 @@ void blockingCacheableEnabled() {
322322
});
323323
}
324324

325+
/**
326+
* <pre>
327+
* - no property related to cacheable in the reactive implementation is set, as such:
328+
* - KubernetesReactiveDiscoveryClient is present
329+
* - KubernetesCacheableReactiveDiscoveryClient is not present
330+
* </pre>
331+
*/
332+
@Test
333+
void reactiveCacheableDefault() {
334+
setupWithFilteredClassLoader(null,
335+
"spring.main.cloud-platform=KUBERNETES", "spring.cloud.config.enabled=false",
336+
"spring.cloud.kubernetes.discovery.discovery-server-url=http://k8sdiscoveryserver");
337+
applicationContextRunner.run(context -> {
338+
assertThat(context).hasSingleBean(KubernetesReactiveDiscoveryClient.class);
339+
assertThat(context).doesNotHaveBean(KubernetesCacheableReactiveDiscoveryClient.class);
340+
});
341+
}
342+
343+
/**
344+
* <pre>
345+
* - cacheable in the reactive implementation = false, as such:
346+
* - KubernetesReactiveDiscoveryClient is present
347+
* - KubernetesCacheableReactiveDiscoveryClient is not present
348+
* </pre>
349+
*/
350+
@Test
351+
void reactiveCacheableDisabled() {
352+
setupWithFilteredClassLoader(null,
353+
"spring.main.cloud-platform=KUBERNETES", "spring.cloud.config.enabled=false",
354+
"spring.cloud.kubernetes.discovery.cacheable.reactive.enabled=false",
355+
"spring.cloud.kubernetes.discovery.discovery-server-url=http://k8sdiscoveryserver");
356+
applicationContextRunner.run(context -> {
357+
assertThat(context).hasSingleBean(KubernetesReactiveDiscoveryClient.class);
358+
assertThat(context).doesNotHaveBean(KubernetesCacheableReactiveDiscoveryClient.class);
359+
});
360+
}
361+
362+
/**
363+
* <pre>
364+
* - cacheable in the reactive implementation = true, as such:
365+
* - KubernetesReactiveDiscoveryClient is not present
366+
* - KubernetesCacheableReactiveDiscoveryClient is present
367+
* </pre>
368+
*/
369+
@Test
370+
void reactiveCacheableEnabled() {
371+
setupWithFilteredClassLoader(null,
372+
"spring.main.cloud-platform=KUBERNETES", "spring.cloud.config.enabled=false",
373+
"spring.cloud.kubernetes.discovery.cacheable.reactive.enabled=true",
374+
"spring.cloud.kubernetes.discovery.discovery-server-url=http://k8sdiscoveryserver");
375+
applicationContextRunner.run(context -> {
376+
assertThat(context).doesNotHaveBean(KubernetesReactiveDiscoveryClient.class);
377+
assertThat(context).hasSingleBean(KubernetesCacheableReactiveDiscoveryClient.class);
378+
});
379+
}
380+
325381
private ApplicationContextRunner applicationContextRunner;
326382

327383
private void setupWithFilteredClassLoader(Class<?> cls, String... properties) {

0 commit comments

Comments
 (0)