
package qms.group;

import qms.messages.*;
import qms.queue.*;
import qms.io.*;
import qms.util.*;
import java.util.*;
import java.net.*;
/**
 * Gestisce tutto quello che riguarda l'invio di messaggi verso tutti gli altri nodi del gruppo
 * @author  Enrico
 */

public class Multicaster implements Runnable {
    protected Group group;
    protected Thread sender_thread;
    protected LinkedList message_queue;
    protected Messenger messenger;
    public static String ALL = "all_nodes";
    public List getUnsentMessages() { return new Vector(message_queue); }
    
    public Multicaster(Group group) {
        this.group = group;
        this.message_queue = new LinkedList();
        this.sender_thread = null;
        this.messenger = new Messenger();
    }
    /**
     * Realizza l'operazione di multicast. Questo metodo non è bloccante. I messaggi vengono posti in una coda che in modo asincrono viene servita
     */
    public void castMessage(Message msg) {
        if (msg.getRecipient().equals(ALL)) {
            //Logger.log("Multicasting msg:\n"+msg.asXML());
            for (Iterator i = group.iterator(); i.hasNext(); ) {
                Node n = (Node) i.next();
                Message copy = msg.copy();                
                copy.setRecipient(n.getName());
                copy.update();
                //Logger.log("Msg copy:\n"+msg.asXML());
                enqueueMessage(copy);
            }
        } else {
            enqueueMessage(msg);
        }
    }
    /**
     * Invia un messaggio in modo sincrono
     */
    public boolean castMessage(Message m, Node node) {
        return messenger.sendMessage(m,node);
    }
    
    /**
     * Aggiunge un messaggio alla coda di invio dei messaggi asincroni
     */
    protected synchronized void enqueueMessage(Message m) {
        synchronized(message_queue) {
            message_queue.addLast(m);
            if (sender_thread == null) {
                if (message_queue.size() == 1) {
                    startSenderThread();
                }
            }
        }
    }
    
    /**
     * Avvia il thread di supporto per l'invio asincrono dei messaggi
     */
    protected void startSenderThread() {
        if (sender_thread != null) {
            throw new IllegalStateException("Can't start two times sender thread for a multicaster");
        }
        sender_thread = new Thread(this);        
        sender_thread.setDaemon(true);
        sender_thread.start();
    }
    /**
     * Invia un messaggio
     */
    protected void sendMessage(Message m) {        
        Node node = group.lookupNode(m.getRecipient());
        if (node == null) {
            System.out.println("Can't find node "+m.getRecipient());
            m.notifyListener(false);
            return;
        }
        if (node.equals(Node.getLocalNode())) {            
            Node.getLocalAccessPoint().messageReceived(m) ;
        } else {
            URL access_point = node.getAccessPoint();
            messenger.sendMessage(m,node);            
        }
    }
    /**
     * Corpo del thread di supporto per l'invio dei messaggi
     */
    public void run() {
        while (true) {
            synchronized(message_queue) {
                if (message_queue.size() == 0) {
                    sender_thread = null;
                    return;
                }                
                Message msg = (Message) message_queue.removeFirst();
                //System.out.println("Sending MSG:\n"+msg);
                sendMessage(msg);
            }
        }
    }
}
