#5669

Committed 21 Jul 2026 02:11AM UTC coverage: 99.965% (-0.01%) from 99.976%

Build # #5669

Build Type

push

github

Committed by

ben-manes

Commit Message

Record a read-through get miss on a loader throw

LoadingCacheProxy.getOrLoad recorded the miss after cache.get runs the
read-through load, so a throwing loader propagated before the miss was
counted -- getCacheMisses stayed 0. The loading getAll, the RI
(increaseCacheMisses before cacheLoader.load), and core all count the
miss regardless of load outcome (the find already failed). Record it
before the load. Not TCK-observable (its stats tests use no loader).

Pin untested jcache close-gate and config paths

Add coverage for public-API paths the TCK never exercises: clear() and
register/deregisterCacheEntryListener on a closed cache throw ISE
(CG-1, CG-2); createCache of a name defined in application.conf throws
CacheException (CG-6); and createCache with a plain non-Complete
Configuration (types + store-by-value only) succeeds via the basic
resolveConfigurationFor branch (CG-7). Test-only; the code was already
correct.

Coverage Stats

4194 of 4203 branches covered (99.79%)

9 of 9 new or added lines in 2 files covered. (100.0%)

2 existing lines in 2 files now uncovered.

8472 of 8475 relevant lines covered (99.96%)

1.0 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.88

/jcache/src/main/java/com/github/benmanes/caffeine/jcache/event/EventDispatcher.java

/*
 * Copyright 2015 Ben Manes. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.github.benmanes.caffeine.jcache.event;

import static java.util.Objects.requireNonNull;

import java.lang.System.Logger;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executor;
import java.util.function.Function;

import javax.cache.Cache;
import javax.cache.configuration.CacheEntryListenerConfiguration;
import javax.cache.configuration.MutableCacheEntryListenerConfiguration;
import javax.cache.event.CacheEntryEventFilter;
import javax.cache.event.CacheEntryListener;
import javax.cache.event.CacheEntryListenerException;
import javax.cache.event.EventType;

import org.jspecify.annotations.Nullable;

import com.google.errorprone.annotations.Var;

/**
 * A dispatcher that publishes cache events to listeners for asynchronous execution.
 * <p>
 * A {@link CacheEntryListener} is required to receive events in the order of the actions being
 * performed on the associated key. This implementation supports this by using a dispatch queue for
 * each listener and key pair, and provides the following characteristics:
 * <ul>
 *   <li>A listener may be executed in parallel for events with different keys
 *   <li>A listener is executed sequentially for events with the same key. This creates a dependency
 *       relationship between events and waiting dependents do not consume threads.
 *   <li>A listener receives a single event per invocation; batch processing is not supported
 *   <li>Multiple listeners may be executed in parallel for the same event
 *   <li>Listeners process events at their own rate and do not explicitly block each other
 *   <li>Listeners share a pool of threads for event processing. A slow listener may limit the
 *       throughput if all threads are busy handling distinct events, causing the execution of other
 *       listeners to be delayed until the executor is able to process the work.
 * </ul>
 * <p>
 * Some listeners may be configured as <code>synchronous</code>, meaning that the publishing thread
 * should wait until the listener has processed the event. The calling thread should publish within
 * an atomic block that mutates the entry, and complete the operation by calling
 * {@link #awaitSynchronous()} or {@link #ignoreSynchronous()}.
 *
 * @author ben.manes@gmail.com (Ben Manes)
 */
public final class EventDispatcher<K, V> {
  static final Logger logger = System.getLogger(EventDispatcher.class.getName());

  final ConcurrentMap<
      Registration<K, V>,
      ConcurrentMap<K, CompletableFuture<@Nullable CacheEntryListenerException>>> dispatchQueues;
  final ThreadLocal<List<CompletableFuture<@Nullable CacheEntryListenerException>>> pending;
  final Executor executor;

  public EventDispatcher(Executor executor) {
    this.pending = ThreadLocal.withInitial(ArrayList::new);
    this.dispatchQueues = new ConcurrentHashMap<>();
    this.executor = requireNonNull(executor);
  }

  /** Returns the cache entry listener registrations. */
  public Set<Registration<K, V>> registrations() {
    return Collections.unmodifiableSet(dispatchQueues.keySet());
  }

