package cs451.net.handler;

import cs451.net.NetManager;
import cs451.net.event.Message;
import cs451.net.event.NetEvent;
import cs451.net.event.NetEventType;
import cs451.parser.Host;
import cs451.parser.Parser;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * NetEventHandler for First In First Out
 *
 * DeliverIf: packet from queue satisfies FIFO ordering
 *
 * Send: sends next packet of current host
 * Deliver: Add packet to received packet queue for DeliverIf
 *
 * @author C. Hölzl
 */
public class NetHandlerFIFO extends NetEventHandlerAbstract {

    private final AtomicInteger sn = new AtomicInteger(1);
    private final Map<Integer,Integer> rsn = new ConcurrentHashMap<>();
    private final PriorityBlockingQueue<Message> pending = new PriorityBlockingQueue<>();

    private Host me;

    public NetHandlerFIFO(Class<? extends NetEventHandlerAbstract> deliverLayer, Class<? extends NetEventHandlerAbstract> broadcastLayer) {
        super(deliverLayer,broadcastLayer);
    }

    @Override
    public void send(NetEvent ne) {
        Integer snv = sn.getAndIncrement();
        NetManager.complete(NetEventType.SEND, NetEvent.Message(me, snv, Message.TYPE.BCST));
        sendNextAsync(NetEvent.Message(me.getId(),snv));
    }

    @Override
    public void deliverIf(){
        pending.removeIf(hmp -> {
            Integer crsn = rsn.getOrDefault(hmp.src, 1);
            if (hmp.id.equals(crsn)) {
                NetManager.complete(NetEventType.DLVR,NetEvent.Message(hmp));
                deliverNextSync(NetEvent.Message(hmp));
                rsn.put(hmp.src, crsn + 1);
                return true;
            }
            return false;
        });
    }

    @Override
    public void deliver(NetEvent ne) {
        pending.add(ne.message);
        deliverIf();
    }

    @Override
    public void start(Host h, Parser p) {
        super.start(h,p);
        me = h;
    }
}
