Event Handling

/**
 * Manages and dispatches JCache events to registered listeners
 */
public final class EventDispatcher<K, V> {
    
    /**
     * Register a cache entry listener configuration
     * @param cacheEntryListenerConfiguration the listener configuration to register
     */
    public void register(CacheEntryListenerConfiguration<K, V> cacheEntryListenerConfiguration);
    
    /**
     * Deregister a cache entry listener configuration
     * @param cacheEntryListenerConfiguration the listener configuration to remove
     */
    public void deregister(CacheEntryListenerConfiguration<K, V> cacheEntryListenerConfiguration);
    
    /**
     * Publish entry created event to all registered listeners
     * @param cache the cache where the event occurred
     * @param key the key of the created entry
     * @param value the value of the created entry
     */
    public void publishCreated(Cache<K, V> cache, K key, V value);
    
    /**
     * Publish entry updated event to all registered listeners
     * @param cache the cache where the event occurred
     * @param key the key of the updated entry
     * @param oldValue the previous value
     * @param newValue the new value
     */
    public void publishUpdated(Cache<K, V> cache, K key, V oldValue, V newValue);
    
    /**
     * Publish entry removed event to all registered listeners
     * @param cache the cache where the event occurred
     * @param key the key of the removed entry
     * @param value the value of the removed entry
     */
    public void publishRemoved(Cache<K, V> cache, K key, V value);
    
    /**
     * Publish entry removed event quietly (no error propagation)
     * @param cache the cache where the event occurred
     * @param key the key of the removed entry
     * @param value the value of the removed entry
     */
    public void publishRemovedQuietly(Cache<K, V> cache, K key, V value);
    
    /**
     * Publish entry expired event to all registered listeners
     * @param cache the cache where the event occurred
     * @param key the key of the expired entry
     * @param value the value of the expired entry
     */
    public void publishExpired(Cache<K, V> cache, K key, V value);
    
    /**
     * Publish entry expired event quietly (no error propagation)
     * @param cache the cache where the event occurred
     * @param key the key of the expired entry
     * @param value the value of the expired entry
     */
    public void publishExpiredQuietly(Cache<K, V> cache, K key, V value);
    
    /**
     * Wait for all synchronous listeners to complete processing
     */
    public void awaitSynchronous();
    
    /**
     * Ignore completion of synchronous listeners (used during error handling)
     */
    public void ignoreSynchronous();
    
    /**
     * Get all active listener registrations
     * @return set of current registrations
     */
    public Set<Registration<K, V>> registrations();
}

Cache<String, String> cache = cacheManager.getCache("events", String.class, String.class);

// Create listener
CacheEntryListener<String, String> listener = new CacheEntryCreatedListener<String, String>() {
    @Override
    public void onCreated(Iterable<CacheEntryEvent<? extends String, ? extends String>> events) {
        for (CacheEntryEvent<? extends String, ? extends String> event : events) {
            System.out.println("Entry created: " + event.getKey() + " = " + event.getValue());
        }
    }
};

// Register listener
CacheEntryListenerConfiguration<String, String> listenerConfig = 
    new MutableCacheEntryListenerConfiguration<>(
        FactoryBuilder.factoryOf(listener),
        null, // No filter
        false, // Not synchronous
        true   // Include old value
    );

cache.registerCacheEntryListener(listenerConfig);

// Cache operations will trigger events
cache.put("key1", "value1"); // Triggers created event
cache.put("key1", "value2"); // Triggers updated event
cache.remove("key1");        // Triggers removed event

/**
 * Represents a cache entry listener registration
 */
public final class Registration<K, V> {
    
    /**
     * Get the registered cache entry listener
     * @return the CacheEntryListener instance
     */
    public CacheEntryListener<? super K, ? super V> getCacheEntryListener();
    
    /**
     * Get the entry filter for this registration
     * @return the CacheEntryFilter or null if none configured
     */
    public CacheEntryFilter<? super K, ? super V> getCacheEntryFilter();
    
