#5665

Committed 20 Jul 2026 03:32AM UTC coverage: 99.976% (-0.02%) from 100.0%

Build # #5665

Build Type

push

github

Committed by

ben-manes

Commit Message

Propagate a synchronous jcache listener exception

JSR-107 requires a synchronous CacheEntryListener exception to propagate
to the caller, wrapped as a CacheEntryListenerException (javax.cache.event
package-info: "this will propagate back to the caller"; CacheEntryListener:
"catch any other Exception ... wrap and rethrow"). The EventDispatcher
swallowed it: EventTypeAwareListener.dispatch caught and logged, so the
synchronous future completed normally and awaitSynchronous never saw the
failure.

Now dispatch returns the exception (a CacheEntryListenerException as-is, or
a RuntimeException wrapped in one; an Error is logged and rethrown as-is,
not wrapped, matching the RI which catches only Exception); the per-key
chain future carries it as its result -- so a throwing listener never fails
the future and same-key ordering is preserved -- and awaitSynchronous throws
the first (extras addSuppressed). An asynchronous or quiet (background
refresh reload / native eviction) failure is logged, not propagated (no
synchronous caller); an executor-level failure surfaces as a
CacheEntryListenerException, and a listener Error propagates as-is.

The ecosystem is split (all source-verified): spec + RI + cache2k +
Hazelcast + Infinispan propagate; Ehcache 3 and Coherence -- the two
executor-dispatch implementations, as Caffeine was -- swallow and log.
Followed the spec and the RI.

Pinned by EventDispatcherTest (sync propagates and wraps, async swallowed,
same-key chain survives a throw, multiple exceptions suppressed) and
EventTypeAwareListenerTest. Catalogued in jsr107-conformance.md, whose
filter-exception entry is corrected: a filter runs inside the compute so it
must swallow, while the listener runs after commit so it can propagate.

Coverage Stats

4199 of 4207 branches covered (99.81%)

31 of 33 new or added lines in 2 files covered. (93.94%)

8473 of 8475 relevant lines covered (99.98%)

1.0 hits per line

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

98.82

/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 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() {
    var futures = pending.get();
    if (futures.isEmpty()) {
      return;
    }
    try {
      CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new)).join();
      var error = futures.stream()
          .map(CompletableFuture::join)
          .filter(Objects::nonNull)
          .reduce((e1, e2) -> {
            e1.addSuppressed(e2);
            return e1;
          }).orElse(null);
      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);
    } finally {
      futures.clear();
    }
  }

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

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

ben-manes / caffeine / #5665

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