@Test void shouldCacheResultOfPreviousTask() { final StubTask task1 = new StubTask(4); final StubTask task2 = new StubTask(4); final SafeFuture<Optional<String>> result1 = taskQueue.perform(task1); task1.completeTask(); task1.assertPerformedWithoutRebase(); assertThat(result1).isCompletedWithValue(task1.getExpectedValue()); final SafeFuture<Optional<String>> result = taskQueue.perform(task2); assertThat(result).isCompletedWithValue(task2.getExpectedValue()); task2.assertNotPerformed(); task2.assertNotRebased(); assertCacheSizeMetric(1); assertCacheHitCount(1); }
public synchronized SafeFuture<Optional<V>> perform(final CacheableTask<K, V> task) { final V cachedResult = cache.get(task.getKey()); if (cachedResult != null) { cachedTaskCounter.inc(); return SafeFuture.completedFuture(Optional.of(cachedResult)); } final SafeFuture<Optional<V>> currentPendingTask = pendingTasks.get(task.getKey()); if (currentPendingTask != null) { duplicateTaskCounter.inc(); return currentPendingTask; } final SafeFuture<Optional<V>> generationResult = new SafeFuture<>(); pendingTasks.put(task.getKey(), generationResult); final Optional<SafeFuture<Optional<V>>> newBase = task.streamIntermediateSteps() .map( key -> Optional.ofNullable(cache.get(key)) .map(value -> SafeFuture.completedFuture(Optional.of(value))) .orElse(pendingTasks.get(key))) .filter(Objects::nonNull) .findFirst(); if (newBase.isPresent()) { rebasedTaskCounter.inc(); newBase .get() .thenAccept(ancestorResult -> queueTask(task.rebase(ancestorResult))) .finish(error -> completePendingTask(task, Optional.empty(), error)); return generationResult; } newTaskCounter.inc(); queueTask(task); return generationResult; }
CachingTaskQueue { public synchronized SafeFuture<Optional<V>> perform(final CacheableTask<K, V> task) { final V cachedResult = cache.get(task.getKey()); if (cachedResult != null) { cachedTaskCounter.inc(); return SafeFuture.completedFuture(Optional.of(cachedResult)); } final SafeFuture<Optional<V>> currentPendingTask = pendingTasks.get(task.getKey()); if (currentPendingTask != null) { duplicateTaskCounter.inc(); return currentPendingTask; } final SafeFuture<Optional<V>> generationResult = new SafeFuture<>(); pendingTasks.put(task.getKey(), generationResult); final Optional<SafeFuture<Optional<V>>> newBase = task.streamIntermediateSteps() .map( key -> Optional.ofNullable(cache.get(key)) .map(value -> SafeFuture.completedFuture(Optional.of(value))) .orElse(pendingTasks.get(key))) .filter(Objects::nonNull) .findFirst(); if (newBase.isPresent()) { rebasedTaskCounter.inc(); newBase .get() .thenAccept(ancestorResult -> queueTask(task.rebase(ancestorResult))) .finish(error -> completePendingTask(task, Optional.empty(), error)); return generationResult; } newTaskCounter.inc(); queueTask(task); return generationResult; } }
CachingTaskQueue { public synchronized SafeFuture<Optional<V>> perform(final CacheableTask<K, V> task) { final V cachedResult = cache.get(task.getKey()); if (cachedResult != null) { cachedTaskCounter.inc(); return SafeFuture.completedFuture(Optional.of(cachedResult)); } final SafeFuture<Optional<V>> currentPendingTask = pendingTasks.get(task.getKey()); if (currentPendingTask != null) { duplicateTaskCounter.inc(); return currentPendingTask; } final SafeFuture<Optional<V>> generationResult = new SafeFuture<>(); pendingTasks.put(task.getKey(), generationResult); final Optional<SafeFuture<Optional<V>>> newBase = task.streamIntermediateSteps() .map( key -> Optional.ofNullable(cache.get(key)) .map(value -> SafeFuture.completedFuture(Optional.of(value))) .orElse(pendingTasks.get(key))) .filter(Objects::nonNull) .findFirst(); if (newBase.isPresent()) { rebasedTaskCounter.inc(); newBase .get() .thenAccept(ancestorResult -> queueTask(task.rebase(ancestorResult))) .finish(error -> completePendingTask(task, Optional.empty(), error)); return generationResult; } newTaskCounter.inc(); queueTask(task); return generationResult; } CachingTaskQueue(
      final AsyncRunner asyncRunner,
      final MetricsSystem metricsSystem,
      final String metricsPrefix,
      final IntSupplier activeTaskLimit,
      final int maxCacheSize); }
