Chapter 12. Listeners and notifications
Chapter 12. Listeners and notifications Use listeners with Data Grid to get notifications when events occur for the Cache Manager or for caches. 12.1. Listeners and notifications Data Grid offers a listener API, where clients can register for and get notified when events take place. This annotation-driven API applies to 2 different levels: cache level events and Cache Manager level events. Events trigger a notification which is dispatched to listeners. Listeners are simple POJO s annotated with @Listener and registered using the methods defined in the Listenable interface. Both Cache and CacheManager implement Listenable, which means you can attach listeners to either a cache or a Cache Manager, to receive either cache-level or Cache Manager-level notifications. For example, the following class defines a listener to print out some information every time a new entry is added to the cache, in a non blocking fashion: @Listener public class PrintWhenAdded { Queue<CacheEntryCreatedEvent> events = new ConcurrentLinkedQueue<>(); @CacheEntryCreated public CompletionStage<Void> print(CacheEntryCreatedEvent event) { events.add(event); return null; } } For more comprehensive examples, please see the Javadocs for @Listener . 12.2. Cache-level notifications Cache-level events occur on a per-cache basis, and by default are only raised on nodes where the events occur. Note in a distributed cache these events are only raised on the owners of data being affected. Examples of cache-level events are entries being added, removed, modified, etc. These events trigger notifications to listeners registered to a specific cache. Please see the Javadocs on the org.infinispan.notifications.cachelistener.annotation package for a comprehensive list of all cache-level notifications, and their respective method-level annotations. Note Please refer to the Javadocs on the org.infinispan.notifications.cachelistener.annotation package for the list of cache-level notifications available in Data Grid. Cluster listeners The cluster listeners should be used when it is desirable to listen to the cache events on a single node. To do so all that is required is set to annotate your listener as being clustered. @Listener (clustered = true) public class MyClusterListener { .... } There are some limitations to cluster listeners from a non clustered listener. A cluster listener can only listen to @CacheEntryModified , @CacheEntryCreated , @CacheEntryRemoved and @CacheEntryExpired events. Note this means any other type of event will not be listened to for this listener. Only the post event is sent to a cluster listener, the pre event is ignored. Event filtering and conversion All applicable events on the node where the listener is installed will be raised to the listener. It is possible to dynamically filter what events are raised by using a KeyFilter (only allows filtering on keys) or CacheEventFilter (used to filter for keys, old value, old metadata, new value, new metadata, whether command was retried, if the event is before the event (ie. isPre) and also the command type). The example here shows a simple KeyFilter that will only allow events to be raised when an event modified the entry for the key Only Me . public class SpecificKeyFilter implements KeyFilter<String> { private final String keyToAccept; public SpecificKeyFilter(String keyToAccept) { if (keyToAccept == null) { throw new NullPointerException(); } this.keyToAccept = keyToAccept; } public boolean accept(String key) { return keyToAccept.equals(key); } } ... cache.addListener(listener, new SpecificKeyFilter("Only Me")); ... This can be useful when you want to limit what events you receive in a more efficient manner. There is also a CacheEventConverter that can be supplied that allows for converting a value to another before raising the event. This can be nice to modularize any code that does value conversions. Note The mentioned filters and converters are especially beneficial when used in conjunction with a Cluster Listener. This is because the filtering and conversion is done on the node where the event originated and not on the node where event is listened to. This can provide benefits of not having to replicate events across the cluster (filter) or even have reduced payloads (converter). Initial State Events When a listener is installed it will only be notified of events after it is fully installed. It may be desirable to get the current state of the cache contents upon first registration of listener by having an event generated of type @CacheEntryCreated for each element in the cache. Any additionally generated events during this initial phase will be queued until appropriate events have been raised. Note This only works for clustered listeners at this time. ISPN-4608 covers adding this for non clustered listeners. Duplicate Events It is possible in a non transactional cache to receive duplicate events. This is possible when the primary owner of a key goes down while trying to perform a write operation such as a put. Data Grid internally will rectify the put operation by sending it to the new primary owner for the given key automatically, however there are no guarantees in regards to if the write was first replicated to backups. Thus more than 1 of the following write events ( CacheEntryCreatedEvent , CacheEntryModifiedEvent & CacheEntryRemovedEvent ) may be sent on a single operation. If more than one event is generated Data Grid will mark the event that it was generated by a retried command to help the user to know when this occurs without having to pay attention to view changes. @Listener public class MyRetryListener { @CacheEntryModified public void entryModified(CacheEntryModifiedEvent event) { if (event.isCommandRetried()) { // Do something } } } Also when using a CacheEventFilter or CacheEventConverter the EventType contains a method isRetry to tell if the event was generated due to retry. 12.3. Cache Manager notifications Events that occur on a Cache Manager level are cluster-wide and involve events that affect all caches created by a single Cache Manager. Examples of Cache Manager events are nodes joining or leaving a cluster, or caches starting or stopping. See the org.infinispan.notifications.cachemanagerlistener.annotation package for a comprehensive list of all Cache Manager notifications, and their respective method-level annotations. 12.4. Synchronicity of events By default, all async notifications are dispatched in the notification thread pool. Sync notifications will delay the operation from continuing until the listener method completes or the CompletionStage completes (the former causing the thread to block). Alternatively, you could annotate your listener as asynchronous in which case the operation will continue immediately, while the notification is completed asynchronously on the notification thread pool. To do this, simply annotate your listener such: Asynchronous Listener @Listener (sync = false) public class MyAsyncListener { @CacheEntryCreated void listen(CacheEntryCreatedEvent event) { } } Blocking Synchronous Listener @Listener public class MySyncListener { @CacheEntryCreated void listen(CacheEntryCreatedEvent event) { } } Non-Blocking Listener @Listener public class MyNonBlockingListener { @CacheEntryCreated CompletionStage<Void> listen(CacheEntryCreatedEvent event) { } } Asynchronous thread pool To tune the thread pool used to dispatch such asynchronous notifications, use the <listener-executor /> XML element in your configuration file.
@Listener public class PrintWhenAdded { Queue<CacheEntryCreatedEvent> events = new ConcurrentLinkedQueue<>(); @CacheEntryCreated public CompletionStage<Void> print(CacheEntryCreatedEvent event) { events.add(event); return null; } } @Listener (clustered = true) public class MyClusterListener { .... } public class SpecificKeyFilter implements KeyFilter<String> { private final String keyToAccept; public SpecificKeyFilter(String keyToAccept) { if (keyToAccept == null) { throw new NullPointerException(); } this.keyToAccept = keyToAccept; } public boolean accept(String key) { return keyToAccept.equals(key); } } cache.addListener(listener, new SpecificKeyFilter("Only Me")); @Listener public class MyRetryListener { @CacheEntryModified public void entryModified(CacheEntryModifiedEvent event) { if (event.isCommandRetried()) { // Do something } } } @Listener (sync = false) public class MyAsyncListener { @CacheEntryCreated void listen(CacheEntryCreatedEvent event) { } } @Listener public class MySyncListener { @CacheEntryCreated void listen(CacheEntryCreatedEvent event) { } } @Listener public class MyNonBlockingListener { @CacheEntryCreated CompletionStage<Void> listen(CacheEntryCreatedEvent event) { } }