/*
 * @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.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import net.ranides.assira.trace.LoggerUtils;
import net.ranides.assira.trace.ThreadDump;
import org.slf4j.Logger;

/**
 * EventReactor handles all events in message que.
 * It executes event listeners in unspecified future.
 *
 * Event listeners are executed in private background thread managed by this reactor.
 *
 * <div class="message-note">thread-safe method</div>
 * @author ranides
 */
public class EventReactor extends EventRouter {
    
    private static final Logger LOGGER = LoggerUtils.getLogger();

    private final BlockingQueue<Event> events;
    private final long maxtime;
    private final Thread distributor;
    private final Events.Stop exit;
    
    /**
     * Creates new instance of {@code EventRouter} with specified name.
     * It stores unbounded message queue and it does not impose timeouts for event handlers.
     *
     * Created router should be destroyed by calling "stop" or "dispose".
     *
     * @param name name
     * @return EventReactor
     */
    public static EventReactor newInstance(String name) {
        return newInstance(name, 0, 0, EventJoiner.NONE);
    }
    
    /**
     * Creates new instance of {@code EventRouter} with specified name.
     * It stores unbounded message queue and it does not impose timeouts for event handlers.
     * It uses provided EventJoiner strategy.
     *
     * Created router should be destroyed by calling "stop" or "dispose".
     *
     * @param name name
     * @param joiner joiner
     * @return EventReactor
     */
    public static EventReactor newInstance(String name, EventJoiner joiner) {
        return newInstance(name, 0, 0, joiner);
    }
    
    /**
     * Creates new instance of {@code EventRouter} with specified name.
     * It stores at most "size" last messages: if queue becomes full,
     * it waits "maxtime" milliseconds for place in queue, and discards message on timeout.
     *
     * Created router should be destroyed by calling "stop" or "dispose".
     *
     * @param name name
     * @param size size
     * @param maxtime maxtime
     * @return EventReactor
     */
    public static EventReactor newInstance(String name, int size, long maxtime) {
        return newInstance(name, size, maxtime, EventJoiner.NONE);
    }

    /**
     * Creates new instance of {@code EventRouter} with specified name.
     * It uses provided EventJoiner strategy.
     * It stores at most "size" last messages: if queue becomes full,
     * it waits "maxtime" milliseconds for place in queue, and discards message on timeout.
     *
     * Created router should be destroyed by calling "stop" or "dispose".
     *
     * @param name name
     * @param size size
     * @param maxtime maxtime
     * @param joiner joiner
     * @return EventReactor
     */
    public static EventReactor newInstance(String name, int size, long maxtime, EventJoiner joiner) {
        return newInstance(name, () -> new EventReactor(name, size, maxtime, joiner)).start();
    }

    /**
     * Private constructor, should not be called directly.
     * {@code EventRouter} works after calling {@link #start()}.
     *
     * @param name name
     * @param maxtime maxtime
     * @param size size
     * @param joiner joiner
     */
    private EventReactor(final String name, int size, long maxtime, final EventJoiner joiner) {
        super(name);
        this.maxtime = maxtime;
        this.events = createQueue(size);
        this.exit = Events.stop(this);
        this.distributor = new Thread(name) {

            @Override
            public void run() {
                try {
                    while (true) {
                        Event event = events.take();
                        while(joiner.isJoinable(event, events.peek()) ) {
                            event = joiner.join(event, events.poll());
                        }
                        executeListenersNow(event);
                        if (isExitEvent(event)) {
                            break;
                        }
                    }
                } catch (InterruptedException _e) {
                    executeListenersNow(Events.interrupt(EventReactor.this));
                }
                executeListenersNow(Events.shutdown(EventReactor.this));
            }
        };
    }
    
    private static BlockingQueue<Event> createQueue(int size) {
        if( 0==size || Integer.MAX_VALUE==size) {
            return new LinkedBlockingQueue<>();
        } else {
            return new ArrayBlockingQueue<>(size, true);
        }
    }

    private EventReactor start() {
        distributor.start();
        return this;
    }

    /**
     * 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 EventReactor stop() {
        signalEvent(exit);
        join();
        executeDisposeNow();
        return this;
    }

    /**
     * It blocks calling thread until current reactor is stopped.
     *
     * Current reactor must be stopped by other thread, by calling {@link #stop} or {@link #dispose()}.
     *
     * @return this
     */
    public EventReactor join() {
        try { distributor.join(); }
        catch (InterruptedException _e) { /* do nothing */ }
        return this;
    }

    @SuppressWarnings("PMD.CompareObjectsWithEquals")
    private boolean isExitEvent(Event event) {
        if(exit == event) { return true; }
        return (event instanceof Events.Dispose) && this == ((Events.Dispose)event).router();
    }

    /**
     * {@inheritDoc}.
     *
     * Puts message into message queue.
     * If queue is full, it waits for release spaces (i.e. processing some events) for "maxtime".
     * It ignores event on timeout.
     *
     * This method does not fire any event listeners: event handling is delegated to private thread.
     *
     * It does not wait for successfull processing: it returns "false" if there was no place in queue.
     * It returns "true" if event is scheduled.
     *
     * @param event event
     * @return boolean
     */
    @Override
    public boolean signalEvent(Event event) {
        ThreadDump.printIf(LOGGER.isTraceEnabled(), LOGGER::trace);
        if(0 == maxtime) {
            return events.offer(event);
        }
        try {
            return events.offer(event, maxtime, TimeUnit.MILLISECONDS);
        } catch (InterruptedException _e) {
            return false;
        }
    }

    @Override
    public void dispose() {
        signalEvent(exit);
        new Thread(() -> {
            EventReactor.this.join();
            EventReactor.this.executeDisposeNow();
        }).start();
    }

    @Override
    public String toString() {
        return "EventReactor<" + name() + ">";
    }

}
