@Test public void testLoaderTimeoutHandling() throws InterruptedException { final TypedCache<String, String> cache = new LocalAsyncCache<>(); final TypedCache<String, String> loadingCache = new LoadingCacheDelegate<>(cache, SLOW_CACHE_LOADER, "meh", metricFactory, 1, TimeUnit.MILLISECONDS); try { loadingCache.getAsync("key").get(); } catch (final ExecutionException e) { Assert.assertEquals("cache timeouts", 1, registry.getCounters().get("LoadingCacheDelegate.meh.loaderTimeouts").getCount()); } }
@Override public ComposableFuture<V> getAsync(final K key) { return ComposableFutures.build(consumer -> { final ComposablePromise<V> promise = newPromise(); final ComposablePromise<V> prev = futureValues.putIfAbsent(key, promise); if (prev != null) { consumeFrom(prev.future(), consumer); return; } final ComposableFuture<V> cachedResult = cache.getAsync(key).withTimeout(duration, timeUnit, "LoadingCacheDelegate fetch from cache named: " + cacheName); cachedResult.consume(res -> { if (res.isSuccess()) { final V result = res.getValue(); if (result == null) { if (cacheMiss != null) { cacheMiss.inc(); } fetchFromLoader(key, promise); } else { if (cacheHits != null) { cacheHits.inc(); } promise.set(result); futureValues.remove(key); } } else { final Throwable error = res.getError(); if (cacheErrors != null) { cacheErrors.inc(); if (error instanceof TimeoutException) { cacheTimeouts.inc(); } } promise.setException(error); futureValues.remove(key); } }); consumeFrom(promise.future(), consumer); } ); }
LoadingCacheDelegate implements TypedCache<K, V> { @Override public ComposableFuture<V> getAsync(final K key) { return ComposableFutures.build(consumer -> { final ComposablePromise<V> promise = newPromise(); final ComposablePromise<V> prev = futureValues.putIfAbsent(key, promise); if (prev != null) { consumeFrom(prev.future(), consumer); return; } final ComposableFuture<V> cachedResult = cache.getAsync(key).withTimeout(duration, timeUnit, "LoadingCacheDelegate fetch from cache named: " + cacheName); cachedResult.consume(res -> { if (res.isSuccess()) { final V result = res.getValue(); if (result == null) { if (cacheMiss != null) { cacheMiss.inc(); } fetchFromLoader(key, promise); } else { if (cacheHits != null) { cacheHits.inc(); } promise.set(result); futureValues.remove(key); } } else { final Throwable error = res.getError(); if (cacheErrors != null) { cacheErrors.inc(); if (error instanceof TimeoutException) { cacheTimeouts.inc(); } } promise.setException(error); futureValues.remove(key); } }); consumeFrom(promise.future(), consumer); } ); } }
LoadingCacheDelegate implements TypedCache<K, V> { @Override public ComposableFuture<V> getAsync(final K key) { return ComposableFutures.build(consumer -> { final ComposablePromise<V> promise = newPromise(); final ComposablePromise<V> prev = futureValues.putIfAbsent(key, promise); if (prev != null) { consumeFrom(prev.future(), consumer); return; } final ComposableFuture<V> cachedResult = cache.getAsync(key).withTimeout(duration, timeUnit, "LoadingCacheDelegate fetch from cache named: " + cacheName); cachedResult.consume(res -> { if (res.isSuccess()) { final V result = res.getValue(); if (result == null) { if (cacheMiss != null) { cacheMiss.inc(); } fetchFromLoader(key, promise); } else { if (cacheHits != null) { cacheHits.inc(); } promise.set(result); futureValues.remove(key); } } else { final Throwable error = res.getError(); if (cacheErrors != null) { cacheErrors.inc(); if (error instanceof TimeoutException) { cacheTimeouts.inc(); } } promise.setException(error); futureValues.remove(key); } }); consumeFrom(promise.future(), consumer); } ); } LoadingCacheDelegate(final TypedCache<K, V> cache, final CacheLoader<K, V> loader, final String cacheName);  LoadingCacheDelegate(final TypedCache<K, V> cache, final CacheLoader<K, V> loader, final String cacheName,
                              final MetricFactory metricFactory);  LoadingCacheDelegate(final TypedCache<K, V> cache, final CacheLoader<K, V> loader, final String cacheName,
                              final MetricFactory metricFactory, final long duration, final TimeUnit timeUnit);  LoadingCacheDelegate(final TypedCache<K, V> cache, final CacheLoader<K, V> loader, final String cacheName,
                              final MetricFactory metricFactory, final long duration, final TimeUnit timeUnit,
                              final boolean failOnError); }
