
package qms.messages;
import java.util.*;
import qms.messages.*;
import qms.group.*;
import qms.group.Node;
import qms.util.*;
import java.io.IOException;
/**
 * Una coda. Il mattoncino base di questo sistema.
 * @author  Enrico
 */
public class MessageQueue {
    protected LinkedList messages;
    protected CoordinationMessagesStack stack;        
    protected String queueId;
    protected Group nodes;    
    protected boolean locked;
    protected Set listeners;
    protected Set life_listeners;
    protected Set owners;
    
    public String getId() { return queueId; }
    public MessageQueue(String queueId, Group nodes) {
        this.queueId = queueId;
        this.nodes = nodes;
        this.listeners = new HashSet();
        this.owners = new HashSet();
        this.life_listeners = new HashSet();
        
        this.messages = new LinkedList();
        this.stack = new CoordinationMessagesStack(nodes);        
        this.locked = false;
    }
    /**
     * Aggiunge un oggetto interessato ai cambiamenti nello stato della coda
     */
    public void addListener(QueueListener l) { listeners.add(l); }
    /**
     * Rimuove un oggetto interessato ai cambiamenti nello stato della coda
     */
    public void removeListener(QueueListener l) { listeners.remove(l); }
    /**
     * Aggiunge un oggetto interessato alla vita di questa coda
     */
    public void addQueueLifeListener(QueueLifeListener l) { life_listeners.add(l); }
    /**
     * Rimuove un oggetto interessato alla vita di questa coda
     */
    public void removeQueueLifeListener(QueueLifeListener l) { life_listeners.remove(l); }
    /**
     * Invia una notifica a tutti i listener che informa che la coda è stata distrutta
     */
    public void notifyQueueDestroyed() {
        for (Iterator i = life_listeners.iterator(); i.hasNext();) {
            QueueLifeListener l = (QueueLifeListener) i.next();
            try {
                l.queueDestroyed(this);
            }catch (Throwable t) {
                Logger.log(t);
            }
        }
    }
    /**
     * Invia una notifica a tutti i listener che un messaggio è stato aggiunto
     */
    public void notifyMessageAdded(Message m) {
        for (Iterator i = listeners.iterator(); i.hasNext();) {
            QueueListener l = (QueueListener) i.next();
            try {
                l.messageAdded(this,m);
            }catch (Throwable t) {
                Logger.log(t);
            }
        }
    }
    /**
     * Invia una notifica a tutti i listener che un messaggio è stato rimosso
     */
    public void notifyMessageRemoved(Message m) {
        for (Iterator i = listeners.iterator(); i.hasNext();) {
            QueueListener l = (QueueListener) i.next();
            try {
                l.messageRemoved(this,m);
            } catch (Throwable t) {
                Logger.log(t);
            }
        }
    }
    /**
     * Ritorna un messaggio, senza rimuoverlo
     */
    public synchronized Message getMessage(String id) {
        for (Iterator i = messages.iterator(); i.hasNext();) {
            Message m = (Message) i.next();
            if (m.getId().equals(id)) return m;
        }
        return null;
    }
    /**
     * Ritorna un messaggio, Rimuovendolo
     */
    public synchronized Message extractMessage(String id) {
        for (Iterator i = messages.iterator(); i.hasNext();) {
            Message m = (Message) i.next();
            if (m.getId().equals(id)) {
                i.remove();
                notifyMessageRemoved(m);
                return m;
            }
        }
        return null;
    }
    /**
     * Indica se il nodo corrente ha bloccato la coda
     */
    public boolean isLocked() {return locked; }
    /**
     * Ottiene una lista di tutti i messaggi nella coda
     */
    public List getMessages() { return Collections.unmodifiableList(messages); }
    /**
     * Raccoglie un messaggio di coordinamento
     */
    public void pushMessageOntoStack(Message msg) {
        if (msg.getType().equals("queue_post_message")) {            
            String serialized = msg.getBody();
            Logger.log("Add message to queue "+queueId);
            Logger.log("Message is:\n"+serialized);
            try {
                Message data = MessageFactory.decodeMessage(serialized);
                Logger.log("Deserialized OK");
                addMessage(data);
            } catch (IOException ioe) {
                Logger.log(ioe);
            }
            return;
        }
        if (msg.getType().equals("queue_delete_message")) {            
            String msgid = msg.getBody();
            Logger.log("Remove message from  queue "+queueId);
            Logger.log("Message is "+msgid);
            removeMessage(msgid);
            return;
        }
        if (msg.getType().equals("queue_unlock")) {            
            String lockid = msg.getBody();
            Logger.log("Remove lock "+lockid+"+ from  queue "+queueId);            
            removeLock(lockid);
            return;
        }
        stack.pushMessage(msg);
    }
    /**
     * Elimina un messaggio di coordinamento
     */
    public void popMessageFromStack(Message msg) {
        stack.removeMessage(msg);
    }       
    /**
     * Aggiunge un messaggio utente
     */
    public synchronized void addMessage(Message msg) {
        //System.out.println("----------------------------------------------");
        //System.out.println("Adding to queue "+queueId+" msg\n"+msg.asXML());
        //System.out.println("----------------------------------------------");
        messages.add(msg);
        notifyMessageAdded(msg);
    }    
    /**
     * rimuove un lock sulla coda
     */
    protected synchronized void removeLock(String lockid) {
        Node owner = stack.removeLockID(lockid);
        if (owner == null) return;
        if (owner == Node.getLocalNode()) locked = false;
        owners.remove(owner);
    }
    /**
     * Rimuove un messaggio utente dalla coda
     */ 
    public synchronized void removeMessage(String id) {
        //System.out.println("----------------------------------------------");
        //System.out.println("Adding to queue "+queueId+" msg\n"+msg.asXML());
        //System.out.println("----------------------------------------------");
        for (Iterator i = messages.iterator(); i.hasNext();) {
            Message m = (Message) i.next();
            if (m.getId().equals(id)) {
                i.remove();
                notifyMessageRemoved(m);
                break;
            }
        }
    }    
    /**
     * Metodo solo per DEBUG
     */
    public CoordinationMessagesStack getStack() { return stack; }
    