    /**
     * Get the original listener configuration
     * @return the CacheEntryListenerConfiguration
     */
    public CacheEntryListenerConfiguration<K, V> getConfiguration();
    
    /**
     * Check if this registration should receive events of the specified type
     * @param eventType the event type to check
     * @return true if this registration matches the event type
     */
    public boolean matches(EventType eventType);
}

// Event types
enum EventType {
    CREATED, UPDATED, REMOVED, EXPIRED
}

// Listener interfaces
interface CacheEntryCreatedListener<K, V> extends CacheEntryListener<K, V> {
    void onCreated(Iterable<CacheEntryEvent<? extends K, ? extends V>> events);
}

interface CacheEntryUpdatedListener<K, V> extends CacheEntryListener<K, V> {
    void onUpdated(Iterable<CacheEntryEvent<? extends K, ? extends V>> events);
}

interface CacheEntryRemovedListener<K, V> extends CacheEntryListener<K, V> {
    void onRemoved(Iterable<CacheEntryEvent<? extends K, ? extends V>> events);
}

interface CacheEntryExpiredListener<K, V> extends CacheEntryListener<K, V> {
    void onExpired(Iterable<CacheEntryEvent<? extends K, ? extends V>> events);
}

// Multi-event listener
CacheEntryListener<String, String> multiListener = new CacheEntryListener<String, String>() 
    implements CacheEntryCreatedListener<String, String>,
               CacheEntryUpdatedListener<String, String>,
               CacheEntryRemovedListener<String, String> {
    
    @Override
    public void onCreated(Iterable<CacheEntryEvent<? extends String, ? extends String>> events) {
        events.forEach(event -> 
            System.out.println("Created: " + event.getKey() + " = " + event.getValue()));
    }
    
    @Override
    public void onUpdated(Iterable<CacheEntryEvent<? extends String, ? extends String>> events) {
        events.forEach(event -> 
            System.out.println("Updated: " + event.getKey() + " from " + 
                event.getOldValue() + " to " + event.getValue()));
    }
    
    @Override
    public void onRemoved(Iterable<CacheEntryEvent<? extends String, ? extends String>> events) {
        events.forEach(event -> 
            System.out.println("Removed: " + event.getKey() + " = " + event.getOldValue()));
    }
};

/**
 * Filter for cache entry events
 */
interface CacheEntryEventFilter<K, V> extends CacheEntryFilter<K, V> {
    /**
     * Evaluate whether an event should be delivered to the listener
     * @param event the cache entry event
     * @return true if the event should be delivered
     */
    boolean evaluate(CacheEntryEvent<? extends K, ? extends V> event);
}

// Filter to only receive events for keys starting with "user:"
CacheEntryEventFilter<String, String> userFilter = event -> 
    event.getKey().startsWith("user:");

// Register filtered listener
CacheEntryListenerConfiguration<String, String> filteredConfig = 
    new MutableCacheEntryListenerConfiguration<>(
        FactoryBuilder.factoryOf(listener),
        FactoryBuilder.factoryOf(userFilter),
        false, // Not synchronous
        true   // Include old value
    );

cache.registerCacheEntryListener(filteredConfig);

cache.put("user:123", "John");    // Event delivered (matches filter)
cache.put("session:456", "abc");  // Event filtered out

// Synchronous listener configuration
CacheEntryListenerConfiguration<String, String> syncConfig = 
    new MutableCacheEntryListenerConfiguration<>(
        FactoryBuilder.factoryOf(listener),
        null,
        true,  // Synchronous execution
        false
    );

// Asynchronous listener configuration  
CacheEntryListenerConfiguration<String, String> asyncConfig = 
    new MutableCacheEntryListenerConfiguration<>(
        FactoryBuilder.factoryOf(listener),
        null,
        false, // Asynchronous execution
        false
    );

