// 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.IOException;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.ejb.ConcurrencyManagement;
import javax.ejb.ConcurrencyManagementType;
import javax.ejb.Singleton;
import javax.enterprise.event.Observes;
import javax.json.JsonArray;
import javax.websocket.EncodeException;
import javax.websocket.OnClose;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
package com.airhacks.headlands.notifications.boundary;


/**
 *
 * @author airhacks.com
 */
@Singleton
@ConcurrencyManagement(ConcurrencyManagementType.BEAN)
@ServerEndpoint(value = "/firehose/{cache-name}", encoders = JsonArrayEncoder.class)
public class WebSocketFirehose {

    CopyOnWriteArrayList<Session> sessions;

    @PostConstruct
    public void init() {
        this.sessions = new CopyOnWriteArrayList<>();
    }

    @OnOpen
    public void onConnect(Session session, @PathParam("cache-name") String cacheName) {
        System.out.println("uri: " + session.getRequestURI().toString());
        this.sessions.add(session);
    }

    @OnClose
    public void onClose(Session session) {
        this.sessions.remove(session);
    }

// 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/boundary/WebSocketFirehose.java
import com.airhacks.headlands.notifications.entity.CacheChangedEvent;
import java.io.IOException;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.ejb.ConcurrencyManagement;
import javax.ejb.ConcurrencyManagementType;
import javax.ejb.Singleton;
import javax.enterprise.event.Observes;
import javax.json.JsonArray;
import javax.websocket.EncodeException;
import javax.websocket.OnClose;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;

package com.airhacks.headlands.notifications.boundary;


/**
 *
 * @author airhacks.com
 */
@Singleton
@ConcurrencyManagement(ConcurrencyManagementType.BEAN)
@ServerEndpoint(value = "/firehose/{cache-name}", encoders = JsonArrayEncoder.class)
public class WebSocketFirehose {

    CopyOnWriteArrayList<Session> sessions;

    @PostConstruct
    public void init() {
        this.sessions = new CopyOnWriteArrayList<>();
    }

    @OnOpen
    public void onConnect(Session session, @PathParam("cache-name") String cacheName) {
        System.out.println("uri: " + session.getRequestURI().toString());
        this.sessions.add(session);
    }

    @OnClose
    public void onClose(Session session) {
        this.sessions.remove(session);
    }

    public void onChange(@Observes CacheChangedEvent event) {