AsyncLoadingCache { @SuppressWarnings("squid:S2789") @Nonnull public CompletableFuture<V> orElseGet(@Nonnull K key, @Nonnull Supplier<CompletableFuture<V>> supplier) { try { Optional<V> cachedValue = cache.getIfPresent(key); if (cachedValue == null) { return MoreAsyncUtil.getWithDeadline(deadlineTimeMillis, supplier).thenApply(value -> { final Optional<V> existingValue = cache.asMap().putIfAbsent(key, Optional.ofNullable(value)); return existingValue == null ? value : existingValue.orElse(null); }); } else { return CompletableFuture.completedFuture(cachedValue.orElse(null)); } } catch (Exception e) { throw new RecordCoreException("failed getting value", e).addLogInfo("cacheKey", key); } } AsyncLoadingCache(long refreshTimeMillis);  AsyncLoadingCache(long refreshTimeMillis, long deadlineTimeMillis);  AsyncLoadingCache(long refreshTimeMillis, long deadlineTimeMillis, long maxSize); @SuppressWarnings("squid:S2789") // comparison of null and optional used to differentiate absence of key and presence of null @Nonnull CompletableFuture<V> orElseGet(@Nonnull K key, @Nonnull Supplier<CompletableFuture<V>> supplier); long getRefreshTimeSeconds(); long getMaxSize(); void clear(); @Override String toString(); static final long DEFAULT_DEADLINE_TIME_MILLIS; static final long UNLIMITED; }
@Test public void testRefreshTime() { AsyncLoadingCache<String, Boolean> cachedResult = new AsyncLoadingCache<>(100); final AtomicBoolean called = new AtomicBoolean(false); Supplier<CompletableFuture<Boolean>> supplier = () -> CompletableFuture.completedFuture(called.getAndSet(true)); consistently("we see the cached value up until the expiration", () -> cachedResult.orElseGet("a", supplier).join(), is(false), 75, 10); eventually("the cached result expires and we start reading true", () -> cachedResult.orElseGet("a", supplier).join(), is(true), 100, 10); consistently("any subsequent reads are true", () -> cachedResult.orElseGet("a", supplier).join(), is(true), 50, 10); }
@Test public void testSupplierExceptionDoesNotCacheValue() { AsyncLoadingCache<Integer, Boolean> cachedResult = new AsyncLoadingCache<>(30000); final AtomicInteger counter = new AtomicInteger(); final Supplier<CompletableFuture<Boolean>> supplier = () -> { counter.incrementAndGet(); throw new RecordCoreException("this is only a test"); }; for (int i = 0; i < 10; i++) { try { cachedResult.orElseGet(1, supplier).join(); fail("should throw RecordCoreException"); } catch (RecordCoreException e) { assertThat(e.getMessage(), containsString("failed getting value")); assertThat(e.getCause().getMessage(), containsString("this is only a test")); } } assertThat("we have to call the supplier each time", counter.get(), is(10)); }
@Test public void testGettingAsyncFailures() { AsyncLoadingCache<Integer, Boolean> cachedResult = new AsyncLoadingCache<>(30000); final AtomicInteger callCount = new AtomicInteger(); final Supplier<CompletableFuture<Boolean>> supplier = () -> MoreAsyncUtil.delayedFuture(1 + random.nextInt(5), TimeUnit.MILLISECONDS).thenApply(ignore -> { int count = callCount.getAndIncrement(); if (count == 0) { throw new RecordCoreException("this is only a test"); } return true; }); try { cachedResult.orElseGet(1, supplier).join(); fail("should throw exception"); } catch (CompletionException ex) { assertThat("we got the expected exception", ex.getCause(), is(instanceOf(RecordCoreException.class))); assertThat("it's the test exception", ex.getCause().getMessage(), containsString("this is only a test")); } assertThat("before future is ready we return the in progress cached future", callCount.get(), is(1)); cachedResult.orElseGet(1, supplier).join(); assertThat("after cached future completes exceptionally we attempt to get the value again", callCount.get(), is(2)); }
@Test public void testGettingImmediateFailure() { AsyncLoadingCache<Integer, Boolean> cachedResult = new AsyncLoadingCache<>(30000); final AtomicInteger callCount = new AtomicInteger(); final Supplier<CompletableFuture<Boolean>> supplier = () -> { int count = callCount.getAndIncrement(); if (count == 0) { CompletableFuture<Boolean> future = new CompletableFuture<>(); RecordCoreException e = new RecordCoreException("this is only a test"); future.completeExceptionally(e); return future; } return CompletableFuture.completedFuture(true); }; try { cachedResult.orElseGet(1, supplier).join(); fail("should throw exception"); } catch (CompletionException ex) { assertThat("we got the expected exception", ex.getCause(), is(instanceOf(RecordCoreException.class))); assertThat("it's the test exception", ex.getCause().getMessage(), containsString("this is only a test")); } assertThat("before future is ready we return the in progress cached future", callCount.get(), is(1)); cachedResult.orElseGet(1, supplier).join(); assertThat("after cached future completes exceptionally we attempt to get the value again", callCount.get(), is(2)); }
@Test public void testReloadFailedGets() throws Exception { AsyncLoadingCache<String, Integer> cachedResult = new AsyncLoadingCache<>(250); AtomicInteger counter1 = new AtomicInteger(); AtomicInteger counter2 = new AtomicInteger(); for (int i = 1; i <= 3; i++) { assertThat(cachedResult.orElseGet("k1", getSupplier(234, counter1, false)).join(), is(234)); assertThat("we do not call the supplier while the cache is valid", counter1.get(), is(1)); assertThrows(CompletionException.class, () -> cachedResult.orElseGet("k2", getSupplier(987, counter2, true)).join()); assertThat("we retry the supplier after a failure", counter2.get(), is(i)); } }
@Test public void testParallelGets() { AsyncLoadingCache<String, Boolean> cachedResult = new AsyncLoadingCache<>(100); final AtomicInteger counter = new AtomicInteger(); CompletableFuture<Void> signal = new CompletableFuture<>(); final Supplier<CompletableFuture<Boolean>> supplier = () -> { counter.incrementAndGet(); return signal.thenApply(ignored -> true); }; List<String> keys = ImmutableList.of("key-1", "key-2", "key-3"); List<CompletableFuture<Boolean>> parallelOperations = new ArrayList<>(); for (int i = 0; i < 20; i++) { for (String key : keys) { parallelOperations.add(cachedResult.orElseGet(key, supplier)); } } signal.complete(null); List<Boolean> values = AsyncUtil.getAll(parallelOperations).join(); for (Boolean value : values) { assertTrue(value); } List<CompletableFuture<Boolean>> afterCompleteOperations = new ArrayList<>(); for (int i = 0; i < 3; i++) { for (String key : keys) { afterCompleteOperations.add(cachedResult.orElseGet(key, supplier)); } } values = AsyncUtil.getAll(afterCompleteOperations).join(); for (Boolean value : values) { assertTrue(value); } assertThat("supplier is called once per incomplete access", counter.get(), is(parallelOperations.size())); }
@Test public void cacheNulls() { AsyncLoadingCache<String, String> cache = new AsyncLoadingCache<>(100); final AtomicInteger counter = new AtomicInteger(); CompletableFuture<Void> signal = new CompletableFuture<>(); final Supplier<CompletableFuture<String>> supplier = () -> { counter.incrementAndGet(); return signal.thenApply(ignored -> null); }; CompletableFuture<String> future = cache.orElseGet("key", supplier); assertEquals(1, counter.get()); signal.complete(null); String value = future.join(); assertNull(value); CompletableFuture<String> cachedFuture = cache.orElseGet("key", supplier); assertEquals(1, counter.get()); value = cachedFuture.join(); assertNull(value); }
@Test public void testDeadline() { AsyncLoadingCache<String, Integer> cachedResult = new AsyncLoadingCache<>(100, 10); final Supplier<CompletableFuture<Integer>> tooLateSupplier = () -> MoreAsyncUtil.delayedFuture(1, TimeUnit.SECONDS) .thenApply(ignore -> 2); final Supplier<CompletableFuture<Integer>> onTimeSupplier = () -> MoreAsyncUtil.delayedFuture(5, TimeUnit.MILLISECONDS) .thenApply(ignore -> 3); try { cachedResult.orElseGet("a-key", tooLateSupplier).join(); fail("should throw CompletionException"); } catch (CompletionException ex) { assertThat("it is caused by a deadline exception", ex.getCause(), is(instanceOf(DeadlineExceededException.class))); assertThat(ex.getCause(), hasMessageContaining("deadline exceeded")); } assertThat("we get the value before the deadline", cachedResult.orElseGet("a-key", onTimeSupplier).join(), is(3)); }