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

Allow batching the emitted alive instances flux. (#1206)

Browse Source
This commit is contained in:
Olga Maciaszek-Sharma
2023-03-07 11:33:06 +01:00
committed by GitHub
parent dbdb21a199
commit 80643ef3ed
5 changed files with 96 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,5 +1,5 @@
/*
* Copyright 2012-2021 the original author or authors.
* Copyright 2012-2023 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.
@@ -114,10 +114,13 @@ public class HealthCheckServiceInstanceListSupplier extends DelegatingServiceIns
checks.add(alive);
}
List<ServiceInstance> result = new ArrayList<>();
return Flux.merge(checks).map(alive -> {
result.add(alive);
return result;
}).defaultIfEmpty(result);
if (healthCheck.isUpdateResultsList()) {
return Flux.merge(checks).map(alive -> {
result.add(alive);
return result;
}).defaultIfEmpty(result);
}
return Flux.merge(checks).collectList();
}).repeatWhen(healthCheckFluxRepeat);
}

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

@@ -276,6 +276,39 @@ class HealthCheckServiceInstanceListSupplierTests {
.expectNoEvent(properties.getHealthCheck().getInterval()).thenCancel().verify(VERIFY_TIMEOUT);
}
@Test
void shouldEmitOnEntireBatchOfInstancesWhenUpdateDisabled() {
LoadBalancerProperties.HealthCheck healthCheck = properties.getHealthCheck();
healthCheck.setInitialDelay(Duration.ofSeconds(1));
healthCheck.setUpdateResultsList(false);
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);
Mockito.when(delegate.getServiceId()).thenReturn(SERVICE_ID);
Mockito.when(delegate.get()).thenReturn(Flux.just(Lists.list(serviceInstance1, serviceInstance2)));
HealthCheckServiceInstanceListSupplier mock = mock(HealthCheckServiceInstanceListSupplier.class);
Mockito.doReturn(Mono.just(true)).when(mock).isAlive(serviceInstance1);
Mockito.doReturn(Mono.just(true)).when(mock).isAlive(serviceInstance2);
listSupplier = new HealthCheckServiceInstanceListSupplier(delegate,
buildLoadBalancerClientFactory(SERVICE_ID, properties), healthCheckFunction(webClient)) {
@Override
protected Mono<Boolean> isAlive(ServiceInstance serviceInstance) {
return mock.isAlive(serviceInstance);
}
};
return listSupplier.get();
}).expectSubscription().expectNoEvent(properties.getHealthCheck().getInitialDelay())
.expectNext(Lists.list(serviceInstance1, serviceInstance2))
.expectNoEvent(properties.getHealthCheck().getInterval()).thenCancel().verify(VERIFY_TIMEOUT);
}
@Test
void shouldNotFailIfIsAliveReturnsError() {
properties.getHealthCheck().setInitialDelay(Duration.ofSeconds(1));
@@ -458,6 +491,41 @@ class HealthCheckServiceInstanceListSupplierTests {
.expectNext(Lists.list(serviceInstance1, serviceInstance2)).thenCancel().verify(VERIFY_TIMEOUT);
}
@Test
void shouldReturnAllInstancesWhenUpdateDisabled() {
LoadBalancerProperties.HealthCheck healthCheck = properties.getHealthCheck();
healthCheck.setInitialDelay(Duration.ofSeconds(1));
healthCheck.setUpdateResultsList(false);
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);
Mockito.when(delegate.getServiceId()).thenReturn(SERVICE_ID);
Flux<List<ServiceInstance>> instances = Flux.just(Lists.list(serviceInstance1))
.concatWith(Flux.just(Lists.list(serviceInstance1, serviceInstance2))
.delayElements(properties.getHealthCheck().getInterval().dividedBy(2)));
Mockito.when(delegate.get()).thenReturn(instances);
listSupplier = new HealthCheckServiceInstanceListSupplier(delegate,
buildLoadBalancerClientFactory(SERVICE_ID, properties), healthCheckFunction(webClient)) {
@Override
protected Mono<Boolean> isAlive(ServiceInstance serviceInstance) {
return Mono.just(true);
}
};
return listSupplier.get();
}).expectSubscription().expectNoEvent(properties.getHealthCheck().getInitialDelay())
.expectNext(Lists.list(serviceInstance1))
.thenAwait(properties.getHealthCheck().getInterval().dividedBy(2))
.expectNext(Lists.list(serviceInstance1, serviceInstance2))
.expectNoEvent(properties.getHealthCheck().getInterval())
.expectNext(Lists.list(serviceInstance1, serviceInstance2)).thenCancel().verify(VERIFY_TIMEOUT);
}
@Test
void shouldRefetchInstances() {
properties.getHealthCheck().setInitialDelay(Duration.ofSeconds(1));