CachingTaskQueue { public synchronized SafeFuture<Optional<V>> perform(final CacheableTask<K, V> task) { final V cachedResult = cache.get(task.getKey()); if (cachedResult != null) { cachedTaskCounter.inc(); return SafeFuture.completedFuture(Optional.of(cachedResult)); } final SafeFuture<Optional<V>> currentPendingTask = pendingTasks.get(task.getKey()); if (currentPendingTask != null) { duplicateTaskCounter.inc(); return currentPendingTask; } final SafeFuture<Optional<V>> generationResult = new SafeFuture<>(); pendingTasks.put(task.getKey(), generationResult); final Optional<SafeFuture<Optional<V>>> newBase = task.streamIntermediateSteps() .map( key -> Optional.ofNullable(cache.get(key)) .map(value -> SafeFuture.completedFuture(Optional.of(value))) .orElse(pendingTasks.get(key))) .filter(Objects::nonNull) .findFirst(); if (newBase.isPresent()) { rebasedTaskCounter.inc(); newBase .get() .thenAccept(ancestorResult -> queueTask(task.rebase(ancestorResult))) .finish(error -> completePendingTask(task, Optional.empty(), error)); return generationResult; } newTaskCounter.inc(); queueTask(task); return generationResult; } CachingTaskQueue(
      final AsyncRunner asyncRunner,
      final MetricsSystem metricsSystem,
      final String metricsPrefix,
      final IntSupplier activeTaskLimit,
      final int maxCacheSize); static CachingTaskQueue<K, V> create(
      final AsyncRunner asyncRunner,
      final MetricsSystem metricsSystem,
      final String metricsPrefix,
      final int maxCacheSize); void startMetrics(); synchronized SafeFuture<Optional<V>> perform(final CacheableTask<K, V> task); Optional<V> getIfAvailable(final K key); void cache(final K key, final V value); void cacheAll(final Map<K, V> values); void remove(final K key); }
CachingTaskQueue { public synchronized SafeFuture<Optional<V>> perform(final CacheableTask<K, V> task) { final V cachedResult = cache.get(task.getKey()); if (cachedResult != null) { cachedTaskCounter.inc(); return SafeFuture.completedFuture(Optional.of(cachedResult)); } final SafeFuture<Optional<V>> currentPendingTask = pendingTasks.get(task.getKey()); if (currentPendingTask != null) { duplicateTaskCounter.inc(); return currentPendingTask; } final SafeFuture<Optional<V>> generationResult = new SafeFuture<>(); pendingTasks.put(task.getKey(), generationResult); final Optional<SafeFuture<Optional<V>>> newBase = task.streamIntermediateSteps() .map( key -> Optional.ofNullable(cache.get(key)) .map(value -> SafeFuture.completedFuture(Optional.of(value))) .orElse(pendingTasks.get(key))) .filter(Objects::nonNull) .findFirst(); if (newBase.isPresent()) { rebasedTaskCounter.inc(); newBase .get() .thenAccept(ancestorResult -> queueTask(task.rebase(ancestorResult))) .finish(error -> completePendingTask(task, Optional.empty(), error)); return generationResult; } newTaskCounter.inc(); queueTask(task); return generationResult; } CachingTaskQueue(
      final AsyncRunner asyncRunner,
      final MetricsSystem metricsSystem,
      final String metricsPrefix,
      final IntSupplier activeTaskLimit,
      final int maxCacheSize); static CachingTaskQueue<K, V> create(
      final AsyncRunner asyncRunner,
      final MetricsSystem metricsSystem,
      final String metricsPrefix,
      final int maxCacheSize); void startMetrics(); synchronized SafeFuture<Optional<V>> perform(final CacheableTask<K, V> task); Optional<V> getIfAvailable(final K key); void cache(final K key, final V value); void cacheAll(final Map<K, V> values); void remove(final K key);  }