/*
 *  @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.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import net.ranides.assira.trace.AdvLogger;
import net.ranides.assira.trace.LoggerUtils;

/**
 * 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 AdvLogger LOGGER = LoggerUtils.getLogger();

    private final ThreadPoolExecutor executor;
    private final String name;
    private final AtomicInteger threads = new AtomicInteger(0);
    private final BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>();
    private final RejectedExecutionHandler handler = new RejectedExecutionHandler() {
        @Override
        public void rejectedExecution(Runnable task, ThreadPoolExecutor executor) {
            LOGGER.fwarn("ignored task: %s", task);
        }
    };
    ThreadFactory factory = new ThreadFactory() {
        @Override
        public Thread newThread(Runnable runnable) {
            String tid = name + "-" + threads.getAndIncrement();
            LOGGER.fdebug("new thread: %s", tid);
            return new Thread(runnable, tid);
        }
    };

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

    protected EventProactor(final String name, final int size) {
        this.name = name;
        this.executor = new ThreadPoolExecutor(size, size, 0L, TimeUnit.MILLISECONDS, queue, factory, handler);
    }

    public EventProactor start() {
        return this;
    }

    public EventProactor stop() {
        // read updated, signal update, in atomic
        synchronized(executor) {
            if( executor.isShutdown()) {
                return this;
            }
            signalEvent(CoreEvent.stop(this));
            executor.shutdown();
        }
        this.join();
        dispatchEvent(CoreEvent.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;
    }

    public String name() {
        return name;
    }

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

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

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

    @Override
    public String toString() {
        return "EventProactor<" + name + ":" + threads.get() + ">";
    }

    private static final class EventWorker<T extends Event> implements Runnable {
        final EventListener<? super T> listener;
        final T event;

        public EventWorker(EventListener<? super T> listener, T event) {
            this.listener = listener;
            this.event = event;
        }

        @Override
        public void run() {
            listener.handleEvent(event);
        }

        @Override
        public String toString() {
            return "EventWorker{ listener=" + listener + ", event=" + event + " }";
        }

    }

}
