@Test public void getUnresolvedFutureUntilSingleIterationHasRun() { CumulativeLeaderPinger cumulativeLeaderPinger = new CumulativeLeaderPinger( pingerWithContext, Duration.ofSeconds(1), Duration.ofSeconds(1), HOST_UUID); Future<LeaderPingResult> clientWhoIsLed = cumulativeLeaderPinger.registerAndPing(HOST_UUID, CLIENT_WHO_IS_LED); assertThat(clientWhoIsLed) .as("it's hard to test that it never gets resolved prior to running the single iteration") .isNotDone(); when(batchPingableLeader.ping(ImmutableSet.of(CLIENT_WHO_IS_LED))) .thenReturn(ImmutableSet.of(CLIENT_WHO_IS_LED)); cumulativeLeaderPinger.runOneIteration(); assertThat(clientWhoIsLed) .as("we expect this to be done, everything is running within a single thread in this test") .isDone(); }
@Override public Future<LeaderPingResult> registerAndPing(UUID requestedUuid, Client client) { Instant requestTime = Instant.now(); return FluentFuture.from(hasProcessedFirstRequest.computeIfAbsent(client, $ -> SettableFuture.create())) .transform( $ -> { checkNotShutdown(); return lastSuccessfulResult.get().result( client, requestTime.minus(leaderPingResponseWait), remoteClient.hostAndPort(), requestedUuid); }
CumulativeLeaderPinger extends AbstractScheduledService implements ClientAwareLeaderPinger { @Override public Future<LeaderPingResult> registerAndPing(UUID requestedUuid, Client client) { Instant requestTime = Instant.now(); return FluentFuture.from(hasProcessedFirstRequest.computeIfAbsent(client, $ -> SettableFuture.create())) .transform( $ -> { checkNotShutdown(); return lastSuccessfulResult.get().result( client, requestTime.minus(leaderPingResponseWait), remoteClient.hostAndPort(), requestedUuid); } }
CumulativeLeaderPinger extends AbstractScheduledService implements ClientAwareLeaderPinger { @Override public Future<LeaderPingResult> registerAndPing(UUID requestedUuid, Client client) { Instant requestTime = Instant.now(); return FluentFuture.from(hasProcessedFirstRequest.computeIfAbsent(client, $ -> SettableFuture.create())) .transform( $ -> { checkNotShutdown(); return lastSuccessfulResult.get().result( client, requestTime.minus(leaderPingResponseWait), remoteClient.hostAndPort(), requestedUuid); } CumulativeLeaderPinger(
            LeaderPingerContext<BatchPingableLeader> remoteClient,
            Duration leaderPingRate,
            Duration leaderPingResponseWait,
            UUID nodeUuid); }
CumulativeLeaderPinger extends AbstractScheduledService implements ClientAwareLeaderPinger { @Override public Future<LeaderPingResult> registerAndPing(UUID requestedUuid, Client client) { Instant requestTime = Instant.now(); return FluentFuture.from(hasProcessedFirstRequest.computeIfAbsent(client, $ -> SettableFuture.create())) .transform( $ -> { checkNotShutdown(); return lastSuccessfulResult.get().result( client, requestTime.minus(leaderPingResponseWait), remoteClient.hostAndPort(), requestedUuid); } CumulativeLeaderPinger(
            LeaderPingerContext<BatchPingableLeader> remoteClient,
            Duration leaderPingRate,
            Duration leaderPingResponseWait,
            UUID nodeUuid); @Override Future<LeaderPingResult> registerAndPing(UUID requestedUuid, Client client); }
CumulativeLeaderPinger extends AbstractScheduledService implements ClientAwareLeaderPinger { @Override public Future<LeaderPingResult> registerAndPing(UUID requestedUuid, Client client) { Instant requestTime = Instant.now(); return FluentFuture.from(hasProcessedFirstRequest.computeIfAbsent(client, $ -> SettableFuture.create())) .transform( $ -> { checkNotShutdown(); return lastSuccessfulResult.get().result( client, requestTime.minus(leaderPingResponseWait), remoteClient.hostAndPort(), requestedUuid); } CumulativeLeaderPinger(
            LeaderPingerContext<BatchPingableLeader> remoteClient,
            Duration leaderPingRate,
            Duration leaderPingResponseWait,
            UUID nodeUuid); @Override Future<LeaderPingResult> registerAndPing(UUID requestedUuid, Client client);  }