/*
 * @author Ranides Atterwim <ranides@gmail.com>
 * @copyright Ranides Atterwim
 * @license WTFPL
 * @url http://ranides.net/projects/assira
 */
package net.ranides.assira.events;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import net.ranides.assira.trace.LoggerUtils;
import org.slf4j.Logger;

/**
 * EventProactor, obsługuje każdego listenera i każde zdarzenie asynchronicznie.
 * Korzysta z puli wątków, żeby nie powodować eksplozji wątków.
 *
 * Standardowy ThreadPoolExecutor ma w sumie głupią tę strategię zarządzania pulą,
 * napisać *kiedyś* własną wersję, albo przemyśleć konfigurację.
 *
 * <div class="message-note">thread-safe method</div>
 * @author ranides
 */
public class EventProactor extends EventDispatcher {

    private static final Logger LOGGER = LoggerUtils.getLogger();

    private final Object lock = new Object();
    private final ThreadPoolExecutor executor;
    private final AtomicInteger threads = new AtomicInteger(0);
    private final BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>();

    public static EventProactor newInstance(String name, int size) {
        return new EventProactor(name, size);
    }

    protected EventProactor(final String name, final int size) {
        super(name);
        this.executor = new ThreadPoolExecutor(
            size, size, 
            0L, TimeUnit.MILLISECONDS, 
            queue, 
            this::newThread, 
            this::rejectedExecution
        );
    }
    
    public EventProactor stop() {
        // read updated, signal update, in atomic
        synchronized(lock) {
            if( executor.isShutdown()) {
                return this;
            }
            signalEvent(Events.stop(this));
            executor.shutdown();
        }
        this.join();
        dispatchEvent(Events.shutdown(EventProactor.this), true);
        dispose(true);
        return this;
    }

    public EventProactor join() {
        try { executor.awaitTermination(Integer.MAX_VALUE, TimeUnit.MILLISECONDS); }
        catch (InterruptedException _e) { /* do nothing */ }
        return this;
    }

    @Override
    protected <T extends Event> void dispatchEvent(final EventListener<? super T> listener, final T event) {
        executor.execute(()->listener.handleEvent(event));
    }

    @Override
    public void dispose() {
        // read updated, signal update, in atomic
        synchronized(lock) {
            if( executor.isShutdown()) {
                return;
            }
            signalEvent(Events.stop(this));
            executor.shutdown();
        }

        new Thread(){
            @Override
            public void run() {
                EventProactor.this.join();
                dispatchEvent(Events.shutdown(EventProactor.this), true);
                dispose(true);
            }
        }.start();
    }

    @Override
    public String toString() {
        return "EventProactor<" + name() + ":" + threads.get() + ">";
    }
    
    private Thread newThread(Runnable runnable) {
        String tid = name() + "-" + threads.getAndIncrement();
        LOGGER.debug("new thread: {}", tid);
        return new Thread(runnable, tid);
    }
    
    private void rejectedExecution(Runnable task, ThreadPoolExecutor $0) {
        LOGGER.warn("ignored task: {}", task);
    }

}
