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 {
                    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);
                    ((EventSubscriber<Event>) s).handle(e);
                }
            }.start();
        }
    }

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

@@ -2681,24 +2681,54 @@
  to %22 + s);%0A
+                    try %7B%0A    
             
@@ -2771,24 +2771,168 @@
 .handle(e);%0A
+                    %7D catch (Exception e) %7B%0A                        e.printStackTrace();%0A                        throw e;%0A                    %7D%0A
             

Print event handler exceptions to standard error.