Fixes ehcache#3169: Register only those CacheEvents with correspondin…

…g listener
SamuelBussmann · Jul 18, 2024 · 14ec4e1 · 14ec4e1
1 parent 155fb3d
commit 14ec4e1
Show file tree

Hide file tree

Showing 16 changed files with 131 additions and 19 deletions.
diff --git a/...ache-client/src/main/java/org/ehcache/clustered/client/internal/store/ClusteredStore.java b/...ache-client/src/main/java/org/ehcache/clustered/client/internal/store/ClusteredStore.java
@@ -1016,6 +1016,12 @@ public synchronized void removeEventListener(StoreEventListener<K, V> eventListe
       }
       delegate.removeEventListener(eventListener);
     }
+
+    @Override
+    public void listenerModified() {
+      delegate.listenerModified();
+    }
+
     @Override
     public void addEventFilter(StoreEventFilter<K, V> eventFilter) {
       delegate.addEventFilter(eventFilter);

diff --git a/ehcache-api/src/main/java/org/ehcache/event/EventType.java b/ehcache-api/src/main/java/org/ehcache/event/EventType.java
@@ -16,6 +16,9 @@
 
 package org.ehcache.event;
 
+import java.util.EnumSet;
+import java.util.Set;
+
 /**
  * The different event types.
  */
@@ -44,6 +47,11 @@ public enum EventType {
   /**
    * Represents an existing {@link org.ehcache.Cache.Entry cache entry} being updated for a given key
    */
-  UPDATED,
+  UPDATED;
+
+  private static final Set<EventType> ALL_EVENT_TYPES = EnumSet.allOf(EventType.class);
 
+  public static Set<EventType> allAsSet() {
+    return ALL_EVENT_TYPES;
+  }
 }
diff --git a/ehcache-core/src/main/java/org/ehcache/core/events/NullStoreEventDispatcher.java b/ehcache-core/src/main/java/org/ehcache/core/events/NullStoreEventDispatcher.java
@@ -87,6 +87,11 @@ public void removeEventListener(StoreEventListener<K, V> eventListener) {
     // Do nothing
   }
 
+  @Override
+  public void listenerModified() {
+    // Do nothing
+  }
+
   @Override
   public void addEventFilter(StoreEventFilter<K, V> eventFilter) {
     // Do nothing

diff --git a/ehcache-core/src/main/java/org/ehcache/core/spi/store/events/StoreEventListener.java b/ehcache-core/src/main/java/org/ehcache/core/spi/store/events/StoreEventListener.java
@@ -16,7 +16,10 @@
 
 package org.ehcache.core.spi.store.events;
 
+import java.util.Set;
+
 import org.ehcache.core.events.StoreEventDispatcher;
+import org.ehcache.event.EventType;
 
 /**
  * Interface used to register on a {@link StoreEventSource} to get notified of events happening to mappings the
@@ -36,4 +39,15 @@ public interface StoreEventListener<K, V> {
    * @param event the actual {@link StoreEvent}
    */
   void onEvent(StoreEvent<K, V> event);
+
+  /**
+   * Specify which Events this Listener is handling.
+   * <p>
+   * Defaults return is all values of {@link EventType}
+   *
+   * @return Set of the {@link EventType} this listener handles.
+   */
+  default Set<EventType> getEventTypes() {
+    return EventType.allAsSet();
+  }
 }
diff --git a/ehcache-core/src/main/java/org/ehcache/core/spi/store/events/StoreEventSource.java b/ehcache-core/src/main/java/org/ehcache/core/spi/store/events/StoreEventSource.java
@@ -58,4 +58,9 @@ public interface StoreEventSource<K, V> {
    * @return {@code true} if ordering is on, {@code false} otherwise
    */
   boolean isEventOrdering();
+
+  /**
+   * Indicates that a listener was modified
+   */
+  void listenerModified();
 }
diff --git a/ehcache-impl/src/main/java/org/ehcache/impl/events/CacheEventDispatcherImpl.java b/ehcache-impl/src/main/java/org/ehcache/impl/events/CacheEventDispatcherImpl.java
@@ -39,6 +39,7 @@
 import java.util.EnumSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
@@ -70,6 +71,7 @@ public class CacheEventDispatcherImpl<K, V> implements CacheEventDispatcher<K, V
     .collect(toMap(identity(), t -> new CopyOnWriteArrayList<>(), (a, b) -> { throw new AssertionError(); }, () -> new EnumMap<>(EventType.class))));
   private final Map<EventType, List<EventListenerWrapper<K, V>>> asyncListenersList = unmodifiableMap(allOf(EventType.class).stream()
     .collect(toMap(identity(), t -> new CopyOnWriteArrayList<>(), (a, b) -> { throw new AssertionError(); }, () -> new EnumMap<>(EventType.class))));
