
package qms.queue;
import qms.group.*;
import qms.group.Node;
import qms.messages.*;
import qms.util.*;
import qms.queue.actions.*;
import java.net.*;
import java.util.*;
import java.io.IOException;

/**
 * Il componente che decide per il nodo locale cosa fare
 * @author  Enrico
 */
public class Controller implements Runnable, ActionListener  {
    
    protected static Controller instance;
    
    public static final boolean DEBUG = true;
    /**
     * Instanzia l'istanza Singleton
     */
    public static void createLocalInstance(Group nodes) {
        instance = new Controller(nodes);
    }
    public static Controller getLocalInstance() { return instance; }
    
    protected Multicaster multicast;
    protected Group nodes;
    protected QueuesManager queues;
    protected Vector actions;
    protected Thread life;
    protected boolean stop;
    
    protected List freezed_messages;
    protected boolean freezed;
    /**
     * Indica se il nodo è "freezed"
     */
    public boolean isFreezed() { return freezed; }
    /**
     * Ottiene una lista dei messaggi congelati
     */
    public List getFreezedMessages() { return new Vector(freezed_messages); }
    /**
     * Ottiene una lista delle azioni in esecuzione
     */
    public List getActions() { synchronized(actions) { return new Vector(actions); } }
    
    public Group getGroup() {return nodes;}
    public QueuesManager getQueues() { return queues;}
    public Multicaster getMulticaster() {return multicast; }
    
    protected Controller(Group nodes) {
        this.multicast = new Multicaster(nodes);
        this.nodes = nodes;
        this.queues = new QueuesManager(nodes);
        this.actions = new Vector();
        this.freezed_messages = new Vector();
        this.freezed = false;
        this.stop = false;
        this.life = new Thread(this);
        this.life.setName("Node Controller life thread");
        this.life.setDaemon(true);
        this.life.start();
        // azione da compiere sempre...
        this.addAction( new AnswerLockRequestsAction(this));
        this.addAction( new PingGroupAction(this));
    }
    /**
     * Libera il nodo del freeze ed avvia la fase di elaborazione dei messaggi "freezati"
     */
    public void unfreeze() {
        freezeQueueMessages(false);
    }
    /**
     * Blocca o sblocca la ricezione di messaggio per le "code".
     * Nel caso di sblocco vengono inviati tutti i messaggi "pendenti"
     */
    protected synchronized void freezeQueueMessages(boolean freeze) {
        if (freeze) {
            freezed = true;
        } else {
            freezed = false;
            for (Iterator i = freezed_messages.iterator(); i.hasNext(); ) {
                Message m = (Message)  i.next();
                i.remove();
                messageReceived(m);
            }
        }
    }
    /**
     * Aggiunge una azione
     */
    protected synchronized void addAction(Action action) {
        actions.add(action);
    }
    
    
    protected boolean isGroupManagementMessage(Message cm) {
        String a = cm.getType();
        return a.equals("group_join_request")  ||
        a.equals("group_node_added")  ||
        a.equals("group_node_leave")||
        
        a.equals("group_ping")      ||
        
        a.equals("group_freeze")      ||
        a.equals("group_ready")       ||
        
        a.equals("group_node_exists")||
        a.equals("group_bulk_copy_message");
    }
    
    
    /*
     * Invia un messaggio a tuti i nodi del gruppo di "congelarsi" perchè bisogna intervenire sulla struttura del gruppo
     */
    public void freezeGroup() {
        Message msg = new Message( Node.getLocalNode().getName(), Multicaster.ALL, "system","group_freeze","");
        multicast.castMessage(msg);
    }
    