  /**
   * Registers a cache entry listener based on the supplied configuration.
   *
   * @param configuration the listener's configuration.
   */
  @SuppressWarnings("PMD.CloseResource")
  public void register(CacheEntryListenerConfiguration<K, V> configuration) {
    if (configuration.getCacheEntryListenerFactory() == null) {
      return;
    }
    var listener = new EventTypeAwareListener<K, V>(
        configuration.getCacheEntryListenerFactory().create());

    var factory = configuration.getCacheEntryEventFilterFactory();
    CacheEntryEventFilter<K, V> filter = (factory == null)
        ? event -> true
        : new EventTypeFilter<>(listener, factory.create());

    var registration = new Registration<>(configuration, filter, listener);
    dispatchQueues.putIfAbsent(registration, new ConcurrentHashMap<>());
  }

  /**
   * Deregisters a cache entry listener based on the supplied configuration.
   *
   * @param configuration the listener's configuration.
   */
  public void deregister(CacheEntryListenerConfiguration<K, V> configuration) {
    requireNonNull(configuration);
    var key = new MutableCacheEntryListenerConfiguration<>(configuration);
    dispatchQueues.keySet().removeIf(registration -> key.equals(registration.getConfiguration()));
  }

  /**
   * Publishes a creation event for the entry to the interested listeners.
   *
   * @param cache the cache where the entry was created
   * @param key the entry's key
   * @param value the entry's value
   */
  public void publishCreated(Cache<K, V> cache, K key, V value) {
    publish(cache, EventType.CREATED, key, /* hasOldValue= */ false,
        /* oldValue= */ null, /* newValue= */ value, /* quiet= */ false);
  }

  /**
   * Publishes an update event for the entry to the interested listeners.
   *
   * @param cache the cache where the entry was updated
   * @param key the entry's key
   * @param oldValue the entry's old value
   * @param newValue the entry's new value
   */
  public void publishUpdated(Cache<K, V> cache, K key, V oldValue, V newValue) {
    publish(cache, EventType.UPDATED, key, /* hasOldValue= */ true,
        oldValue, newValue, /* quiet= */ false);
  }

  /**
   * Publishes an update event for the entry to the interested listeners. This method does not
   * register the synchronous listener's future with {@link #awaitSynchronous()}.
   *
   * @param cache the cache where the entry was updated
   * @param key the entry's key
   * @param oldValue the entry's old value
   * @param newValue the entry's new value
   */
  public void publishUpdatedQuietly(Cache<K, V> cache, K key, V oldValue, V newValue) {
    publish(cache, EventType.UPDATED, key, /* hasOldValue= */ true,
        oldValue, newValue, /* quiet= */ true);
  }

  /**
   * Publishes a removal event for the entry to the interested listeners.
   *
   * @param cache the cache where the entry was removed
   * @param key the entry's key
   * @param value the entry's value
   */
  public void publishRemoved(Cache<K, V> cache, K key, V value) {
    publish(cache, EventType.REMOVED, key, /* hasOldValue= */ true,
        /* oldValue= */ value, /* newValue= */ value, /* quiet= */ false);
  }

  /**
   * Publishes a removal event for the entry to the interested listeners. This method does not
   * register the synchronous listener's future with {@link #awaitSynchronous()}.
   *
   * @param cache the cache where the entry was removed
   * @param key the entry's key
   * @param value the entry's value
   */
  public void publishRemovedQuietly(Cache<K, V> cache, K key, V value) {
    publish(cache, EventType.REMOVED, key, /* hasOldValue= */ true,
        /* oldValue= */ value, /* newValue= */ value, /* quiet= */ true);
  }

  /**
   * Publishes an expiration event for the entry to the interested listeners.
   *
   * @param cache the cache where the entry expired
   * @param key the entry's key
   * @param value the entry's value
   */
  public void publishExpired(Cache<K, V> cache, K key, V value) {
    publish(cache, EventType.EXPIRED, key, /* hasOldValue= */ true,
        /* oldValue= */ value, /* newValue= */ value, /* quiet= */ false);
  }