    /**
     * Cerca nello stack se sono presenti lengthautorizzazioni sufficienti ad ottenere il lock sulla coda
     * e se ci sono dichiara la coda "bloccata"
     */
    public synchronized boolean checkAndAcquireLock(String lockid) {
        // se la coda era già bloccata non possiamo bloccarla di nuovo !!
        if (locked) return false;
        
        if (stack.checkAndAcquireLock(lockid, owners)) {
            locked = true;            
            return true;
        }
        
        return false;
    }
        
    
    public synchronized void addOwner(Node owner) {
        owners.add(owner);
    }
    public synchronized void removeOwner(Node owner) {
        owners.remove(owner);
    }
    public synchronized List findAgreeableLockMessages() {
        return stack.findAgreeableLockMessages(owners);
    }
    /**
     * Ritorna il nodo più prioritario nell'insieme dei possibili nodi proprietari della coda al momento
     */
    public synchronized Node getMostImportantOwner() {        
        Node best = null;
        int prio = -1;
        for (Iterator i = owners.iterator(); i.hasNext();) {
            Node n = (Node) i.next();
            int np = n.getPriority();
            if ( np > prio) {
                prio = np;
                best = n;
            }
        }
        return best;
    }
    /**
     * Invocato per indicare che un nodo è stato rimosso del gruppo, perciò bisogna eliminare tutti i messaggi per esso
     */
    public synchronized void notifyNodeRemoved(Node n) {
        stack.removeAllMessagesByNode(n);
        if (owners.contains(n)) owners.remove(n);
    }
}
