ThrottledAsyncChecker implements AsyncChecker<K, V> { @Override public synchronized Optional<ListenableFuture<V>> schedule( Checkable<K, V> target, K context) { if (checksInProgress.containsKey(target)) { return Optional.empty(); } final LastCheckResult<V> result = completedChecks.get(target); if (result != null) { final long msSinceLastCheck = timer.monotonicNow() - result.completedAt; if (msSinceLastCheck < minMsBetweenChecks) { LOG.debug("Skipped checking {}. Time since last check {}ms " + "is less than the min gap {}ms.", target, msSinceLastCheck, minMsBetweenChecks); return Optional.empty(); } } LOG.info("Scheduling a check for {}", target); final ListenableFuture<V> lfWithoutTimeout = executorService.submit( new Callable<V>() { @Override public V call() throws Exception { return target.check(context); } }); final ListenableFuture<V> lf; if (diskCheckTimeout > 0) { lf = TimeoutFuture .create(lfWithoutTimeout, diskCheckTimeout, TimeUnit.MILLISECONDS, scheduledExecutorService); } else { lf = lfWithoutTimeout; } checksInProgress.put(target, lf); addResultCachingCallback(target, lf); return Optional.of(lf); } ThrottledAsyncChecker(final Timer timer,
                        final long minMsBetweenChecks,
                        final long diskCheckTimeout,
                        final ExecutorService executorService); @Override synchronized Optional<ListenableFuture<V>> schedule(
      Checkable<K, V> target, K context); @Override void shutdownAndWait(long timeout, TimeUnit timeUnit); static final Logger LOG; }
@Test(timeout=60000) public void testScheduler() throws Exception { final NoOpCheckable target1 = new NoOpCheckable(); final NoOpCheckable target2 = new NoOpCheckable(); final FakeTimer timer = new FakeTimer(); ThrottledAsyncChecker<Boolean, Boolean> checker = new ThrottledAsyncChecker<>(timer, MIN_ERROR_CHECK_GAP, 0, getExecutorService()); assertTrue(checker.schedule(target1, true).isPresent()); waitTestCheckableCheckCount(target1, 1L); assertFalse(checker.schedule(target1, true).isPresent()); waitTestCheckableCheckCount(target1, 1L); assertTrue(checker.schedule(target2, true).isPresent()); waitTestCheckableCheckCount(target2, 1L); timer.advance(MIN_ERROR_CHECK_GAP - 1); assertFalse(checker.schedule(target1, true).isPresent()); waitTestCheckableCheckCount(target1, 1L); assertFalse(checker.schedule(target2, true).isPresent()); waitTestCheckableCheckCount(target2, 1L); timer.advance(MIN_ERROR_CHECK_GAP); assertTrue(checker.schedule(target1, true).isPresent()); waitTestCheckableCheckCount(target1, 2L); assertTrue(checker.schedule(target2, true).isPresent()); waitTestCheckableCheckCount(target2, 2L); }
@Test (timeout=60000) public void testConcurrentChecks() throws Exception { final StalledCheckable target = new StalledCheckable(); final FakeTimer timer = new FakeTimer(); ThrottledAsyncChecker<Boolean, Boolean> checker = new ThrottledAsyncChecker<>(timer, MIN_ERROR_CHECK_GAP, 0, getExecutorService()); final Optional<ListenableFuture<Boolean>> olf1 = checker.schedule(target, true); final Optional<ListenableFuture<Boolean>> olf2 = checker.schedule(target, true); assertTrue(olf1.isPresent()); assertFalse(olf2.isPresent()); }
@Test(timeout=60000) public void testContextIsPassed() throws Exception { final NoOpCheckable target1 = new NoOpCheckable(); final FakeTimer timer = new FakeTimer(); ThrottledAsyncChecker<Boolean, Boolean> checker = new ThrottledAsyncChecker<>(timer, MIN_ERROR_CHECK_GAP, 0, getExecutorService()); assertTrue(checker.schedule(target1, true).isPresent()); waitTestCheckableCheckCount(target1, 1L); timer.advance(MIN_ERROR_CHECK_GAP + 1); assertTrue(checker.schedule(target1, false).isPresent()); waitTestCheckableCheckCount(target1, 2L); }
@Test(timeout=60000) public void testExceptionIsPropagated() throws Exception { final ThrowingCheckable target = new ThrowingCheckable(); final FakeTimer timer = new FakeTimer(); ThrottledAsyncChecker<Boolean, Boolean> checker = new ThrottledAsyncChecker<>(timer, MIN_ERROR_CHECK_GAP, 0, getExecutorService()); final Optional<ListenableFuture<Boolean>> olf = checker.schedule(target, true); assertTrue(olf.isPresent()); try { olf.get().get(); fail("Failed to get expected ExecutionException"); } catch(ExecutionException ee) { assertTrue(ee.getCause() instanceof DummyException); } }
@Test(timeout=60000) public void testExceptionCaching() throws Exception { final ThrowingCheckable target1 = new ThrowingCheckable(); final FakeTimer timer = new FakeTimer(); ThrottledAsyncChecker<Boolean, Boolean> checker = new ThrottledAsyncChecker<>(timer, MIN_ERROR_CHECK_GAP, 0, getExecutorService()); assertTrue(checker.schedule(target1, true).isPresent()); waitTestCheckableCheckCount(target1, 1L); assertFalse(checker.schedule(target1, true).isPresent()); waitTestCheckableCheckCount(target1, 1L); }