SpringCloud/spring-cloud-commons
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/2.2.x'

Browse Source 
# Conflicts:
#	spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/core/HealthCheckServiceInstanceListSupplier.java
This commit is contained in:
Olga Maciaszek-Sharma
2021-02-17 15:44:21 +01:00
parent 2e71703df5 95a22796eb
commit 36d5350661
2 changed files with 35 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/core/HealthCheckServiceInstanceListSupplier.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2012-2020 the original author or authors.
* Copyright 2012-2021 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.
@@ -40,6 +40,7 @@ import org.springframework.cloud.client.loadbalancer.LoadBalancerProperties;
*
* @author Olga Maciaszek-Sharma
* @author Roman Matiushchenko
* @author Roman Chigvintsev
* @since 2.2.0
*/
public class HealthCheckServiceInstanceListSupplier extends DelegatingServiceInstanceListSupplier
@@ -67,10 +68,9 @@ public class HealthCheckServiceInstanceListSupplier extends DelegatingServiceIns
Repeat<Object> aliveInstancesReplayRepeat = Repeat
.onlyIf(repeatContext -> this.healthCheck.getRefetchInstances())
.fixedBackoff(healthCheck.getRefetchInstancesInterval());
Flux<List<ServiceInstance>> aliveInstancesFlux = Flux.defer(delegate)
Flux<List<ServiceInstance>> aliveInstancesFlux = Flux.defer(delegate).repeatWhen(aliveInstancesReplayRepeat)
.switchMap(serviceInstances -> healthCheckFlux(serviceInstances)
.map(alive -> Collections.unmodifiableList(new ArrayList<>(alive))))
.repeatWhen(aliveInstancesReplayRepeat);
.map(alive -> Collections.unmodifiableList(new ArrayList<>(alive))));
aliveInstancesReplay = aliveInstancesFlux.delaySubscription(healthCheck.getInitialDelay()).replay(1)
.refCount(1);
}

32
spring-cloud-loadbalancer/src/test/java/org/springframework/cloud/loadbalancer/core/HealthCheckServiceInstanceListSupplierTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2012-2020 the original author or authors.
* Copyright 2012-2021 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.
@@ -21,6 +21,7 @@ import java.util.Collections;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiFunction;
import org.assertj.core.api.Assertions;
import org.assertj.core.util.Lists;
@@ -59,6 +60,7 @@ import static org.springframework.cloud.loadbalancer.core.ServiceInstanceListSup
*
* @author Olga Maciaszek-Sharma
* @author Roman Matiushchenko
* @author Roman Chigvintsev
*/
@ExtendWith(SpringExtension.class)
@SpringBootTest(classes = HealthCheckServiceInstanceListSupplierTests.TestApplication.class,
@@ -440,6 +442,34 @@ class HealthCheckServiceInstanceListSupplierTests {
.thenCancel().verify(VERIFY_TIMEOUT);
}
@Test
void shouldRefetchInstancesWithRepeatingHealthCheck() {
healthCheck.setInitialDelay(Duration.ofSeconds(1));
healthCheck.setRepeatHealthCheck(true);
healthCheck.setRefetchInstancesInterval(Duration.ofSeconds(1));
healthCheck.setRefetchInstances(true);
ServiceInstance serviceInstance1 = new DefaultServiceInstance("ignored-service-1", SERVICE_ID, "127.0.0.1",
port, false);
ServiceInstance serviceInstance2 = new DefaultServiceInstance("ignored-service-2", SERVICE_ID, "127.0.0.2",
port, false);
StepVerifier.withVirtualTime(() -> {
ServiceInstanceListSupplier delegate = mock(ServiceInstanceListSupplier.class);
when(delegate.get()).thenReturn(Flux.just(Collections.singletonList(serviceInstance1)))
.thenReturn(Flux.just(Collections.singletonList(serviceInstance2)));
BiFunction<ServiceInstance, String, Mono<Boolean>> healthCheckFunc = healthCheckFunction(webClient);
listSupplier = new HealthCheckServiceInstanceListSupplier(delegate, healthCheck, healthCheckFunc) {
@Override
protected Mono<Boolean> isAlive(ServiceInstance serviceInstance) {
return Mono.just(true);
}
};
return listSupplier.get();
}).expectSubscription().expectNoEvent(healthCheck.getInitialDelay()).expectNext(Lists.list(serviceInstance1))
.thenAwait(healthCheck.getRefetchInstancesInterval()).expectNext(Lists.list(serviceInstance2))
.thenCancel().verify(VERIFY_TIMEOUT);
}
@Test
void shouldCacheResultIfAfterPropertiesSetInvoked() {
healthCheck.setInitialDelay(Duration.ofSeconds(1));