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

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 net.ranides.assira.trace.ThreadDump;
import org.slf4j.Logger;

/**
 * EventProactor handles every listener and every event asynchronously.
 * It uses private thread pool to avoid spawning too much threads.
 *
 * Implementation note:
 * Default ThreadPoolExecutor has relatively dumb stragegy for thread pool management,
 * we could write our own executor or review configuration options.
 *
 * <div class="message-note">thread-safe method</div>
 * @author ranides
 */
public class EventProactor extends EventRouter {

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

    private final Object lock = new Object();
    private final ThreadPoolExecutor executor;
    private final AtomicInteger threads = new AtomicInteger(0);

    /**
     * Creates new EventProactor with specified name.
     * It allocates at most number of threads proportional to number of cores.
     *
     * Created router should be destroyed by calling "stop" or "dispose".
     *
     * @param name name
     * @return EventProactor
     */
    public static EventProactor newInstance(String name) {
        return newInstance(name, 0, Runtime.getRuntime().availableProcessors());
    }

    /**
     * Creates new EventProactor with specified name.
     * It allocates exactly "size" threads inside pool.
     *
     * Created router should be destroyed by calling "stop" or "dispose".
     *
     * @param name name
     * @param size size
     * @return EventProactor
     */
    public static EventProactor newInstance(String name, int size) {
        return newInstance(name, size, size);
    }

    /**
     * Creates new EventProactor with specified name.
     * It allocates at least "min" and at most "max" threads inside pool.
     *
     * Created router should be destroyed by calling "stop" or "dispose".
     *
     * @param name name
     * @param min min
     * @param max max
     * @return EventProactor
     */
    public static EventProactor newInstance(String name, int min, int max) {
        return EventDispatcher.newInstance(name, () -> new EventProactor(name, min, max));
    }

    private EventProactor(String name, int min, int max) {
        super(name);
        this.executor = new ThreadPoolExecutor(
            min, max,
            0L, TimeUnit.MINUTES,
            new LinkedBlockingQueue<>(),
            this::newThread, 
            this::rejectedExecution
        );
    }

    /**
     * It initiates shutdown of this EventRouter.
     *
     * It processes all already enqued events, but does not accept new ones.
     *
     * It sends few loop-back messages:
     *  - Events.Stop
     *  - Events.Shutdown
     *  - Events.Dispose
     *
     * It blocks calling thread until whole procedure of shutdown is finished.
     *
     * @return this
     */
    public EventProactor stop() {
        // read updated, signal update, in atomic
        synchronized(lock) {
            if( executor.isShutdown()) {
                return this;
            }
            signalEvent(Events.stop(this));
            executor.shutdown();
        }
        this.join();
        executeListeners(Events.shutdown(EventProactor.this), EventListener::handleEvent);
        executeDisposeNow();
        return this;
    }

    /**
     * It blocks calling thread until current proactor is stopped.
     *
     * @return this
     */
    public EventProactor join() {
        try { executor.awaitTermination(Integer.MAX_VALUE, TimeUnit.MILLISECONDS); }
        catch (InterruptedException _e) { /* do nothing */ }
        return this;
    }

    @Override
    public boolean signalEvent(Event event) {
        ThreadDump.printIf(LOGGER.isTraceEnabled(), LOGGER::trace);
        executeListeners(event, (listener, e) -> executor.execute(()->listener.handleEvent(e)));
        return true;
    }

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

        new Thread(() -> {
            EventProactor.this.join();
            executeListeners(Events.shutdown(EventProactor.this), EventListener::handleEvent);
            executeDisposeNow();
        }).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);
    }

}
