package au.gov.ga.geodesy.domain.model;

import java.util.ArrayList;
import java.util.List;

import javax.annotation.PostConstruct;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

public class AsynchronousEventPublisher implements EventPublisher {

    private static final Logger log = LoggerFactory.getLogger(AsynchronousEventPublisher.class);

    @Autowired
    private EventRepository events;

    private List<EventSubscriber<?>> subscribers = new ArrayList<EventSubscriber<?>>();

    @PostConstruct
    public void startEventLoop() {
        new EventLoop(events).start();
    }

    public void subscribe(EventSubscriber<?> s) {
        subscribers.add(s);
    }

    public void publish(Event... es) {
        for (Event e : es) {
            for (EventSubscriber<?> s : subscribers) {
                try {
                    Event published = (Event) e.clone();
                    published.setSubscriber(s.getClass().toString());
                    events.save(published);
                } catch (CloneNotSupportedException ex) {
                    throw new RuntimeException(ex);
                }
            }
            log.info("Scheduling event " + e + " for asynchronous publishing");
        }
    }

    public class EventLoop extends Thread {

        private EventRepository events; 

        public EventLoop(EventRepository es) {
            events = es;
        }

        @Override
        public void run() {
            try {
                while(true) {
                    Thread.sleep(3000);
                    List<Event> es = events.getPendingEvents();
                    log.info("Processing " + es.size() + " pending event(s)");
                    System.out.println(subscribers.size());
                    for (Event e : events.getPendingEvents()) {
                        e.published();
                        events.saveAndFlush(e);
                        synchronized(subscribers) {
                            for (EventSubscriber<?> s : subscribers) {
                                if (s.canHandle(e) && s.getClass().toString().equals(e.getSubscriber())) {
                                    handle(s, e);
                                }
                            }
                        }
                    }
                }
            } catch (InterruptedException ok) {
            }
        }

        /* @Transactional */
        private void handle(final EventSubscriber<?> s, final Event e) {
            new Thread() {
                @SuppressWarnings("unchecked")
                public void run() {
                    log.info("Publishing event " + e + " to " + s);
                    ((EventSubscriber<Event>) s).handle(e);
                    /* events.saveAndFlush(e); */
                }
            }.start();
        }
    }

    public void handled(Event e) {
        e.handled();
        events.saveAndFlush(e);
    }
}

@@ -877,32 +877,78 @@
           try %7B%0A
+                    if (s.canHandle(e)) %7B%0A    
                 
@@ -984,24 +984,28 @@
  e.clone();%0A
+    
             
@@ -1074,32 +1074,36 @@
                 
+    
 events.save(publ
@@ -1110,16 +1110,38 @@
 ished);%0A
+                    %7D%0A
         
@@ -1822,68 +1822,8 @@
 %22);%0A
-                    System.out.println(subscribers.size());%0A
     
@@ -2132,26 +2132,8 @@
 if (
-s.canHandle(e) && 
 s.ge
@@ -2438,37 +2438,8 @@
  %7D%0A%0A
-        /* @Transactional */%0A
     
@@ -2681,24 +2681,24 @@
  to %22 + s);%0A
+
             
@@ -2749,58 +2749,8 @@
 e);%0A
-                    /* events.saveAndFlush(e); */%0A
     

Fix publishing of multiple event types.