+  private final Set<EventType> registeredEventTypes = EnumSet.noneOf(EventType.class);
 
   private final StoreEventListener<K, V> eventListener = new StoreListener();
 
@@ -117,6 +119,8 @@ private synchronized void registerCacheEventListener(EventListenerWrapper<K, V>
       storeEventSource.setEventOrdering(true);
     }
 
+    registeredEventTypes.addAll(wrapper.getEventTypes()); // add EventType of new wrapper to list or relevant EntryTypes
+
     switch (wrapper.getFiringMode()) {
       case ASYNCHRONOUS:
         wrapper.getEventTypes().forEach(type -> asyncListenersList.get(type).add(wrapper));
@@ -133,6 +137,8 @@ private synchronized void registerCacheEventListener(EventListenerWrapper<K, V>
 
     if (firstListener) {
       storeEventSource.addEventListener(eventListener);
+    } else {
+      storeEventSource.listenerModified();
     }
   }
 
@@ -164,6 +170,8 @@ public synchronized  void deregisterCacheEventListener(CacheEventListener<? supe
       throw new IllegalStateException("Unknown cache event listener: " + listener);
     }
 
+    refreshRegisteredEventTypes();
+
     if (!allListeners().findAny().isPresent()) {
       storeEventSource.setSynchronous(false);
       storeEventSource.setEventOrdering(false);
@@ -178,6 +186,14 @@ public synchronized  void deregisterCacheEventListener(CacheEventListener<? supe
     }
   }
 
+  private void refreshRegisteredEventTypes() {
+    // collect all registered EventTypes
+    EnumSet<EventType> newRegisteredEventTypes = EnumSet.noneOf(EventType.class);
+    allListeners().forEach(listener -> newRegisteredEventTypes.addAll(listener.getEventTypes()));
+    // drop irrelevant EventTypes
+    registeredEventTypes.retainAll(newRegisteredEventTypes);
+  }
+
   /**
    * {@inheritDoc}
    */
@@ -264,8 +280,12 @@ public void onEvent(StoreEvent<K, V> event) {
           throw new AssertionError("Unexpected StoreEvent value: " + event.getType());
       }
     }
-  }
 
+    @Override
+    public Set<EventType> getEventTypes() {
+      return registeredEventTypes;
+    }
+  }
   /**
    * {@inheritDoc}
    */

diff --git a/...che-impl/src/main/java/org/ehcache/impl/internal/events/AbstractStoreEventDispatcher.java b/...che-impl/src/main/java/org/ehcache/impl/internal/events/AbstractStoreEventDispatcher.java
@@ -20,7 +20,9 @@
 import org.ehcache.core.events.StoreEventSink;
 import org.ehcache.core.spi.store.events.StoreEventFilter;
 import org.ehcache.core.spi.store.events.StoreEventListener;
+import org.ehcache.event.EventType;
 
+import java.util.EnumSet;
 import java.util.Set;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CopyOnWriteArraySet;
