### Question:
ClusterManagerBucketRefresher implements BucketRefresher { @Override public synchronized Mono<Void> register(final String name) { if (registrations.containsKey(name)) { return Mono.empty(); } return Mono.defer(() -> { Disposable registration = registerStream(core.context(), name); registrations.put(name, registration); return Mono.empty(); }); } ClusterManagerBucketRefresher(final ConfigurationProvider provider, final Core core); @Override synchronized Mono<Void> register(final String name); @Override synchronized Mono<Void> deregister(final String name); @Override synchronized Mono<Void> shutdown(); @Override void markTainted(String name); @Override void markUntainted(String name);  }### Answer:
@Test void shouldReconnectIfStreamCloses() { final AtomicReference<BucketConfigStreamingResponse> responseRef = new AtomicReference<>(); final AtomicInteger streamingRequestAttempts = new AtomicInteger(); doAnswer(i -> { streamingRequestAttempts.incrementAndGet(); BucketConfigStreamingRequest request = i.getArgument(0); HttpResponse httpResponse = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK); BucketConfigStreamingResponse response = request.decode(httpResponse, null); responseRef.set(response); request.succeed(response); return null; }).when(core).send(any(BucketConfigStreamingRequest.class)); refresher.register("bucketName").block(); responseRef.get().completeStream(); waitUntilCondition(() -> streamingRequestAttempts.get() >= 2); }
@Test void shouldReconnectIfStreamErrors() { final AtomicReference<BucketConfigStreamingResponse> responseRef = new AtomicReference<>(); final AtomicInteger streamingRequestAttempts = new AtomicInteger(); doAnswer(i -> { streamingRequestAttempts.incrementAndGet(); BucketConfigStreamingRequest request = i.getArgument(0); HttpResponse httpResponse = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK); BucketConfigStreamingResponse response = request.decode(httpResponse, null); responseRef.set(response); request.succeed(response); return null; }).when(core).send(any(BucketConfigStreamingRequest.class)); refresher.register("bucketName").block(); responseRef.get().failStream(new RuntimeException("Something Happened")); waitUntilCondition(() -> streamingRequestAttempts.get() >= 2); }
@Test void shouldKeepTryingIfHttpResponseHasNonSuccessStatus() { final AtomicInteger streamingRequestAttempts = new AtomicInteger(); doAnswer(i -> { streamingRequestAttempts.incrementAndGet(); BucketConfigStreamingRequest request = i.getArgument(0); HttpResponse httpResponse = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.INTERNAL_SERVER_ERROR); BucketConfigStreamingResponse response = request.decode(httpResponse, null); request.succeed(response); return null; }).when(core).send(any(BucketConfigStreamingRequest.class)); refresher.register("bucketName").block(); waitUntilCondition(() -> streamingRequestAttempts.get() >= 2); }
@Test void shouldKeepTryingIfRequestFailsCompletely() { final AtomicInteger streamingRequestAttempts = new AtomicInteger(); doAnswer(i -> { streamingRequestAttempts.incrementAndGet(); BucketConfigStreamingRequest request = i.getArgument(0); request.fail(new RuntimeException()); return null; }).when(core).send(any(BucketConfigStreamingRequest.class)); refresher.register("bucketName").block(); waitUntilCondition(() -> streamingRequestAttempts.get() >= 2); }