    /**
     * Avvia lengthazioni necessarie al post di un nuovo messaggio
     */
    protected void scheduleMessagePosting(Message m) {
        String lockid = QueuesManager.createNewLockID();
        UnlockQueueAction unlock = new UnlockQueueAction(m.getQueueId(),lockid, this);
        PostMessageAction distribute = new PostMessageAction(m,this,unlock);
        LockQueueAction lock = new LockQueueAction(m.getQueueId(),lockid, this, distribute);
        addAction(lock);
        addAction(distribute);
        addAction(unlock);
    }
    /**
     * Avvia la sequenza di azioni necessarie alla cancellazione di un messaggio
     */
    protected void scheduleMessageDeletion(Message m) {
        String lockid = QueuesManager.createNewLockID();
        UnlockQueueAction unlock = new UnlockQueueAction(m.getQueueId(),lockid, this);
        DeleteMessageAction delete = new DeleteMessageAction(m.getQueueId(), m.getBody(),this,unlock);
        LockQueueAction lock = new LockQueueAction(m.getQueueId(),lockid, this, delete);
        addAction(lock);
        addAction(delete);
        addAction(unlock);
    }
    /**
     * Risponde ad un client con un messaggio presente in una coda
     */
    protected Message answerGetMessage(Message m) {
        MessageQueue q = queues.getQueue(m.getQueueId());
        Message result = q.getMessage(m.getBody());
        if (result == null) {
            result = new Message("system",m.getSender(),m.getQueueId(),"failure","Message "+m.getBody()+" does not exists");
        }
        return result;
    }
    /**
     * Risponde ad un client con la lista degli id dei messaggi in una coda
     */
    protected Message answerListMessages(Message m) {
        MessageQueue q = queues.getQueue(m.getQueueId());
        Message result = new Message("system",m.getSender(),m.getQueueId(),"messages_list","");
        StringBuffer ids = new StringBuffer();
        for (Iterator i = q.getMessages().iterator(); i.hasNext();) {
            Message ms = (Message) i.next();
            ids.append(ms.getId()+"\n");
        }
        result.setBody(ids.toString());
        return result;
    }
    /**
     * Invocato dalla UserInterface alla ricezione di un messaggio dal client
     */
    public Message userMessageReceived(Message m) {
        String type = m.getType();
        
        Message response = new Message("system",m.getSender(),"system","ok","request received");
        
        if (type.equals("user_post_message")) {
            m.setType("user_msg");
            scheduleMessagePosting(m);
            return response;
        }
        if (type.equals("user_delete_message")) {
            scheduleMessageDeletion(m);
            return response;
        }
        
        if (type.equals("user_get_message")) {
            return answerGetMessage(m);
        }
        if (type.equals("user_list_messages")) {
            return answerListMessages(m);
        }
        
        if (type.equals("user_delete_queue")) {
            // invio sulla coda system un messaggio di controllo di eliminare una coda
            Message internal_delete_queue = new Message(Node.getLocalNode().getName(),Multicaster.ALL,QueuesManager.SYSTEM_QUEUE_ID,"system_delete_queue",m.getQueueId());
            scheduleMessagePosting(internal_delete_queue);
            return response;
        }
        
        response = new Message("system",m.getSender(),"system","failure","Bad request");
        return response;
    }
    /**
     * Invocato dalla CoordinationInterface alla ricezione di un messaggio di coordinamento
     */
    public  void messageReceived(Message m) {
        Logger.log("Message received:"+m);
        if (isGroupManagementMessage(m))  {
            dispatchGroupManagementMessage(m);
            return;
        }
        if (freezed) {
            freezed_messages.add(m);
            return;
        }
        if (isQueueManagementMessage(m)) {
            dispatchQueueManagementMessage(m);
            return;
        }
        
        dispatchUserRequest(m);
    }
    /**
     * Avvia lengthazioni necessarie ad aggiungere un nodo al gruppo
     */
    protected void scheduleNewNodeJoin(Message m) {
        try {
            String node_name = m.getSender();
            String node_sap = m.getBody();
            URL url = new URL(node_sap);
            Node newnode = new Node();
            newnode.setName(node_name);
            newnode.setAccessPoint(url);
            
            UnFreezeGroupAction unfreeze = new UnFreezeGroupAction(this,new WaitingAction() { public void canStart(){}; public void cancel() {} });
            InformNewNodeExistanceAction inform = new InformNewNodeExistanceAction(this,newnode,unfreeze);
            BulkCopyAction    bulk_copy = new BulkCopyAction(this,newnode,inform);
            FreezeGroupAction freeze = new FreezeGroupAction(this,bulk_copy);
            
            addAction(freeze);
            addAction(bulk_copy);
            addAction(inform);
            addAction(unfreeze);
        } catch (MalformedURLException e) {
            Logger.log("Impossibile aggiungere il nodo. Il suo Accessp Point non è corretto");
            Logger.log(e);
        }
        
    }
    /**
     * Invocato alla ricezione di un messaggio che indica al gruppo che esiste un nuovo nodo
     */
    protected void newNodeExists(Message m) {
        String node_name = m.getSender();
        String sap = m.getBody();
        try {
            URL url = new URL(sap);
            Node n = new Node();
            n.setName(node_name);
            n.setAccessPoint(url);
            nodes.add(n);
        } catch (MalformedURLException e) {
            Logger.log(e);
        }
    }
    /**
     * Bandisce un nodo
     */
    public void banNode(Node node) {
        Logger.log("System required me directly to ban "+node);
        if (node == null) return;
        notifyNodeBanned(node);
        scheduleBanNode(node);
    }
    /**
     * Inizia le azioni necessarie a bandire un nodo dal gruppo
     */
    protected void scheduleBanNode(Node node) {
        Logger.log("Scheduling Banning of node "+node);
        InformNodeBannedAction banner = new InformNodeBannedAction(this,node);
        addAction(banner);
    }
    /**
     * Inizia il protocollo di leave dal gruppo
     **/
    public void leaveGroup() {
        Logger.log("Leaving group");
        SendWannaLeaveAction leaver = new SendWannaLeaveAction(this);
        addAction(leaver);
    }
    protected void notifyNodeBanned(Node n ) {
        if (n == null) return;
        if ( n != Node.getLocalNode()) {
            Logger.log("Removing "+n+" from group knowledge");
            nodes.remove(n);
            Logger.log("Notifying queues");
            queues.notifyNodeRemoved(n);
            Logger.log("Notifying actions");
            synchronized(actions) {
                for (Iterator a = actions.iterator(); a.hasNext();) {
                    Action action = (Action) a.next();
                    action.notifyNodeRemoved(n);
                }
            }
            Logger.log("Finished notifying..");
        }
    }
    /**
     * Invocato alla ricezione di un messaggio che indica al gruppo che esiste un nuovo nodo
     */
    protected void nodeBanned(Message m) {
        Logger.log("Processing node_leave "+m);
        try {
            String node_name = m.getSender();
            String sap = m.getBody();
            URL url = new URL(sap);
            Node n = new Node();
            n.setName(node_name);
            n.setAccessPoint(url);
            notifyNodeBanned(n);
        } catch (MalformedURLException e) {
            Logger.log(e);
        }
    }
    /**
     * Invocato alla ricezione di un messaggio che indica un messaggio da aggiungere durante l'ingresso nel gruppo
     */
    protected void bulkCopyMessage(Message m) {
        Logger.log("Received bulk copy msg");
        try {
            Message msg = MessageFactory.decodeMessage(m.getBody());
            Logger.log(msg);
            MessageQueue queue = queues.getQueue(msg.getQueueId());
            queue.addMessage(msg);
            Logger.log("Saved");
        }catch (IOException io) {
            Logger.log("troubles decoding bulk copy msg\n"+m.getBody());
            Logger.log(io);
        }
    }
    protected void dispatchGroupManagementMessage(Message m) {
        String type = m.getType();
        if (type.equals("group_freeze")) {
            freezeQueueMessages(true);
            return;
        }
        if (type.equals("group_ready")) {
            freezeQueueMessages(false);
            return;
        }
        if (type.equals("group_bulk_copy_message")) {
            bulkCopyMessage(m);
            return;
        }
        if (type.equals("group_join_request")) {
            scheduleNewNodeJoin(m);
            return;
        }
        if (type.equals("group_node_exists")) {
            newNodeExists(m);
            return;
        }
        if (type.equals("group_node_leave")) {
            nodeBanned(m);
            return;
        }
        Logger.log("Trascuro il messaggio di gestione del gruppo "+m);
    }
    protected void dispatchQueueManagementMessage(Message m) {
        Logger.log("Accetto il messaggio di gestione di una coda "+m);
        MessageQueue queue = queues.getQueue(m.getQueueId());
        queue.pushMessageOntoStack(m);
    }
    protected void dispatchUserRequest(Message m) {
        Logger.log("Trascuro il messaggio utente "+m);
    }
    protected boolean isQueueManagementMessage(Message cm) {
        String a = cm.getType();
        return
        a.equals("queue_request_lock")||
        a.equals("queue_agree_lock")||
        a.equals("queue_unlock")||
        a.equals("queue_delete_message")||
        a.equals("queue_post_message") ;
    }
    
