HealthCheckedEndpointGroup extends DynamicEndpointGroup { public static HealthCheckedEndpointGroup of(EndpointGroup delegate, String path) { return builder(delegate, path).build(); } HealthCheckedEndpointGroup(
            EndpointGroup delegate, SessionProtocol protocol, int port,
            Backoff retryBackoff, ClientOptions clientOptions,
            Function<? super HealthCheckerContext, ? extends AsyncCloseable> checkerFactory,
            HealthCheckStrategy healthCheckStrategy); static HealthCheckedEndpointGroup of(EndpointGroup delegate, String path); static HealthCheckedEndpointGroupBuilder builder(EndpointGroup delegate, String path); MeterBinder newMeterBinder(String groupName); MeterBinder newMeterBinder(MeterIdPrefix idPrefix); @Override String toString();  }
@Test void delegateUpdateCandidatesWhileCreatingHealthCheckedEndpointGroup() { final MockEndpointGroup delegate = new MockEndpointGroup(); final CompletableFuture<List<Endpoint>> future = delegate.whenReady(); future.complete(ImmutableList.of(Endpoint.of("127.0.0.1", 8080), Endpoint.of("127.0.0.1", 8081))); final CountDownLatch latch = new CountDownLatch(1); final EventLoopGroup executors = CommonPools.workerGroup(); executors.schedule( () -> { delegate.set(Endpoint.of("127.0.0.1", 8082)); latch.countDown(); }, 1, TimeUnit.SECONDS); new AbstractHealthCheckedEndpointGroupBuilder(delegate) { @Override protected Function<? super HealthCheckerContext, ? extends AsyncCloseable> newCheckerFactory() { return (Function<HealthCheckerContext, AsyncCloseable>) ctx -> { new Thread(() -> { try { Thread.sleep(2000); latch.await(); } catch (InterruptedException e) { } ctx.updateHealth(1); }).start(); return AsyncCloseableSupport.of(); }; } }.build(); }
@Test void startsUnhealthyAwaitsForEmptyEndpoints() throws Exception { final MockEndpointGroup delegate = new MockEndpointGroup(); delegate.set(Endpoint.of("foo")); final AtomicReference<HealthCheckerContext> ctxCapture = new AtomicReference<>(); try (HealthCheckedEndpointGroup group = new AbstractHealthCheckedEndpointGroupBuilder(delegate) { @Override protected Function<? super HealthCheckerContext, ? extends AsyncCloseable> newCheckerFactory() { return ctx -> { ctxCapture.set(ctx); ctx.updateHealth(0); return AsyncCloseableSupport.of(); }; } }.build()) { assertThat(group.whenReady().get(10, TimeUnit.SECONDS)).isEmpty(); } }
@Test void disappearedEndpoint() { final MockEndpointGroup delegate = new MockEndpointGroup(); delegate.set(Endpoint.of("foo")); final AtomicReference<HealthCheckerContext> ctxCapture = new AtomicReference<>(); try (HealthCheckedEndpointGroup group = new AbstractHealthCheckedEndpointGroupBuilder(delegate) { @Override protected Function<? super HealthCheckerContext, ? extends AsyncCloseable> newCheckerFactory() { return ctx -> { ctxCapture.set(ctx); ctx.updateHealth(1); return AsyncCloseableSupport.of(); }; } }.build()) { final HealthCheckerContext ctx = ctxCapture.get(); assertThat(ctx).isNotNull(); assertThat(ctx.endpoint()).isEqualTo(Endpoint.of("foo")); final AtomicBoolean taskRun = new AtomicBoolean(); ctx.executor().execute(() -> taskRun.set(true)); await().untilAsserted(() -> assertThat(taskRun).isTrue()); delegate.set(); assertThat(group.endpoints()).isEmpty(); ctx.updateHealth(1); assertThat(group.endpoints()).isEmpty(); assertThat(group.healthyEndpoints).isEmpty(); assertThatThrownBy(() -> ctx.executor().execute(() -> {})) .isInstanceOf(RejectedExecutionException.class) .hasMessageContaining("destroyed"); } }
@Test void updatesSelectedCandidatesNoStackOverflowEvenUpdatesOnEqualThread() { final AtomicReference<HealthCheckerContext> firstSelectedCandidates = new AtomicReference<>(); final Function<? super HealthCheckerContext, ? extends AsyncCloseable> checkFactory = ctx -> { if (firstSelectedCandidates.get() == null) { firstSelectedCandidates.set(ctx); } ctx.updateHealth(HEALTHY); return AsyncCloseableSupport.of(); }; final Endpoint candidate1 = Endpoint.of("candidate1"); final Endpoint candidate2 = Endpoint.of("candidate2"); final MockEndpointGroup delegate = new MockEndpointGroup(); delegate.set(candidate1, candidate2); try (HealthCheckedEndpointGroup group = new HealthCheckedEndpointGroup(delegate, SessionProtocol.HTTP, 80, DEFAULT_HEALTH_CHECK_RETRY_BACKOFF, ClientOptions.of(), checkFactory, new InfinityUpdateHealthCheckStrategy())) { assertThat(group.healthyEndpoints).containsOnly(candidate1, candidate2); firstSelectedCandidates.get().updateHealth(UNHEALTHY); assertThat(group.healthyEndpoints).containsOnly(candidate2); } }