@@ -77,6 +79,7 @@ public void evicted(Object key, Supplier<Object> value) {
   private final Set<StoreEventFilter<K, V>> filters = new CopyOnWriteArraySet<>();
   private final Set<StoreEventListener<K, V>> listeners = new CopyOnWriteArraySet<>();
   private final BlockingQueue<FireableStoreEventHolder<K, V>>[] orderedQueues;
+  private final Set<EventType> relevantEventTypes = EnumSet.noneOf(EventType.class);
   private volatile boolean ordered = false;
 
   protected AbstractStoreEventDispatcher(int dispatcherConcurrency) {
@@ -92,6 +95,16 @@ protected AbstractStoreEventDispatcher(int dispatcherConcurrency) {
     }
   }
 
+  private void computeRelevantEventTypes() {
+    // collect all EventTypes the listeners are interested in.
+    for (StoreEventListener<K, V> listener : listeners) {
+      relevantEventTypes.addAll(listener.getEventTypes());
+    }
+    if (relevantEventTypes.isEmpty()) { // mocks are empty -> handle all types
+      relevantEventTypes.addAll(EnumSet.allOf(EventType.class));
+    }
+  }
+
   protected Set<StoreEventListener<K, V>> getListeners() {
     return listeners;
   }
@@ -104,14 +117,25 @@ protected BlockingQueue<FireableStoreEventHolder<K, V>>[] getOrderedQueues() {
     return orderedQueues;
   }
 
+  protected Set<EventType> getRelevantEventTypes() {
+    return relevantEventTypes;
+  }
+
   @Override
   public void addEventListener(StoreEventListener<K, V> eventListener) {
     listeners.add(eventListener);
+    computeRelevantEventTypes(); // refresh
   }
 
   @Override
   public void removeEventListener(StoreEventListener<K, V> eventListener) {
     listeners.remove(eventListener);
+    computeRelevantEventTypes(); // refresh
+  }
+
+  @Override
+  public void listenerModified() {
+    computeRelevantEventTypes(); // refresh
   }
 
   @Override
@@ -151,6 +175,6 @@ public void reset(StoreEventSink<K, V> eventSink) {
 
   @Override
   public StoreEventSink<K, V> eventSink() {
-    return new InvocationScopedEventSink<>(getFilters(), isEventOrdering(), getOrderedQueues(), getListeners());
+    return new InvocationScopedEventSink<>(getFilters(), isEventOrdering(), getOrderedQueues(), getListeners(), getRelevantEventTypes());
   }
 }
diff --git a/...impl/src/main/java/org/ehcache/impl/internal/events/FudgingInvocationScopedEventSink.java b/...impl/src/main/java/org/ehcache/impl/internal/events/FudgingInvocationScopedEventSink.java
@@ -40,8 +40,9 @@ class FudgingInvocationScopedEventSink<K, V> extends InvocationScopedEventSink<K
 
   FudgingInvocationScopedEventSink(Set<StoreEventFilter<K, V>> filters, boolean ordered,
                                    BlockingQueue<FireableStoreEventHolder<K, V>>[] orderedQueues,
-                                   Set<StoreEventListener<K, V>> listeners) {
-    super(filters, ordered, orderedQueues, listeners);
+                                   Set<StoreEventListener<K, V>> listeners,
+                                   Set<EventType> relevantEventTypes) {
+    super(filters, ordered, orderedQueues, listeners, relevantEventTypes);
   }
 
   @Override

diff --git a/ehcache-impl/src/main/java/org/ehcache/impl/internal/events/InvocationScopedEventSink.java b/ehcache-impl/src/main/java/org/ehcache/impl/internal/events/InvocationScopedEventSink.java
@@ -16,9 +16,11 @@
 
 package org.ehcache.impl.internal.events;
 
-import org.ehcache.event.EventType;
-import org.ehcache.core.spi.store.events.StoreEventFilter;
-import org.ehcache.core.spi.store.events.StoreEventListener;
+import static org.ehcache.impl.internal.events.StoreEvents.createEvent;
+import static org.ehcache.impl.internal.events.StoreEvents.evictEvent;
+import static org.ehcache.impl.internal.events.StoreEvents.expireEvent;
+import static org.ehcache.impl.internal.events.StoreEvents.removeEvent;
+import static org.ehcache.impl.internal.events.StoreEvents.updateEvent;
 
 import java.util.ArrayDeque;
 import java.util.Deque;
@@ -27,11 +29,9 @@
 import java.util.concurrent.BlockingQueue;
 import java.util.function.Supplier;
 
-import static org.ehcache.impl.internal.events.StoreEvents.createEvent;
-import static org.ehcache.impl.internal.events.StoreEvents.evictEvent;
-import static org.ehcache.impl.internal.events.StoreEvents.expireEvent;
-import static org.ehcache.impl.internal.events.StoreEvents.removeEvent;
-import static org.ehcache.impl.internal.events.StoreEvents.updateEvent;
+import org.ehcache.core.spi.store.events.StoreEventFilter;
+import org.ehcache.core.spi.store.events.StoreEventListener;
+import org.ehcache.event.EventType;
 
 /**
  * InvocationScopedEventSink
@@ -44,13 +44,16 @@ class InvocationScopedEventSink<K, V> implements CloseableStoreEventSink<K, V> {
   private final Set<StoreEventListener<K, V>> listeners;
   private final Deque<FireableStoreEventHolder<K, V>> events = new ArrayDeque<>(4);
 
+  private final Set<EventType> relevantEventTypes;
+
   InvocationScopedEventSink(Set<StoreEventFilter<K, V>> filters, boolean ordered,
-                            BlockingQueue<FireableStoreEventHolder<K, V>>[] orderedQueues,
-                            Set<StoreEventListener<K, V>> listeners) {
+      BlockingQueue<FireableStoreEventHolder<K, V>>[] orderedQueues, Set<StoreEventListener<K, V>> listeners,
+      Set<EventType> relevantEventTypes) {
     this.filters = filters;
     this.ordered = ordered;
     this.orderedQueues = orderedQueues;
     this.listeners = listeners;
+    this.relevantEventTypes = relevantEventTypes;
   }
 
   @Override
@@ -98,7 +101,8 @@ protected boolean acceptEvent(EventType type, K key, V oldValue, V newValue) {
         return false;
       }
     }
-    return true;
+    // at least one listener is interested in this event
+    return relevantEventTypes.contains(type);
   }
 
   @Override

diff --git a/...-impl/src/main/java/org/ehcache/impl/internal/events/ThreadLocalStoreEventDispatcher.java b/...-impl/src/main/java/org/ehcache/impl/internal/events/ThreadLocalStoreEventDispatcher.java
@@ -39,7 +39,7 @@ public StoreEventSink<K, V> eventSink() {
     } else {
       StoreEventSink<K, V> eventSink = tlEventSink.get();
       if (eventSink == null) {
-        eventSink = new FudgingInvocationScopedEventSink<>(getFilters(), isEventOrdering(), getOrderedQueues(), getListeners());
+        eventSink = new FudgingInvocationScopedEventSink<>(getFilters(), isEventOrdering(), getOrderedQueues(), getListeners(), getRelevantEventTypes());
         tlEventSink.set(eventSink);
         usageDepth.set(0);
       } else {

diff --git a/ehcache-impl/src/main/java/org/ehcache/impl/internal/store/basic/NopStore.java b/ehcache-impl/src/main/java/org/ehcache/impl/internal/store/basic/NopStore.java
@@ -144,6 +144,10 @@ public void setSynchronous(boolean synchronous) throws IllegalArgumentException
       public boolean isEventOrdering() {
         return false;
       }
+
+      @Override
+      public void listenerModified() {
+      }
     };
   }
 

diff --git a/.../src/test/java/org/ehcache/impl/internal/events/FudgingInvocationScopedEventSinkTest.java b/.../src/test/java/org/ehcache/impl/internal/events/FudgingInvocationScopedEventSinkTest.java
@@ -24,6 +24,7 @@
 import org.junit.Test;
 import org.mockito.InOrder;
 
+import java.util.EnumSet;
 import java.util.HashSet;
 import java.util.Set;
 import java.util.concurrent.ArrayBlockingQueue;
@@ -53,7 +54,7 @@ public void setUp() {
     storeEventListeners.add(listener);
     @SuppressWarnings({"unchecked", "rawtypes"})
     BlockingQueue<FireableStoreEventHolder<String, String>>[] blockingQueues = new BlockingQueue[] { new ArrayBlockingQueue<FireableStoreEventHolder<String, String>>(10) };
-    eventSink = new FudgingInvocationScopedEventSink<>(new HashSet<>(), false, blockingQueues, storeEventListeners);
+    eventSink = new FudgingInvocationScopedEventSink<>(new HashSet<>(), false, blockingQueues, storeEventListeners, EnumSet.allOf(EventType.class));
   }
 
   @Test
@@ -63,6 +64,7 @@ public void testEvictedDifferentKeyNoImpact() {
     eventSink.close();
 
     InOrder inOrder = inOrder(listener);
+    inOrder.verify(listener, times(2)).getEventTypes();
     inOrder.verify(listener).onEvent(argThat(createdMatcher));
     inOrder.verify(listener).onEvent(argThat(evictedMatcher));
     verifyNoMoreInteractions(listener);
@@ -75,6 +77,7 @@ public void testEvictedSameKeyAfterUpdateReplacesWithEvictCreate() {
     eventSink.close();
 
     InOrder inOrder = inOrder(listener);
+    inOrder.verify(listener, times(3)).getEventTypes();
     inOrder.verify(listener).onEvent(argThat(evictedMatcher));
     inOrder.verify(listener).onEvent(argThat(createdMatcher));
     verifyNoMoreInteractions(listener);
@@ -88,6 +91,7 @@ public void testEvictedSameKeyAfterCreateFudgesExpiryToo() {
     eventSink.close();
 
     InOrder inOrder = inOrder(listener);
+    inOrder.verify(listener, times(4)).getEventTypes();
     inOrder.verify(listener).onEvent(argThat(evictedMatcher));
     inOrder.verify(listener).onEvent(argThat(createdMatcher));
     verifyNoMoreInteractions(listener);
@@ -102,6 +106,7 @@ public void testEvictedSameKeyAfterUpdateReplacesWithEvictCreateEvenWithMultiple
     eventSink.close();
 
     InOrder inOrder = inOrder(listener);
+    inOrder.verify(listener, times(5)).getEventTypes();
     inOrder.verify(listener, times(3)).onEvent(argThat(evictedMatcher));
     inOrder.verify(listener).onEvent(argThat(createdMatcher));
     verifyNoMoreInteractions(listener);
@@ -117,6 +122,7 @@ public void testEvictedSameKeyAfterCreateFudgesExpiryTooEvenWithMultipleEvictsIn
     eventSink.close();
 
     InOrder inOrder = inOrder(listener);
+    inOrder.verify(listener, times(6)).getEventTypes();
     inOrder.verify(listener, times(3)).onEvent(argThat(evictedMatcher));
     inOrder.verify(listener).onEvent(argThat(createdMatcher));
     verifyNoMoreInteractions(listener);
@@ -130,6 +136,7 @@ public void testEvictedKeyDoesNotFudgeOlderEvents() {
     eventSink.close();
 
     InOrder inOrder = inOrder(listener);
+    inOrder.verify(listener, times(3)).getEventTypes();
     Matcher<StoreEvent<String, String>> updatedMatcher = eventType(EventType.UPDATED);
     inOrder.verify(listener).onEvent(argThat(updatedMatcher));
     inOrder.verify(listener).onEvent(argThat(createdMatcher));