package cs451.net;

import cs451.net.event.*;

import cs451.net.handler.*;
import cs451.parser.Host;
import cs451.parser.Parser;
import cs451.tools.Logger;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.*;
import java.util.function.BiConsumer;


/**
 * NetManager
 *
 * Main Manager all the NetEventHandlers.
 * Uses a ThreadPool with a TaskQueue to handle Async Events
 * Uses Synchronous processing to handler Sync Events
 *
 * Has {@link BiConsumer} to handle Completion NetEvents and ErrorEvents
 * Has 2 {@link ScheduledExecutorService} , one for Conditional Sending/Delivery (500ms) and one for Heartbeats(1s)
 *
 * Parameters are adjusted from {@link cs451.tools.ParamDetector}
 *
 * @author C. Hölzl
 */
public abstract class NetManager {

    public static int INTERNAL_WAIT;

    public static int SL_WAIT;
    public static int SL_MAX_TRIES;
    public static int FD_WAIT;
    public static int FD_MAX_TRIES;

    public static Integer THREAD_COUNT;
    public static Integer THREAD_BOOST_COUNT;

    public static Integer WINDOW_WIDTH;


    private static boolean isStopped = false;

    private static final Map<Class<? extends NetEventHandlerAbstract>, NetEventHandlerAbstract> nm_listeners = new HashMap<>();

    private static ThreadPoolExecutor ex;
    private static final ScheduledExecutorService ftex = Executors.newSingleThreadScheduledExecutor();
    private static final ScheduledExecutorService stex = Executors.newSingleThreadScheduledExecutor();

    private static BiConsumer<NetEventType,NetEvent> onCompleteHandler;
    private static BiConsumer<NetEventType,Exception> onErrorHandler;


    /**
     * Resigters a NetEventHandler to the NetManager
     * @param handler NetEventHandler to register
     */
    private static void registerNetHandler(NetEventHandlerAbstract handler){
            nm_listeners.put(handler.getClass(),handler);
    }

    /**
     * Starts the NetManager
     * @param h host (self)
     * @param p parser
     * @param och completion consumer
     * @param oeh error consumer
     */
    public static void start(Host h, Parser p, BiConsumer<NetEventType,NetEvent> och, BiConsumer<NetEventType,Exception> oeh) {
        onCompleteHandler = och;
        onErrorHandler = oeh;

        registerNetHandler(new NetHandlerSCKT(   NetHandlerFD.class,   NetHandlerDFLT.class));
        registerNetHandler(new NetHandlerFD(     NetHandlerSL.class,   NetHandlerSCKT.class));
        registerNetHandler(new NetHandlerSL(     NetHandlerPL.class,   NetHandlerFD.class));
        registerNetHandler(new NetHandlerPL(     NetHandlerBEB.class,  NetHandlerSL.class));
        registerNetHandler(new NetHandlerBEB(    NetHandlerURB.class,  NetHandlerPL.class));
        registerNetHandler(new NetHandlerURB(    NetHandlerFIFO.class, NetHandlerBEB.class));
        registerNetHandler(new NetHandlerFIFO(   NetHandlerTOPL.class, NetHandlerURB.class));
        registerNetHandler(new NetHandlerTOPL(   NetHandlerDFLT.class, NetHandlerFIFO.class));
        registerNetHandler(new NetHandlerDFLT(   NetHandlerDFLT.class, NetHandlerDFLT.class));

        nm_listeners.values().forEach(neh -> neh.start(h,p));
        ex = new ThreadPoolExecutor(THREAD_COUNT,THREAD_COUNT+THREAD_BOOST_COUNT,30,TimeUnit.SECONDS,new LinkedBlockingQueue<>());
        ex.prestartAllCoreThreads();

        ftex.scheduleAtFixedRate(() -> nm_listeners.values().parallelStream().forEach(neh->{
            neh.deliverIf();
            neh.sendIf();
        }), 0, INTERNAL_WAIT, TimeUnit.MILLISECONDS);

        stex.scheduleAtFixedRate(()-> nm_listeners.values().forEach(NetEventHandlerInterface::beat
         ), 0, FD_WAIT, TimeUnit.MILLISECONDS);
    }


    /**
     * Stops the NetManager
     */
    public static void stop() {
        isStopped = true;
        nm_listeners.values().forEach(NetEventHandlerAbstract::stop);
        ex.shutdown();
        ftex.shutdown();
        stex.shutdown();
        System.out.println("NetManager handled "+ex.getCompletedTaskCount()+" tasks during this run.");
    }

    /**
     * Checks if the NetManager and its layers are done with their work
     * @return true if NM and NEH are done
     */
    public static boolean isDone() {
        return isStopped || nm_listeners.values().stream().map(NetEventHandlerAbstract::isDone).reduce(ex.getActiveCount()==0 && ex.getQueue().size()==0,(b1, b2) -> b1 && b2);
    }


//=================================================================================================================
//=================================================================================================================

    public static void handle(Class<? extends NetEventHandlerAbstract> layer, NetEventType event, NetEvent ne){
        try {
            ex.getQueue().add(new NetEventRunner(ne, layer, event));
        }catch(IllegalStateException ise){
            error(event,ise);
        }
    }

    public static void complete(NetEventType t,NetEvent ne){
        onCompleteHandler.accept(t,ne);
    }

    public static void error(NetEventType t,Exception e){
        onErrorHandler.accept(t,e);
    }

//=================================================================================================================
//=================================================================================================================

    public static void deliver(Class<? extends NetEventHandlerAbstract> layer, NetEvent ne) {
        handle(layer,NetEventType.DLVR,ne);
    }
    public static void send(Class<? extends NetEventHandlerAbstract> layer, NetEvent ne) {
        handle(layer,NetEventType.SEND,ne);
    }
    public static void crash(Class<? extends NetEventHandlerAbstract> layer, NetEvent ne) {
        handle(layer,NetEventType.CRSH,ne);
    }

    public static void deliverSync(Class<? extends NetEventHandlerAbstract> layer, NetEvent ne) {
        new NetEventRunner(ne,layer,NetEventType.DLVR).run();
    }
    public static void sendSync(Class<? extends NetEventHandlerAbstract> layer, NetEvent ne) {
        new NetEventRunner(ne,layer,NetEventType.SEND).run();
    }
    public static void crashSync(Class<? extends NetEventHandlerAbstract> layer, NetEvent ne) {
        new NetEventRunner(ne,layer,NetEventType.CRSH).run();
    }

    public static void send() {
        new NetEventRunner(NetEvent.EMPTY(), NetHandlerTOPL.class,NetEventType.SEND).run();
    }


//=================================================================================================================
//=================================================================================================================

    /**
     * Runner for NetEvent instances
     */
    private static class NetEventRunner implements Runnable {

        private final NetEvent ne;
        private final Class<? extends NetEventHandlerAbstract> lt;
        private final NetEventType net;

        private NetEventRunner(NetEvent ne, Class<? extends NetEventHandlerAbstract> lt, NetEventType net) {
            this.ne = ne;
            this.lt = lt;
            this.net = net;
        }

        @Override
        public void run() {
            Logger.debug(this.lt.getSimpleName()+"_"+this.net+ ": " + ((ne.peer==null)?"?":ne.peer.getId()) + " - "+ne.message.toString()+"");

            NetEventHandlerAbstract nl = nm_listeners.getOrDefault(this.lt, null);
            if (nl != null) {
                nl.onEvent(this.net,ne);
            }else{
                error(this.net,new UnsupportedOperationException("No Handler for "+this.lt));
            }
        }
    }

}