  /**
   * Publishes an expiration event for the entry to the interested listeners. This method does not
   * register the synchronous listener's future with {@link #awaitSynchronous()}.
   *
   * @param cache the cache where the entry expired
   * @param key the entry's key
   * @param value the entry's value
   */
  public void publishExpiredQuietly(Cache<K, V> cache, K key, V value) {
    publish(cache, EventType.EXPIRED, key, /* hasOldValue= */ true,
        /* oldValue= */ value, /* newValue= */ value, /* quiet= */ true);
  }

  /**
   * Blocks until all of the synchronous listeners have finished processing the events this thread
   * published.
   */
  @SuppressWarnings("PMD.PreserveStackTrace")
  public void awaitSynchronous() {
    try {
      var error = chainSynchronous().join();
      if (error != null) {
        throw error;
      }
    } catch (CompletionException e) {
      if (e.getCause() instanceof CacheEntryListenerException) {
        throw (CacheEntryListenerException) e.getCause();
      } else if (e.getCause() instanceof Error) {
        throw (Error) e.getCause();
      }
      throw new CacheEntryListenerException(e);
    }
  }

  /**
   * Ignores and clears the queued futures to the synchronous listeners that are processing events
   * this thread published.
   */
  public void ignoreSynchronous() {
    pending.get().clear();
  }

  /**
   * Returns a future that completes once this thread's synchronous listeners have finished
   * processing the events it published.
   */
  @SuppressWarnings("UnnamedVariable")
  public CompletableFuture<@Nullable CacheEntryListenerException> chainSynchronous() {
    var synchronous = pending.get();
    var futures = List.copyOf(synchronous);
    CompletableFuture<@Nullable CacheEntryListenerException> future =
        CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new))
            .thenApply(new Function<@Nullable Void, @Nullable CacheEntryListenerException>() {
              @Override @SuppressWarnings("UnnamedVariable")
              public @Nullable CacheEntryListenerException apply(@Nullable Void unused) {
                return futures.stream()
                    .map(CompletableFuture::join)
                    .filter(Objects::nonNull)
                    .reduce((e1, e2) -> {
                      e1.addSuppressed(e2);
                      return e1;
                    }).orElse(null);
              }
            });
    synchronous.clear();
    return future;
  }

  /** Broadcasts the event to the interested listener's dispatch queues. */
  @SuppressWarnings("FutureReturnValueIgnored")
  private void publish(Cache<K, V> cache, EventType eventType, K key,
      boolean hasOldValue, @Nullable V oldValue, @Nullable V newValue, boolean quiet) {
    if (dispatchQueues.isEmpty()) {
      return;
    }

    @Var JCacheEntryEvent<K, V> event = null;
    for (var entry : dispatchQueues.entrySet()) {
      var registration = entry.getKey();
      if (!registration.getCacheEntryListener().isCompatible(eventType)) {
        continue;
      }
      if (event == null) {
        event = new JCacheEntryEvent<>(cache, eventType, key, hasOldValue, oldValue, newValue);
      }
      if (!registration.getCacheEntryFilter().evaluate(event)) {
        continue;
      }

      JCacheEntryEvent<K, V> e = event;
      var dispatchQueue = entry.getValue();
      @SuppressWarnings("PMD.CloseResource")
      var listener = registration.getCacheEntryListener();
      var future = dispatchQueue.compute(key, (k, queue) -> {
        return (queue == null)
            ? CompletableFuture.supplyAsync(() -> listener.dispatch(e), executor)
            : queue.thenApplyAsync(prev -> listener.dispatch(e), executor);
      });
      future.whenComplete((result, error) -> {
        // optimistic check to avoid locking if not a match
        if (dispatchQueue.get(key) == future) {
          dispatchQueue.remove(key, future);
        }
      });
      if (registration.isSynchronous() && !quiet) {
        pending.get().add(future);
      }
    }
  }
}

