<commit_before>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);
    }
}
<commit_msg>Handle events in one thread to avoid deadlocking<commit_after>package au.gov.ga.geodesy.domain.model.event;

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

import javax.annotation.PostConstruct;

import org.apache.commons.lang3.exception.ExceptionUtils;
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) {
            }
        }

        @SuppressWarnings("unchecked")
        private void handle(final EventSubscriber<?> s, final Event e) {
            log.info("publishing event " + e + " to " + s);
            try {
                ((EventSubscriber<Event>) s).handle(e);
            } catch (Exception ex) {
                e.setError(ExceptionUtils.getStackTrace(ex));
                ex.printStackTrace();
            }
            // TODO: try this again at some stage, it deadlocks
            /* 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);
    }
}

{'fraction_non_alphanumeric': 0.10666863381528474, 'fraction_numerical': 0.0019179699026261434, 'mean_word_length': 1.3335628227194491, 'pattern_counts': {'":': 0, '<': 24, '<?xml version=': 0, '>': 24, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1377208', 'n_tokens_mistral': 1709, 'n_tokens_neox': 1622, 'n_words': 348}