/*
 * LockQueueAction.java
 *
 * Created on 26 marzo 2004, 13.37
 */

package qms.queue.actions;
import qms.queue.*;
import qms.group.*;
import qms.util.*;
import qms.messages.*;
/**
 * Semina un messaggio su tutti i nodi
 * @author  Enrico
 */
public class PostMessageAction implements Action, WaitingAction, MessageListener, QueueLifeListener {    
    
    protected String queueid;
    protected Message msg;
    protected Controller controller;
    protected boolean finished;    
    protected boolean waiting;
    protected boolean started;
    protected int sent_count;
    protected int required_count;
    protected WaitingAction waiter;
    
    public PostMessageAction(Message msg, Controller controller, WaitingAction waiter) {
        this.queueid = msg.getQueueId();
        this.controller = controller;
        this.finished = false;    
        this.msg = msg;
        this.waiting = true;
        this.waiter = waiter;
        this.sent_count = 0;
        this.required_count = controller.getGroup().size();
        this.started = false;
        controller.getQueues().getQueue(queueid).addQueueLifeListener(this);
    }
        
    public boolean isFinished() {
        if (finished) return true;
        
        if (sent_count >= required_count) {
            finished = true;
            waiter.canStart();            
        }
        return finished;
    }
    
    public void runStep() {       
        if (waiting) return;
        if (!started) {
            started = true;
            MessageQueue queue = controller.getQueues().getQueue(queueid);
            if (!queue.isLocked()) {
                throw new IllegalStateException("Trying to post a message without being the owner of the queue!!!");
            }
            Message cont = new Message( 
                    Node.getLocalNode().getName(),
                    Multicaster.ALL, 
                    queueid, 
                    "queue_post_message", 
                    msg.asXML());
            cont.setListener(this);
            controller.getMulticaster().castMessage(cont);                     
        }
    }
    public String toString() { return "distribute ("+sent_count+"/"+required_count+") msg "+msg; }
    
    public void canStart() {
        waiting = false;
    }
    
    public void notifyMessageDelivered(Message who, boolean status) {
        if (status) {
            sent_count++;                        
        }         
    }
    
    public void queueDestroyed(MessageQueue queue) {
        finished = true;
    }    
    
    public void cancel() {
        finished = true;
        waiter.cancel();
    }
        public void notifyNodeRemoved(Node n) {
        required_count--;
    }    
}