1	/*
2	* Copyright 2015 Ben Manes. All Rights Reserved.
3	*
4	* Licensed under the Apache License, Version 2.0 (the "License");
5	* you may not use this file except in compliance with the License.
6	* You may obtain a copy of the License at
7	*
8	* http://www.apache.org/licenses/LICENSE-2.0
9	*
10	* Unless required by applicable law or agreed to in writing, software
11	* distributed under the License is distributed on an "AS IS" BASIS,
12	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	* See the License for the specific language governing permissions and
14	* limitations under the License.
15	*/
16	package com.github.benmanes.caffeine.jcache.event;
17
18	import static java.util.Objects.requireNonNull;
19
20	import java.lang.System.Logger;
21	import java.util.ArrayList;
22	import java.util.Collections;
23	import java.util.List;
24	import java.util.Objects;
25	import java.util.Set;
26	import java.util.concurrent.CompletableFuture;
27	import java.util.concurrent.CompletionException;
28	import java.util.concurrent.ConcurrentHashMap;
29	import java.util.concurrent.ConcurrentMap;
30	import java.util.concurrent.Executor;
31	import java.util.function.Function;
32
33	import javax.cache.Cache;
34	import javax.cache.configuration.CacheEntryListenerConfiguration;
35	import javax.cache.configuration.MutableCacheEntryListenerConfiguration;
36	import javax.cache.event.CacheEntryEventFilter;
37	import javax.cache.event.CacheEntryListener;
38	import javax.cache.event.CacheEntryListenerException;
39	import javax.cache.event.EventType;
40
41	import org.jspecify.annotations.Nullable;
42
43	import com.google.errorprone.annotations.Var;
44
45	/**
46	* A dispatcher that publishes cache events to listeners for asynchronous execution.
47	* <p>
48	* A {@link CacheEntryListener} is required to receive events in the order of the actions being
49	* performed on the associated key. This implementation supports this by using a dispatch queue for
50	* each listener and key pair, and provides the following characteristics:
51	* <ul>
52	* <li>A listener may be executed in parallel for events with different keys
53	* <li>A listener is executed sequentially for events with the same key. This creates a dependency
54	* relationship between events and waiting dependents do not consume threads.
55	* <li>A listener receives a single event per invocation; batch processing is not supported
56	* <li>Multiple listeners may be executed in parallel for the same event
57	* <li>Listeners process events at their own rate and do not explicitly block each other
58	* <li>Listeners share a pool of threads for event processing. A slow listener may limit the
59	* throughput if all threads are busy handling distinct events, causing the execution of other
60	* listeners to be delayed until the executor is able to process the work.
61	* </ul>
62	* <p>
63	* Some listeners may be configured as <code>synchronous</code>, meaning that the publishing thread
64	* should wait until the listener has processed the event. The calling thread should publish within
65	* an atomic block that mutates the entry, and complete the operation by calling
66	* {@link #awaitSynchronous()} or {@link #ignoreSynchronous()}.
67	*
68	* @author ben.manes@gmail.com (Ben Manes)
69	*/
70	public final class EventDispatcher<K, V> {
71	static final Logger logger = System.getLogger(EventDispatcher.class.getName());	1✔
72
73	final ConcurrentMap<
74	Registration<K, V>,
75	ConcurrentMap<K, CompletableFuture<@Nullable CacheEntryListenerException>>> dispatchQueues;
76	final ThreadLocal<List<CompletableFuture<@Nullable CacheEntryListenerException>>> pending;
77	final Executor executor;
78
79	public EventDispatcher(Executor executor) {	1✔
80	this.pending = ThreadLocal.withInitial(ArrayList::new);	1✔
81	this.dispatchQueues = new ConcurrentHashMap<>();	1✔
82	this.executor = requireNonNull(executor);	1✔
83	}	1✔
84
85	/** Returns the cache entry listener registrations. */
86	public Set<Registration<K, V>> registrations() {
87	return Collections.unmodifiableSet(dispatchQueues.keySet());	1✔
88	}
89
90	/**
91	* Registers a cache entry listener based on the supplied configuration.
92	*
93	* @param configuration the listener's configuration.
94	*/
95	@SuppressWarnings("PMD.CloseResource")
96	public void register(CacheEntryListenerConfiguration<K, V> configuration) {
97	if (configuration.getCacheEntryListenerFactory() == null) {	1✔
98	return;	1✔
99	}
100	var listener = new EventTypeAwareListener<K, V>(	1✔
101	configuration.getCacheEntryListenerFactory().create());	1✔
102
103	var factory = configuration.getCacheEntryEventFilterFactory();	1✔
104	CacheEntryEventFilter<K, V> filter = (factory == null)	1✔
105	? event -> true	1✔
106	: new EventTypeFilter<>(listener, factory.create());	1✔
107
108	var registration = new Registration<>(configuration, filter, listener);	1✔
109	dispatchQueues.putIfAbsent(registration, new ConcurrentHashMap<>());	1✔
110	}	1✔
111
112	/**
113	* Deregisters a cache entry listener based on the supplied configuration.
114	*
115	* @param configuration the listener's configuration.
116	*/
117	public void deregister(CacheEntryListenerConfiguration<K, V> configuration) {
118	requireNonNull(configuration);	1✔
119	var key = new MutableCacheEntryListenerConfiguration<>(configuration);	1✔
120	dispatchQueues.keySet().removeIf(registration -> key.equals(registration.getConfiguration()));	1✔
121	}	1✔
122
123	/**
124	* Publishes a creation event for the entry to the interested listeners.
125	*
126	* @param cache the cache where the entry was created
127	* @param key the entry's key
128	* @param value the entry's value
129	*/
130	public void publishCreated(Cache<K, V> cache, K key, V value) {
131	publish(cache, EventType.CREATED, key, /* hasOldValue= */ false,	1✔
132	/* oldValue= / null, / newValue= / value, / quiet= */ false);
133	}	1✔
134
135	/**
136	* Publishes an update event for the entry to the interested listeners.
137	*
138	* @param cache the cache where the entry was updated
139	* @param key the entry's key
140	* @param oldValue the entry's old value
141	* @param newValue the entry's new value
142	*/
143	public void publishUpdated(Cache<K, V> cache, K key, V oldValue, V newValue) {
144	publish(cache, EventType.UPDATED, key, /* hasOldValue= */ true,	1✔
145	oldValue, newValue, /* quiet= */ false);
146	}	1✔
147
148	/**
149	* Publishes an update event for the entry to the interested listeners. This method does not
150	* register the synchronous listener's future with {@link #awaitSynchronous()}.
151	*
152	* @param cache the cache where the entry was updated
153	* @param key the entry's key
154	* @param oldValue the entry's old value
155	* @param newValue the entry's new value
156	*/
157	public void publishUpdatedQuietly(Cache<K, V> cache, K key, V oldValue, V newValue) {
158	publish(cache, EventType.UPDATED, key, /* hasOldValue= */ true,	1✔
159	oldValue, newValue, /* quiet= */ true);
160	}	1✔
161
162	/**
163	* Publishes a removal event for the entry to the interested listeners.
164	*
165	* @param cache the cache where the entry was removed
166	* @param key the entry's key
167	* @param value the entry's value
168	*/
169	public void publishRemoved(Cache<K, V> cache, K key, V value) {
170	publish(cache, EventType.REMOVED, key, /* hasOldValue= */ true,	1✔
171	/* oldValue= / value, / newValue= / value, / quiet= */ false);
172	}	1✔
173
174	/**
175	* Publishes a removal event for the entry to the interested listeners. This method does not
176	* register the synchronous listener's future with {@link #awaitSynchronous()}.
177	*
178	* @param cache the cache where the entry was removed
179	* @param key the entry's key
180	* @param value the entry's value
181	*/
182	public void publishRemovedQuietly(Cache<K, V> cache, K key, V value) {
183	publish(cache, EventType.REMOVED, key, /* hasOldValue= */ true,	1✔
184	/* oldValue= / value, / newValue= / value, / quiet= */ true);
185	}	1✔
186
187	/**
188	* Publishes an expiration event for the entry to the interested listeners.
189	*
190	* @param cache the cache where the entry expired
191	* @param key the entry's key
192	* @param value the entry's value
193	*/
194	public void publishExpired(Cache<K, V> cache, K key, V value) {
195	publish(cache, EventType.EXPIRED, key, /* hasOldValue= */ true,	1✔
196	/* oldValue= / value, / newValue= / value, / quiet= */ false);
197	}	1✔
198
199	/**
200	* Publishes an expiration event for the entry to the interested listeners. This method does not
201	* register the synchronous listener's future with {@link #awaitSynchronous()}.
202	*
203	* @param cache the cache where the entry expired
204	* @param key the entry's key
205	* @param value the entry's value
206	*/
207	public void publishExpiredQuietly(Cache<K, V> cache, K key, V value) {
208	publish(cache, EventType.EXPIRED, key, /* hasOldValue= */ true,	1✔
209	/* oldValue= / value, / newValue= / value, / quiet= */ true);
210	}	1✔
211
212	/**
213	* Blocks until all of the synchronous listeners have finished processing the events this thread
214	* published.
215	*/
216	@SuppressWarnings("PMD.PreserveStackTrace")
217	public void awaitSynchronous() {
218	try {
219	var error = chainSynchronous().join();	1✔
220	if (error != null) {	1✔
221	throw error;	1✔
222	}
223	} catch (CompletionException e) {	1✔
224	if (e.getCause() instanceof CacheEntryListenerException) {	1!
UNCOV 225	throw (CacheEntryListenerException) e.getCause();	×
226	} else if (e.getCause() instanceof Error) {	1✔
227	throw (Error) e.getCause();	1✔
228	}
229	throw new CacheEntryListenerException(e);	1✔
230	}	1✔
231	}	1✔
232
233	/**
234	* Ignores and clears the queued futures to the synchronous listeners that are processing events
235	* this thread published.
236	*/
237	public void ignoreSynchronous() {
238	pending.get().clear();	1✔
239	}	1✔
240
241	/**
242	* Returns a future that completes once this thread's synchronous listeners have finished
243	* processing the events it published.
244	*/
245	@SuppressWarnings("UnnamedVariable")
246	public CompletableFuture<@Nullable CacheEntryListenerException> chainSynchronous() {
247	var synchronous = pending.get();	1✔
248	var futures = List.copyOf(synchronous);	1✔
249	CompletableFuture<@Nullable CacheEntryListenerException> future =	1✔
250	CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new))	1✔
251	.thenApply(new Function<@Nullable Void, @Nullable CacheEntryListenerException>() {	1✔
252	@Override @SuppressWarnings("UnnamedVariable")
253	public @Nullable CacheEntryListenerException apply(@Nullable Void unused) {
254	return futures.stream()	1✔
255	.map(CompletableFuture::join)	1✔
256	.filter(Objects::nonNull)	1✔
257	.reduce((e1, e2) -> {	1✔
258	e1.addSuppressed(e2);	1✔
259	return e1;	1✔
260	}).orElse(null);	1✔
261	}
262	});
263	synchronous.clear();	1✔
264	return future;	1✔
265	}
266
267	/** Broadcasts the event to the interested listener's dispatch queues. */
268	@SuppressWarnings("FutureReturnValueIgnored")
269	private void publish(Cache<K, V> cache, EventType eventType, K key,
270	boolean hasOldValue, @Nullable V oldValue, @Nullable V newValue, boolean quiet) {
271	if (dispatchQueues.isEmpty()) {	1✔
272	return;	1✔
273	}
274
275	@Var JCacheEntryEvent<K, V> event = null;	1✔
276	for (var entry : dispatchQueues.entrySet()) {	1✔
277	var registration = entry.getKey();	1✔
278	if (!registration.getCacheEntryListener().isCompatible(eventType)) {	1✔
279	continue;	1✔
280	}
281	if (event == null) {	1✔
282	event = new JCacheEntryEvent<>(cache, eventType, key, hasOldValue, oldValue, newValue);	1✔
283	}
284	if (!registration.getCacheEntryFilter().evaluate(event)) {	1✔
285	continue;	1✔
286	}
287
288	JCacheEntryEvent<K, V> e = event;	1✔
289	var dispatchQueue = entry.getValue();	1✔
290	@SuppressWarnings("PMD.CloseResource")
291	var listener = registration.getCacheEntryListener();	1✔
292	var future = dispatchQueue.compute(key, (k, queue) -> {	1✔
293	return (queue == null)	1✔
294	? CompletableFuture.supplyAsync(() -> listener.dispatch(e), executor)	1✔
295	: queue.thenApplyAsync(prev -> listener.dispatch(e), executor);	1✔
296	});
297	future.whenComplete((result, error) -> {	1✔
298	// optimistic check to avoid locking if not a match
299	if (dispatchQueue.get(key) == future) {	1✔
300	dispatchQueue.remove(key, future);	1✔
301	}
302	});	1✔
303	if (registration.isSynchronous() && !quiet) {	1✔
304	pending.get().add(future);	1✔
305	}
306	}	1✔
307	}	1✔
308	}

ben-manes / caffeine / #5669

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous