Spring/spring-boot
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Polish "Add reactive health indicator for Couchbase"

Browse Source 
Closes gh-13926
This commit is contained in:
Stephane Nicoll
2018-10-05 13:50:59 +02:00
parent 20ff0d97e4
commit 91b4dc2f69
10 changed files with 125 additions and 148 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/couchbase/CouchbaseReactiveHealthIndicator.java
View File

@@ -15,6 +15,9 @@
*/
package org.springframework.boot.actuate.couchbase;
import java.time.Duration;
import java.util.concurrent.TimeUnit;
import com.couchbase.client.java.bucket.BucketInfo;
import com.couchbase.client.java.cluster.ClusterInfo;
import reactor.core.publisher.Mono;
@@ -24,27 +27,32 @@ import rx.Single;
import org.springframework.boot.actuate.health.AbstractReactiveHealthIndicator;
import org.springframework.boot.actuate.health.Health;
import org.springframework.boot.actuate.health.ReactiveHealthIndicator;
import org.springframework.data.couchbase.core.RxJavaCouchbaseOperations;
import org.springframework.util.StringUtils;
/**
* A {@link org.springframework.boot.actuate.health.ReactiveHealthIndicator} for
* Couchbase.
* A {@link ReactiveHealthIndicator} for Couchbase.
*
* @author Mikalai Lushchytski
* @author Stephane Nicoll
* @since 2.1.0
*/
public class CouchbaseReactiveHealthIndicator extends AbstractReactiveHealthIndicator {
private final RxJavaCouchbaseOperations couchbaseOperations;
private final Duration timeout;
/**
* Create a new {@link CouchbaseReactiveHealthIndicator} instance.
* @param couchbaseOperations Reactive couchbase client.
* @param couchbaseOperations the reactive couchbase operations
* @param timeout the request timeout
*/
public CouchbaseReactiveHealthIndicator(
RxJavaCouchbaseOperations couchbaseOperations) {
public CouchbaseReactiveHealthIndicator(RxJavaCouchbaseOperations couchbaseOperations,
Duration timeout) {
this.couchbaseOperations = couchbaseOperations;
this.timeout = timeout;
}
@Override
@@ -53,21 +61,14 @@ public class CouchbaseReactiveHealthIndicator extends AbstractReactiveHealthIndi
String versions = StringUtils
.collectionToCommaDelimitedString(cluster.getAllVersions());
Observable<BucketInfo> bucket = this.couchbaseOperations.getCouchbaseBucket()
.bucketManager().async().info();
.bucketManager().async().info()
.timeout(this.timeout.toMillis(), TimeUnit.MILLISECONDS);
Single<Health> health = bucket.map(BucketInfo::nodeList)
.map(StringUtils::collectionToCommaDelimitedString)
.map((nodes) -> up(builder, versions, nodes))
.onErrorReturn((error) -> down(builder, error)).toSingle();
.map((nodes) -> builder.up().withDetail("versions", versions)
.withDetail("nodes", nodes).build())
.toSingle();
return Mono.from(RxReactiveStreams.toPublisher(health));
}
private Health up(Health.Builder builder, String versions, String nodes) {
return builder.up().withDetail("versions", versions).withDetail("nodes", nodes)
.build();
}
private Health down(Health.Builder builder, Throwable error) {
return builder.down(error).build();
}
}

79
spring-boot-project/spring-boot-actuator/src/test/java/org/springframework/boot/actuate/couchbase/CouchbaseReactiveHealthIndicatorTests.java
View File

