AbstractEndpointSelector implements EndpointSelector { @Override public final CompletableFuture<Endpoint> select(ClientRequestContext ctx, ScheduledExecutorService executor, long timeoutMillis) { Endpoint endpoint = selectNow(ctx); if (endpoint != null) { return UnmodifiableFuture.completedFuture(endpoint); } final ListeningFuture listeningFuture = new ListeningFuture(ctx, executor); endpointGroup.addListener(listeningFuture); endpoint = selectNow(ctx); if (endpoint != null) { endpointGroup.removeListener(listeningFuture); return UnmodifiableFuture.completedFuture(endpoint); } final ScheduledFuture<?> timeoutFuture = executor.schedule(() -> listeningFuture.complete(null), timeoutMillis, TimeUnit.MILLISECONDS); listeningFuture.timeoutFuture = timeoutFuture; if (listeningFuture.isDone()) { timeoutFuture.cancel(false); } return listeningFuture; } protected  AbstractEndpointSelector(EndpointGroup endpointGroup); @Override final CompletableFuture<Endpoint> select(ClientRequestContext ctx,
                                                    ScheduledExecutorService executor,
                                                    long timeoutMillis);  }
@Test void immediateSelection() { final Endpoint endpoint = Endpoint.of("foo"); final ClientRequestContext ctx = ClientRequestContext.of(HttpRequest.of(HttpMethod.GET, "/")); assertThat(newSelector(endpoint).select(ctx, ctx.eventLoop(), Long.MAX_VALUE)) .isCompletedWithValue(endpoint); }
@Test void delayedSelection() { final DynamicEndpointGroup group = new DynamicEndpointGroup(); final ClientRequestContext ctx = ClientRequestContext.of(HttpRequest.of(HttpMethod.GET, "/")); final CompletableFuture<Endpoint> future = newSelector(group).select(ctx, ctx.eventLoop(), Long.MAX_VALUE); assertThat(future).isNotDone(); final Endpoint endpoint = Endpoint.of("foo"); group.addEndpoint(endpoint); assertThat(future.join()).isSameAs(endpoint); }
@Test void timeout() { final DynamicEndpointGroup group = new DynamicEndpointGroup(); final ClientRequestContext ctx = ClientRequestContext.of(HttpRequest.of(HttpMethod.GET, "/")); final CompletableFuture<Endpoint> future = newSelector(group).select(ctx, ctx.eventLoop(), 1000) .handle((res, cause) -> { assertThat(ctx.eventLoop().inEventLoop()).isTrue(); if (cause != null) { Exceptions.throwUnsafely(cause); } return res; }); assertThat(future).isNotDone(); final Stopwatch stopwatch = Stopwatch.createStarted(); assertThat(future.join()).isNull(); assertThat(stopwatch.elapsed(TimeUnit.MILLISECONDS)).isGreaterThan(900); }