LoadingCacheDelegate implements TypedCache<K, V> { @Override public ComposableFuture<V> getAsync(final K key) { return ComposableFutures.build(consumer -> { final ComposablePromise<V> promise = newPromise(); final ComposablePromise<V> prev = futureValues.putIfAbsent(key, promise); if (prev != null) { consumeFrom(prev.future(), consumer); return; } final ComposableFuture<V> cachedResult = cache.getAsync(key).withTimeout(duration, timeUnit, "LoadingCacheDelegate fetch from cache named: " + cacheName); cachedResult.consume(res -> { if (res.isSuccess()) { final V result = res.getValue(); if (result == null) { if (cacheMiss != null) { cacheMiss.inc(); } fetchFromLoader(key, promise); } else { if (cacheHits != null) { cacheHits.inc(); } promise.set(result); futureValues.remove(key); } } else { final Throwable error = res.getError(); if (cacheErrors != null) { cacheErrors.inc(); if (error instanceof TimeoutException) { cacheTimeouts.inc(); } } promise.setException(error); futureValues.remove(key); } }); consumeFrom(promise.future(), consumer); } ); } LoadingCacheDelegate(final TypedCache<K, V> cache, final CacheLoader<K, V> loader, final String cacheName);  LoadingCacheDelegate(final TypedCache<K, V> cache, final CacheLoader<K, V> loader, final String cacheName,
                              final MetricFactory metricFactory);  LoadingCacheDelegate(final TypedCache<K, V> cache, final CacheLoader<K, V> loader, final String cacheName,
                              final MetricFactory metricFactory, final long duration, final TimeUnit timeUnit);  LoadingCacheDelegate(final TypedCache<K, V> cache, final CacheLoader<K, V> loader, final String cacheName,
                              final MetricFactory metricFactory, final long duration, final TimeUnit timeUnit,
                              final boolean failOnError); @Override ComposableFuture<V> getAsync(final K key); @Override ComposableFuture<Map<K, V>> getBulkAsync(final Iterable<? extends K> keys); @Override ComposableFuture<Boolean> setAsync(final K key, final V value); @Override ComposableFuture<Boolean> setIfAbsentAsync(final K key, final V value); @Override ComposableFuture<Boolean> setAsync(final K key, final EntryMapper<K, V> mapper, final int maxIterations); @Override ComposableFuture<Map<K, Boolean>> setBulkAsync(final Map<? extends K, ? extends V> entries); @Override ComposableFuture<Boolean> deleteAsync(final K key); }
LoadingCacheDelegate implements TypedCache<K, V> { @Override public ComposableFuture<V> getAsync(final K key) { return ComposableFutures.build(consumer -> { final ComposablePromise<V> promise = newPromise(); final ComposablePromise<V> prev = futureValues.putIfAbsent(key, promise); if (prev != null) { consumeFrom(prev.future(), consumer); return; } final ComposableFuture<V> cachedResult = cache.getAsync(key).withTimeout(duration, timeUnit, "LoadingCacheDelegate fetch from cache named: " + cacheName); cachedResult.consume(res -> { if (res.isSuccess()) { final V result = res.getValue(); if (result == null) { if (cacheMiss != null) { cacheMiss.inc(); } fetchFromLoader(key, promise); } else { if (cacheHits != null) { cacheHits.inc(); } promise.set(result); futureValues.remove(key); } } else { final Throwable error = res.getError(); if (cacheErrors != null) { cacheErrors.inc(); if (error instanceof TimeoutException) { cacheTimeouts.inc(); } } promise.setException(error); futureValues.remove(key); } }); consumeFrom(promise.future(), consumer); } ); } LoadingCacheDelegate(final TypedCache<K, V> cache, final CacheLoader<K, V> loader, final String cacheName);  LoadingCacheDelegate(final TypedCache<K, V> cache, final CacheLoader<K, V> loader, final String cacheName,
                              final MetricFactory metricFactory);  LoadingCacheDelegate(final TypedCache<K, V> cache, final CacheLoader<K, V> loader, final String cacheName,
                              final MetricFactory metricFactory, final long duration, final TimeUnit timeUnit);  LoadingCacheDelegate(final TypedCache<K, V> cache, final CacheLoader<K, V> loader, final String cacheName,
                              final MetricFactory metricFactory, final long duration, final TimeUnit timeUnit,
                              final boolean failOnError); @Override ComposableFuture<V> getAsync(final K key); @Override ComposableFuture<Map<K, V>> getBulkAsync(final Iterable<? extends K> keys); @Override ComposableFuture<Boolean> setAsync(final K key, final V value); @Override ComposableFuture<Boolean> setIfAbsentAsync(final K key, final V value); @Override ComposableFuture<Boolean> setAsync(final K key, final EntryMapper<K, V> mapper, final int maxIterations); @Override ComposableFuture<Map<K, Boolean>> setBulkAsync(final Map<? extends K, ? extends V> entries); @Override ComposableFuture<Boolean> deleteAsync(final K key);  }