@@ -16,8 +16,10 @@
package org.springframework.boot.actuate.couchbase;
import java.net.InetAddress;
import java.time.Duration;
import java.util.Arrays;
import java.util.Collections;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import com.couchbase.client.java.Bucket;
import com.couchbase.client.java.bucket.AsyncBucketManager;
@@ -45,32 +47,20 @@ import static org.mockito.Mockito.mock;
public class CouchbaseReactiveHealthIndicatorTests {
@Test
public void testCouchbaseIsUp() {
ClusterInfo clusterInfo = mock(ClusterInfo.class);
public void couchbaseIsUp() {
RxJavaCouchbaseOperations rxJavaCouchbaseOperations = mock(
RxJavaCouchbaseOperations.class);
given(rxJavaCouchbaseOperations.getCouchbaseClusterInfo())
.willReturn(clusterInfo);
given(clusterInfo.getAllVersions())
.willReturn(Arrays.asList(new Version(5, 5, 0), new Version(6, 0, 0)));
Bucket bucket = mock(Bucket.class);
BucketManager bucketManager = mock(BucketManager.class);
AsyncBucketManager asyncBucketManager = mock(AsyncBucketManager.class);
given(rxJavaCouchbaseOperations.getCouchbaseBucket()).willReturn(bucket);
given(bucket.bucketManager()).willReturn(bucketManager);
given(bucketManager.async()).willReturn(asyncBucketManager);
AsyncBucketManager asyncBucketManager = mockAsyncBucketManager(
rxJavaCouchbaseOperations);
BucketInfo info = mock(BucketInfo.class);
given(asyncBucketManager.info()).willReturn(Observable.just(info));
InetAddress node1Address = mock(InetAddress.class);
InetAddress node2Address = mock(InetAddress.class);
given(info.nodeList()).willReturn(Arrays.asList(node1Address, node2Address));
given(node1Address.toString()).willReturn("127.0.0.1");
given(node2Address.toString()).willReturn("127.0.0.2");
given(asyncBucketManager.info()).willReturn(Observable.just(info));
CouchbaseReactiveHealthIndicator couchbaseReactiveHealthIndicator = new CouchbaseReactiveHealthIndicator(
rxJavaCouchbaseOperations);
rxJavaCouchbaseOperations, Duration.ofSeconds(2));
Mono<Health> health = couchbaseReactiveHealthIndicator.health();
StepVerifier.create(health).consumeNextWith((h) -> {
assertThat(h.getStatus()).isEqualTo(Status.UP);
@@ -81,25 +71,34 @@ public class CouchbaseReactiveHealthIndicatorTests {
}
@Test
public void testCouchbaseIsDown() {
ClusterInfo clusterInfo = mock(ClusterInfo.class);
public void couchbaseTimeout() {
RxJavaCouchbaseOperations rxJavaCouchbaseOperations = mock(
RxJavaCouchbaseOperations.class);
given(rxJavaCouchbaseOperations.getCouchbaseClusterInfo())
.willReturn(clusterInfo);
given(clusterInfo.getAllVersions())
.willReturn(Collections.singletonList(new Version(5, 5, 0)));
BucketManager bucketManager = mock(BucketManager.class);
AsyncBucketManager asyncBucketManager = mock(AsyncBucketManager.class);
Bucket bucket = mock(Bucket.class);
given(rxJavaCouchbaseOperations.getCouchbaseBucket()).willReturn(bucket);
given(bucket.bucketManager()).willReturn(bucketManager);
given(bucketManager.async()).willReturn(asyncBucketManager);
AsyncBucketManager asyncBucketManager = mockAsyncBucketManager(
rxJavaCouchbaseOperations);
given(asyncBucketManager.info()).willReturn(
Observable.just(mock(BucketInfo.class)).delay(20, TimeUnit.MILLISECONDS));
CouchbaseReactiveHealthIndicator couchbaseReactiveHealthIndicator = new CouchbaseReactiveHealthIndicator(
rxJavaCouchbaseOperations, Duration.ofMillis(10));
Mono<Health> health = couchbaseReactiveHealthIndicator.health();
StepVerifier.create(health).consumeNextWith((h) -> {
assertThat(h.getStatus()).isEqualTo(Status.DOWN);
assertThat(h.getDetails()).containsOnlyKeys("error");
assertThat(h.getDetails().get("error")).asString()
.contains(TimeoutException.class.getName());
}).verifyComplete();
}
@Test
public void couchbaseIsDown() {
RxJavaCouchbaseOperations rxJavaCouchbaseOperations = mock(
RxJavaCouchbaseOperations.class);
AsyncBucketManager asyncBucketManager = mockAsyncBucketManager(
rxJavaCouchbaseOperations);
given(asyncBucketManager.info())
.willReturn(Observable.error(new TranscodingException("Failure")));
CouchbaseReactiveHealthIndicator couchbaseReactiveHealthIndicator = new CouchbaseReactiveHealthIndicator(
rxJavaCouchbaseOperations);
rxJavaCouchbaseOperations, Duration.ofSeconds(2));
Mono<Health> health = couchbaseReactiveHealthIndicator.health();
StepVerifier.create(health).consumeNextWith((h) -> {
assertThat(h.getStatus()).isEqualTo(Status.DOWN);
@@ -109,4 +108,20 @@ public class CouchbaseReactiveHealthIndicatorTests {
}).verifyComplete();
}
private AsyncBucketManager mockAsyncBucketManager(
RxJavaCouchbaseOperations rxJavaCouchbaseOperations) {
ClusterInfo clusterInfo = mock(ClusterInfo.class);
given(rxJavaCouchbaseOperations.getCouchbaseClusterInfo())
.willReturn(clusterInfo);
given(clusterInfo.getAllVersions())
.willReturn(Arrays.asList(new Version(5, 5, 0), new Version(6, 0, 0)));
Bucket bucket = mock(Bucket.class);
BucketManager bucketManager = mock(BucketManager.class);
AsyncBucketManager asyncBucketManager = mock(AsyncBucketManager.class);
given(rxJavaCouchbaseOperations.getCouchbaseBucket()).willReturn(bucket);
given(bucket.bucketManager()).willReturn(bucketManager);
given(bucketManager.async()).willReturn(asyncBucketManager);
return asyncBucketManager;
}
}