    /**
     * Stoppa il thread di elaborazione delle azioni
     */
    public void stop() {
        stop = true;
    }
    /**
     * Corpo del thread che assegna la CPU alle azioni
     */
    public void run() {
        while (!stop) {
            try {
                synchronized(actions) {
                    for (Iterator i = actions.iterator(); i.hasNext(); ) {
                        Action a = (Action) i.next();
                        try {
                            a.runStep();
                            if (a.isFinished()) i.remove();
                        } catch (Throwable t) {
                            Logger.log("Exception while handling action:"+a);
                            Logger.log(t);
                        }
                    }
                }
                
                try {
                    Thread.sleep(10);
                } catch (InterruptedException ie) {}
            } catch (Throwable t) {
                Logger.log(t);
            }
        }
    }
    
    public void actionFinished(Action action) {
    }
    /**
     * Richiede ad un nodo di iniziare la procedura di JOIN
     */
    public void requestJoin(String sap) throws MalformedURLException {
        freezeQueueMessages(true);
        Node tmp = new Node();
        URL url = new URL(sap);
        tmp.setName("unknown");
        tmp.setAccessPoint(url);
        Message request = new Message(Node.getLocalNode().getName(),tmp.getName(),QueuesManager.GROUP_QUEUE_ID,"group_join_request",Node.getLocalNode().getAccessPoint().toString());
        multicast.castMessage(request,tmp);
    }
}

