Class Cache

distributed-cache/src/main/java/Cache.java:14–181

import java.util.function.Function;

public class Cache<KEY, VALUE> {
    private final int maximumSize;
    private final FetchAlgorithm fetchAlgorithm;
    private final Duration expiryTime;
    private final Map<KEY, CompletionStage<Record<KEY, VALUE>>> cache;
    private final ConcurrentSkipListMap<AccessDetails, List<KEY>> priorityQueue;
    private final ConcurrentSkipListMap<Long, List<KEY>> expiryQueue;
    private final DataSource<KEY, VALUE> dataSource;
    private final List<Event<KEY, VALUE>> eventQueue;
    private final ExecutorService[] executorPool;
    private final Timer timer;

    protected Cache(final int maximumSize,
                    final Duration expiryTime,
                    final FetchAlgorithm fetchAlgorithm,
                    final EvictionAlgorithm evictionAlgorithm,
                    final DataSource<KEY, VALUE> dataSource,
                    final Set<KEY> keysToEagerlyLoad,
                    final Timer timer,
                    final int poolSize) {
        this.expiryTime = expiryTime;
        this.maximumSize = maximumSize;
        this.fetchAlgorithm = fetchAlgorithm;
        this.timer = timer;
        this.cache = new ConcurrentHashMap<>();
        this.eventQueue = new CopyOnWriteArrayList<>();
        this.dataSource = dataSource;
        this.executorPool = new ExecutorService[poolSize];
        for (int i = 0; i < poolSize; i++) {
            executorPool[i] = Executors.newSingleThreadExecutor();
        }
        priorityQueue = new ConcurrentSkipListMap<>((first, second) -> {
            final var accessTimeDifference = (int) (first.getLastAccessTime() - second.getLastAccessTime());
            if (evictionAlgorithm.equals(EvictionAlgorithm.LRU)) {
                return accessTimeDifference;
            } else {
                final var accessCountDifference = first.getAccessCount() - second.getAccessCount();
                return accessCountDifference != 0 ? accessCountDifference : accessTimeDifference;
            }
        });
        expiryQueue = new ConcurrentSkipListMap<>();
        final var eagerLoading = keysToEagerlyLoad.stream()
                .map(key -> getThreadFor(key, addToCache(key, loadFromDB(dataSource, key))))
                .toArray(CompletableFuture[]::new);
        CompletableFuture.allOf(eagerLoading).join();
    }

    private <U> CompletionStage<U> getThreadFor(KEY key, CompletionStage<U> task) {
        return CompletableFuture.supplyAsync(() -> task, executorPool[Math.abs(key.hashCode() % executorPool.length)]).thenCompose(Function.identity());
    }

    public CompletionStage<VALUE> get(KEY key) {
        return getThreadFor(key, getFromCache(key));
    }

    public CompletionStage<Void> set(KEY key, VALUE value) {
        return getThreadFor(key, setInCache(key, value));
    }