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

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 {
                    if (s.canHandle(e)) {
                        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)");
                    for (Event e : events.getPendingEvents()) {
                        e.published();
                        events.saveAndFlush(e);
                        synchronized(subscribers) {
                            for (EventSubscriber<?> s : subscribers) {
                                if (s.getClass().toString().equals(e.getSubscriber())) {
                                    handle(s, e);
                                }
                            }
                        }
                    }
                }
            } catch (InterruptedException ok) {
            }
        }

        private void handle(final EventSubscriber<?> s, final Event e) {
            new Thread() {
                @SuppressWarnings("unchecked")
                public void run() {
                    log.info("Publishing event " + e + " to " + s);
                    try {
                        ((EventSubscriber<Event>) s).handle(e);
                    } catch (Exception e) {
                        e.printStackTrace();
                        throw e;
                    }
                }
            }.start();
        }
    }

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

@@ -128,24 +128,82 @@
 Construct;%0A%0A
+import org.apache.commons.lang3.exception.ExceptionUtils;%0A
 import org.s
@@ -215,16 +215,16 @@
 Logger;%0A
-
 import o
@@ -1519,17 +1519,16 @@
  events;
- 
 %0A%0A      
@@ -2489,32 +2489,71 @@
    %7D%0A        %7D%0A%0A
+        @SuppressWarnings(%22unchecked%22)%0A
         private 
@@ -2616,24 +2616,376 @@
 %0A           
+ log.info(%22publishing event %22 + e + %22 to %22 + s);%0A            try %7B%0A                ((EventSubscriber%3CEvent%3E) s).handle(e);%0A            %7D catch (Exception ex) %7B%0A                e.setError(ExceptionUtils.getStackTrace(ex));%0A                ex.printStackTrace();%0A            %7D%0A            // TODO: try this again at some stage, it deadlocks%0A            /*
  new Thread(
@@ -2983,24 +2983,27 @@
 w Thread() %7B
+ */
 %0A           
@@ -2998,24 +2998,27 @@
 %0A           
+ /*
      @Suppre
@@ -3032,32 +3032,35 @@
 ngs(%22unchecked%22)
+ */
 %0A               
@@ -3047,32 +3047,35 @@
 ) */%0A           
+ /*
      public void
@@ -3074,32 +3074,35 @@
 lic void run() %7B
+ */
 %0A               
@@ -3089,32 +3089,35 @@
 %7B */%0A           
+ /*
          log.inf
@@ -3119,17 +3119,17 @@
 g.info(%22
-P
+p
 ublishin
@@ -3148,32 +3148,35 @@
 e + %22 to %22 + s);
+ */
 %0A               
@@ -3163,32 +3163,35 @@
 ; */%0A           
+ /*
          try %7B%0A 
@@ -3180,32 +3180,35 @@
 /*         try %7B
+ */
 %0A               
@@ -3195,32 +3195,35 @@
 %7B */%0A           
+ /*
              ((E
@@ -3250,32 +3250,35 @@
 %3E) s).handle(e);
+ */
 %0A               
@@ -3265,32 +3265,35 @@
 ; */%0A           
+ /*
          %7D catch
@@ -3304,24 +3304,27 @@
 ception e) %7B
+ */
 %0A           
@@ -3315,32 +3315,35 @@
 %7B */%0A           
+ /*
              e.p
@@ -3351,32 +3351,35 @@
 intStackTrace();
+ */
 %0A               
@@ -3366,32 +3366,35 @@
 ; */%0A           
+ /*
              thr
@@ -3398,16 +3398,19 @@
 throw e;
+ */
 %0A       
@@ -3410,32 +3410,34 @@
             
+/*
         
 %7D%0A          
@@ -3416,33 +3416,37 @@
       /*        
-%7D
+ %7D */
 %0A               
@@ -3433,38 +3433,44 @@
 %7D */%0A           
+ /*
      %7D
+ */
 %0A            %7D.s
@@ -3465,16 +3465,19 @@
         
+ /*
  %7D.start
@@ -3475,24 +3475,27 @@
 * %7D.start();
+ */
 %0A        %7D%0A 

Handle events in one thread to avoid deadlocking