#5707

Committed 02 Aug 2026 12:45AM UTC coverage: 0.106% (-99.9%) from 100.0%

Build # #5707

Build Type

push

github

Committed by

ben-manes

Commit Message

fix wikibench trace reader after overly strict audit fixes

Coverage Stats

0 of 4235 branches covered (0.0%)

9 of 8528 relevant lines covered (0.11%)

0.0 hits per line

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

0.0

/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() {
    if (pending.get().isEmpty()) {
      return;
    }
    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) -> {
                      if (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.handleAsync((prev, error) -> 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());	×
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) {	×
80	this.pending = ThreadLocal.withInitial(ArrayList::new);	×
81	this.dispatchQueues = new ConcurrentHashMap<>();	×
82	this.executor = requireNonNull(executor);	×
83	}	×
84
85	/** Returns the cache entry listener registrations. */
86	public Set<Registration<K, V>> registrations() {
87	return Collections.unmodifiableSet(dispatchQueues.keySet());	×
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) {	×
98	return;	×
99	}
100	var listener = new EventTypeAwareListener<K, V>(	×
101	configuration.getCacheEntryListenerFactory().create());	×
102
103	var factory = configuration.getCacheEntryEventFilterFactory();	×
104	CacheEntryEventFilter<K, V> filter = (factory == null)	×
105	? event -> true	×
106	: new EventTypeFilter<>(listener, factory.create());	×
107
108	var registration = new Registration<>(configuration, filter, listener);	×
109	dispatchQueues.putIfAbsent(registration, new ConcurrentHashMap<>());	×
110	}	×
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);	×
119	var key = new MutableCacheEntryListenerConfiguration<>(configuration);	×
120	dispatchQueues.keySet().removeIf(registration -> key.equals(registration.getConfiguration()));	×
121	}	×
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,	×
132	/* oldValue= / null, / newValue= / value, / quiet= */ false);
133	}	×
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,	×
145	oldValue, newValue, /* quiet= */ false);
146	}	×
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,	×
159	oldValue, newValue, /* quiet= */ true);
160	}	×
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,	×
171	/* oldValue= / value, / newValue= / value, / quiet= */ false);
172	}	×
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,	×
184	/* oldValue= / value, / newValue= / value, / quiet= */ true);
185	}	×
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,	×
196	/* oldValue= / value, / newValue= / value, / quiet= */ false);
197	}	×
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,	×
209	/* oldValue= / value, / newValue= / value, / quiet= */ true);
210	}	×
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	if (pending.get().isEmpty()) {	×
219	return;	×
220	}
221	try {
222	var error = chainSynchronous().join();	×
223	if (error != null) {	×
224	throw error;	×
225	}
226	} catch (CompletionException e) {	×
227	if (e.getCause() instanceof CacheEntryListenerException) {	×
228	throw (CacheEntryListenerException) e.getCause();	×
229	} else if (e.getCause() instanceof Error) {	×
230	throw (Error) e.getCause();	×
231	}
232	throw new CacheEntryListenerException(e);	×
233	}	×
234	}	×
235
236	/**
237	* Ignores and clears the queued futures to the synchronous listeners that are processing events
238	* this thread published.
239	*/
240	public void ignoreSynchronous() {
241	pending.get().clear();	×
242	}	×
243
244	/**
245	* Returns a future that completes once this thread's synchronous listeners have finished
246	* processing the events it published.
247	*/
248	@SuppressWarnings("UnnamedVariable")
249	public CompletableFuture<@Nullable CacheEntryListenerException> chainSynchronous() {
250	var synchronous = pending.get();	×
251	var futures = List.copyOf(synchronous);	×
252	CompletableFuture<@Nullable CacheEntryListenerException> future =	×
253	CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new))	×
254	.thenApply(new Function<@Nullable Void, @Nullable CacheEntryListenerException>() {	×
255	@Override @SuppressWarnings("UnnamedVariable")
256	public @Nullable CacheEntryListenerException apply(@Nullable Void unused) {
257	return futures.stream()	×
258	.map(CompletableFuture::join)	×
259	.filter(Objects::nonNull)	×
260	.reduce((e1, e2) -> {	×
261	if (e1 != e2) {	×
262	e1.addSuppressed(e2);	×
263	}
264	return e1;	×
265	}).orElse(null);	×
266	}
267	});
268	synchronous.clear();	×
269	return future;	×
270	}
271
272	/** Broadcasts the event to the interested listener's dispatch queues. */
273	@SuppressWarnings("FutureReturnValueIgnored")
274	private void publish(Cache<K, V> cache, EventType eventType, K key,
275	boolean hasOldValue, @Nullable V oldValue, @Nullable V newValue, boolean quiet) {
276	if (dispatchQueues.isEmpty()) {	×
277	return;	×
278	}
279
280	@Var JCacheEntryEvent<K, V> event = null;	×
281	for (var entry : dispatchQueues.entrySet()) {	×
282	var registration = entry.getKey();	×
283	if (!registration.getCacheEntryListener().isCompatible(eventType)) {	×
284	continue;	×
285	}
286	if (event == null) {	×
287	event = new JCacheEntryEvent<>(cache, eventType, key, hasOldValue, oldValue, newValue);	×
288	}
289	if (!registration.getCacheEntryFilter().evaluate(event)) {	×
290	continue;	×
291	}
292
293	JCacheEntryEvent<K, V> e = event;	×
294	var dispatchQueue = entry.getValue();	×
295	@SuppressWarnings("PMD.CloseResource")
296	var listener = registration.getCacheEntryListener();	×
297	var future = dispatchQueue.compute(key, (k, queue) -> {	×
298	return (queue == null)	×
299	? CompletableFuture.supplyAsync(() -> listener.dispatch(e), executor)	×
300	: queue.handleAsync((prev, error) -> listener.dispatch(e), executor);	×
301	});
302	future.whenComplete((result, error) -> {	×
303	// optimistic check to avoid locking if not a match
304	if (dispatchQueue.get(key) == future) {	×
305	dispatchQueue.remove(key, future);	×
306	}
307	});	×
308	if (registration.isSynchronous() && !quiet) {	×
309	pending.get().add(future);	×
310	}
311	}	×
312	}	×
313	}

ben-manes / caffeine / #5707

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