package cs451.net.handler;

import cs451.net.event.Message;
import cs451.net.NetManager;
import cs451.net.event.*;
import cs451.parser.Host;
import cs451.parser.Parser;
import cs451.tools.Pair;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * NetEventHandler for Stuborn Link
 *
 * SendIf: Sends a message again if not Acknowledged
 *
 * Send: Add Message to "Awaiting Acknowledge" queue
 * Deliver: delivers to next layer (Drops ACK messages and removes peer from "Awaiting Acknowledge"
 * Crash: Mark peer as timed out to stop further sends/acknowledgements
 *
 * @author C. Hölzl
 */
public class NetHandlerSL extends NetEventHandlerAbstract {

    private List<Host> hosts;
    Set<Integer> hasTimeout = ConcurrentHashMap.newKeySet();

    PriorityBlockingQueue<Pair<Pair<Integer, AtomicInteger>,Message>> sending = new PriorityBlockingQueue<>();

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

    @Override
    public void sendIf(){
        sending.forEach(ppm -> {
            ppm.first().second().getAndIncrement();
            Optional<Host> h = hosts.stream().filter(hl->hl.getId()==ppm.first().first()).findFirst();
            h.ifPresent(host -> sendNextAsync(NetEvent.Message(host, ppm.second())));
        });
        sending.removeIf(ppm -> {
            if(hasTimeout.contains(ppm.first().first())){
                return true;
            }
//            if(ppm.first().second().get() > NetManager.SL_MAX_TRIES){
//                hasTimeout.add(ppm.first().first());
//                Optional<Host> h = hosts.stream().filter(hl->hl.getId()==ppm.first().first()).findFirst();
//                h.ifPresent(host -> crashNextAsync(NetEvent.Message(host, ppm.second())));
//                return true;
//            }
            return false;
        });
    }

    @Override
    public void send(NetEvent ne) {
        if(hasTimeout.contains(ne.peer.getId())){
            return;
        }
        sending.add(new Pair<>(new Pair<>(ne.peer.getId(), new AtomicInteger(1)),ne.message));
        sendNextSync(ne);
    }

    @Override
    public void deliver(NetEvent ne) {
        if(hasTimeout.contains(ne.peer.getId())){
            return;
        }
        switch (ne.message.tpe){
            case ACK:
                sending.removeIf(ppm -> (ppm.first().first()==ne.peer.getId() &&
                        ppm.second().equals(ne.message)));
                break;
            case DATA:
                sendNextSync(NetEvent.MessageACK(ne.peer,ne.message));
                deliverNextSync(ne);
                break;
            default:
                break;
        }
    }

    @Override
    public void crash(NetEvent ne) {
        hasTimeout.add(ne.peer.getId());
        crashNextSync(ne);
    }

    @Override
    public boolean isDone() {
        return sending.isEmpty();
    }

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

}
