// Path: headlands/src/main/java/com/airhacks/headlands/notifications/entity/CacheChangedEvent.java
// public class CacheChangedEvent {
// 
//     private EventType type;
//     private String cacheName;
//     private JsonArrayBuilder payload;
// 
//     public CacheChangedEvent(String cacheName, EventType type) {
//         this.type = type;
//         this.cacheName = cacheName;
//         this.payload = Json.createArrayBuilder();
//     }
// 
//     public void consume(CacheEntryEvent<? extends String, ? extends String> event) {
//         String key = event.getKey();
//         String value = event.getValue();
//         String oldValue = event.getOldValue();
//         JsonObjectBuilder builder = Json.createObjectBuilder().add("newValue", value);
//         if (oldValue != null) {
//             builder.add("oldValue", oldValue);
//         }
//         JsonObject diff = builder.build();
// 
//         JsonObject changeSet = Json.createObjectBuilder().
//                 add("cacheName", cacheName).
//                 add("eventType", type.name()).
//                 add("key", key).
//                 add(key, diff).
//                 build();
//         payload.add(changeSet);
//     }
// 
//     public JsonArray getPayload() {
//         return payload.build();
//     }
// 
//     public String getCacheName() {
//         return cacheName;
//     }
// 
// }


import com.airhacks.headlands.notifications.entity.CacheChangedEvent;
import java.io.Serializable;
import java.util.function.Consumer;
import javax.cache.event.CacheEntryCreatedListener;
import javax.cache.event.CacheEntryEvent;
import javax.cache.event.CacheEntryListenerException;
import javax.cache.event.CacheEntryRemovedListener;
import javax.cache.event.CacheEntryUpdatedListener;
import javax.cache.event.EventType;
package com.airhacks.headlands.notifications.control;


/**
 *
 * @author airhacks.com
 */
public class CacheEntryChangedListener implements CacheEntryCreatedListener<String, String>,
        CacheEntryUpdatedListener<String, String>,
        CacheEntryRemovedListener<String, String>,
        Serializable {

    private final transient Consumer<CacheChangedEvent> sink;
    private final String cacheName;

    public CacheEntryChangedListener(String cacheName, Consumer<CacheChangedEvent> sink) {
        this.cacheName = cacheName;
        this.sink = sink;
    }

    @Override
    public void onCreated(Iterable<CacheEntryEvent<? extends String, ? extends String>> events) throws CacheEntryListenerException {
        this.sink.accept(convert(EventType.CREATED, events));
    }

    @Override
    public void onUpdated(Iterable<CacheEntryEvent<? extends String, ? extends String>> events) throws CacheEntryListenerException {
        this.sink.accept(convert(EventType.UPDATED, events));
    }

    @Override
    public void onRemoved(Iterable<CacheEntryEvent<? extends String, ? extends String>> events) throws CacheEntryListenerException {
        this.sink.accept(convert(EventType.REMOVED, events));
    }

    CacheChangedEvent convert(EventType type, Iterable<CacheEntryEvent<? extends String, ? extends String>> events) {
// Path: headlands/src/main/java/com/airhacks/headlands/notifications/entity/CacheChangedEvent.java
// public class CacheChangedEvent {
// 
//     private EventType type;
//     private String cacheName;
//     private JsonArrayBuilder payload;
// 
//     public CacheChangedEvent(String cacheName, EventType type) {
//         this.type = type;
//         this.cacheName = cacheName;
//         this.payload = Json.createArrayBuilder();
//     }
// 
//     public void consume(CacheEntryEvent<? extends String, ? extends String> event) {
//         String key = event.getKey();
//         String value = event.getValue();
//         String oldValue = event.getOldValue();
//         JsonObjectBuilder builder = Json.createObjectBuilder().add("newValue", value);
//         if (oldValue != null) {
//             builder.add("oldValue", oldValue);
//         }
//         JsonObject diff = builder.build();
// 
//         JsonObject changeSet = Json.createObjectBuilder().
//                 add("cacheName", cacheName).
//                 add("eventType", type.name()).
//                 add("key", key).
//                 add(key, diff).
//                 build();
//         payload.add(changeSet);
//     }
// 
//     public JsonArray getPayload() {
//         return payload.build();
//     }
// 
//     public String getCacheName() {
//         return cacheName;
//     }
// 
// }


// Path: headlands/src/main/java/com/airhacks/headlands/notifications/control/CacheEntryChangedListener.java
import com.airhacks.headlands.notifications.entity.CacheChangedEvent;
import java.io.Serializable;
import java.util.function.Consumer;
import javax.cache.event.CacheEntryCreatedListener;
import javax.cache.event.CacheEntryEvent;
import javax.cache.event.CacheEntryListenerException;
import javax.cache.event.CacheEntryRemovedListener;
import javax.cache.event.CacheEntryUpdatedListener;
import javax.cache.event.EventType;

package com.airhacks.headlands.notifications.control;


/**
 *
 * @author airhacks.com
 */
public class CacheEntryChangedListener implements CacheEntryCreatedListener<String, String>,
        CacheEntryUpdatedListener<String, String>,
        CacheEntryRemovedListener<String, String>,
        Serializable {

    private final transient Consumer<CacheChangedEvent> sink;
    private final String cacheName;

    public CacheEntryChangedListener(String cacheName, Consumer<CacheChangedEvent> sink) {
        this.cacheName = cacheName;
        this.sink = sink;
    }

    @Override
    public void onCreated(Iterable<CacheEntryEvent<? extends String, ? extends String>> events) throws CacheEntryListenerException {
        this.sink.accept(convert(EventType.CREATED, events));
    }

    @Override
    public void onUpdated(Iterable<CacheEntryEvent<? extends String, ? extends String>> events) throws CacheEntryListenerException {
        this.sink.accept(convert(EventType.UPDATED, events));
    }

    @Override
    public void onRemoved(Iterable<CacheEntryEvent<? extends String, ? extends String>> events) throws CacheEntryListenerException {
        this.sink.accept(convert(EventType.REMOVED, events));
    }

    CacheChangedEvent convert(EventType type, Iterable<CacheEntryEvent<? extends String, ? extends String>> events) {
        CacheChangedEvent event = new CacheChangedEvent(cacheName, type);