package gr.ntua.vision.monitoring;

import gr.ntua.vision.monitoring.events.Event;
import gr.ntua.vision.monitoring.events.EventListener;

import java.util.Map;

import org.json.simple.JSONValue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zeromq.ZContext;
import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Socket;


/**
 *
 */
public class EventDistributor implements EventListener {
    /***/
    private static final Logger log = LoggerFactory.getLogger(EventDistributor.class);
    /***/
    private final Socket        sock;


    /**
     * Constructor.
     * 
     * @param ctx
     * @param distributionPoint
     */
    public EventDistributor(final ZContext ctx, final String distributionPoint) {
        this.sock = ctx.createSocket(ZMQ.PUB);
        this.sock.setLinger(0);
        this.sock.setSendTimeOut(0); // FIXME: non-blocking for now
        this.sock.bind(distributionPoint);
        log.debug("listening to endpoint={}", distributionPoint);
    }


    /**
     * @see gr.ntua.vision.monitoring.events.EventListener#notify(gr.ntua.vision.monitoring.events.Event)
     */
    @Override
    public void notify(final Event e) {
        @SuppressWarnings("rawtypes")
        final Map dict = (Map) e.get("!dict");
        final String msg = JSONValue.toJSONString(dict);

        // TODO: get back to this. Should we block or should be drop?
        if (this.sock.send(msg.getBytes(), 0))
            log.trace("sent");
        else
            log.trace("dropped event");
    }
}

@@ -79,63 +79,8 @@
 ent;
-%0Aimport gr.ntua.vision.monitoring.events.EventListener;
 %0A%0Aim
@@ -280,16 +280,341 @@
 %0A%0A/**%0A *
+ The event distributor (TODO: better name) is used to pass events received in localhost to all %3Cem%3EVismo%3C/em%3E consumers. Each%0A * event is sent according to its accompanying topic. NOTE: the running thread does not block, which means that for now, if there%0A * is no receiving end (no connected consumers) the event is dropped.
 %0A */%0Apub
@@ -666,32 +666,49 @@
 stener %7B%0A    /**
+ the log target. 
 */%0A    private s
@@ -788,16 +788,29 @@
 %0A    /**
+ the socket. 
 */%0A    p
@@ -898,16 +898,51 @@
 ram ctx%0A
+     *            the zmq context.%0A
      * @
@@ -1274,18 +1274,18 @@
 stening 
-t
 o
+n
  endpoin
@@ -1361,31 +1361,24 @@
 .monitoring.
-events.
 EventListene
@@ -1629,16 +1629,73 @@
 g(dict);
+%0A        final String topic = (String) dict.get(%22topic%22);
 %0A%0A      
@@ -1770,17 +1770,8 @@
     
-if (this.
 sock
@@ -1776,19 +1776,21 @@
 ck.send(
-msg
+topic
 .getByte
@@ -1798,86 +1798,139 @@
 (), 
-0))%0A            log.trace(%22sent%22);%0A        else%0A            log.trace(
+ZMQ.SNDMORE);%0A        final boolean success = sock.send(msg.getBytes(), 0);%0A        log.trace(%22sent: %7B%7D%22, success ? %22ok%22 : 
 %22dropped
  eve
@@ -1925,22 +1925,16 @@
 %22dropped
- event
 %22);%0A    

send events as multi-part messages