// Synchronous listener for critical operations
CacheEntryListener<String, String> auditListener = new CacheEntryCreatedListener<String, String>() {
    @Override
    public void onCreated(Iterable<CacheEntryEvent<? extends String, ? extends String>> events) {
        // Critical audit logging - must complete before cache operation returns
        for (CacheEntryEvent<? extends String, ? extends String> event : events) {
            auditLog.record("Cache entry created: " + event.getKey());
        }
    }
};

CacheEntryListenerConfiguration<String, String> auditConfig = 
    new MutableCacheEntryListenerConfiguration<>(
        FactoryBuilder.factoryOf(auditListener),
        null,
        true,  // Synchronous - blocks cache operation until complete
        false
    );

cache.registerCacheEntryListener(auditConfig);

// Asynchronous listener for notifications
CacheEntryListener<String, String> notificationListener = new CacheEntryCreatedListener<String, String>() {
    @Override
    public void onCreated(Iterable<CacheEntryEvent<? extends String, ? extends String>> events) {
        // Send notifications asynchronously - doesn't block cache operations
        CompletableFuture.runAsync(() -> {
            for (CacheEntryEvent<? extends String, ? extends String> event : events) {
                notificationService.sendNotification("New cache entry: " + event.getKey());
            }
        });
    }
};

CacheEntryListenerConfiguration<String, String> notificationConfig = 
    new MutableCacheEntryListenerConfiguration<>(
        FactoryBuilder.factoryOf(notificationListener),
        null,
        false, // Asynchronous - doesn't block cache operations
        false
    );

cache.registerCacheEntryListener(notificationConfig);

/**
 * Bridges Caffeine removal events to JCache event system
 */
public final class JCacheEvictionListener<K, V> implements RemovalListener<K, Expirable<V>> {
    /**
     * Handle removal notification from Caffeine cache
     * @param key the removed key
     * @param expirable the removed value wrapper
     * @param cause the cause of removal
     */
    @Override
    public void onRemoval(K key, Expirable<V> expirable, RemovalCause cause);
}

// Configure listeners in cache configuration
CaffeineConfiguration<String, String> config = new CaffeineConfiguration<String, String>()
    .setTypes(String.class, String.class)
    .addCacheEntryListenerConfiguration(
        new MutableCacheEntryListenerConfiguration<>(
            FactoryBuilder.factoryOf(createdListener),
            null, false, false))
    .addCacheEntryListenerConfiguration(
        new MutableCacheEntryListenerConfiguration<>(
            FactoryBuilder.factoryOf(updatedListener),
            FactoryBuilder.factoryOf(userFilter),
            true, true));

Cache<String, String> cache = cacheManager.createCache("eventCache", config);

CacheEntryListener<String, String> errorHandlingListener = new CacheEntryCreatedListener<String, String>() {
    @Override
    public void onCreated(Iterable<CacheEntryEvent<? extends String, ? extends String>> events) {
        for (CacheEntryEvent<? extends String, ? extends String> event : events) {
            try {
                processEvent(event);
            } catch (Exception e) {
                // Log error but don't let it propagate to cache operation
                logger.error("Error processing cache event for key: " + event.getKey(), e);
            }
        }
    }
    
    private void processEvent(CacheEntryEvent<? extends String, ? extends String> event) {
        // Event processing logic that might fail
    }
};

CacheEntryListener<String, String> batchListener = new CacheEntryCreatedListener<String, String>() {
    @Override
    public void onCreated(Iterable<CacheEntryEvent<? extends String, ? extends String>> events) {
        List<CacheEntryEvent<? extends String, ? extends String>> eventList = 
            StreamSupport.stream(events.spliterator(), false)
                .collect(Collectors.toList());
        
        System.out.println("Processing batch of " + eventList.size() + " events");
        
        // Process all events in the batch
        eventList.forEach(this::processEvent);
    }
    
    private void processEvent(CacheEntryEvent<? extends String, ? extends String> event) {
        // Process single event
    }
};