contractnetinitiator.java

java实现的P2P多agent中间件

/*****************************************************************
JADE - Java Agent DEvelopment Framework is a framework to develop
multi-agent systems in compliance with the FIPA specifications.
Copyright (C) 2000 CSELT S.p.A.

GNU Lesser General Public License

This library is free software; you can redistribute it and/or
modify it under the terms of the GNU Lesser General Public
License as published by the Free Software Foundation, 
version 2.1 of the License. 

This library is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
Lesser General Public License for more details.

You should have received a copy of the GNU Lesser General Public
License along with this library; if not, write to the
Free Software Foundation, Inc., 59 Temple Place - Suite 330,
Boston, MA  02111-1307, USA.
*****************************************************************/

package jade.proto;

//#CUSTOM_EXCLUDE_FILE

import jade.core.*;
import jade.core.behaviours.*;
import jade.lang.acl.*;
import jade.proto.states.MsgReceiver;
import java.util.Date;
import java.util.Vector;
import java.util.Enumeration;
import jade.util.leap.Iterator;
import jade.util.leap.Map;
import jade.util.leap.HashMap;
import jade.util.leap.List;
import jade.util.leap.ArrayList;
import jade.util.leap.Serializable;

/**
 * This class implements the Fipa-Contract-Net interaction protocol
 * with an API similar and homogeneous to <code>AchieveREInitiator</code>.
 * <br>
 * This implementation works both for 1:1 and 1:N conversation and, of course,
 * implements the role of the initiator of the protocol.
 * <p>
 * The following is a brief description of the protocol. The programmer
 * should however refer to the 
 * <a href=http://www.fipa.org/specs/fipa00061/XC00061D.html>FIPA Spec</a>
 * for a complete description.
 * <p>
 * The initiator solicits proposals from other agents by sending
 * a <code>CFP</code> message that specifies the action to be performed
 * and, if needed, conditions upon its execution. The implementation of
 * the callback method <code>prepareCfps</code> must return the vector of
 * messages to be sent (eventually a single message with multiple receivers).
 * <p>
 * The responders can then reply by sending a <code>PROPOSE</code> message
 * including the preconditions that they set out for the action, for instance
 * the price or the time. 
 * Alternatively, responders may send a <code>REFUSE</code>, to refuse
 * the proposal or, eventually, a <code>NOT-UNDERSTOOD</code> to communicate
 * communication problems.
 * This first category of reply messages has been here identified as a 
 * response and can be handled via the <code>handleAllResponses</code>
 * callback method.
 * Specific handle callback methods for each type of communicative act are also
 * available when the programmer wishes to handle them separately:
 * <code>handleRefuse, handlePropose, handleNotUnderstood</code>.
 * <p> 
 * The initiator can evaluate all the received proposals
 * and make its choice of which agent proposals will be accepted and
 * which will be rejected. 
 * This class provides two ways for this evaluation. It can be done
 * progressively each time a new <code>PROPOSE</code> message is
 * received and a new call to the <code>handlePropose</code> callback 
 * method is executed
 * or,
 * in alternative, it can be done just once when all the <code>PROPOSE</code>
 * messages have been collected (or the <code>reply-by</code> deadline has
 * elapsed) and a single call to the 
 * <code>handleAllResponses</code> callback method is executed. 
 * In both cases, the second parameter of the method, i.e. the Vector
 * <code>acceptances</code>, must be filled with the appropriate
 * <code>ACCEPT/REJECT-PROPOSAL</code> messages.
 * Notice that, for the first case, the method <code>skipNextResponses</code>
 * has been provided that, if called by the programmer
 * when waiting for <code>PROPOSE</code>
 * messages, allows to skip to the next state and ignore all the 
 * responses and proposals that have not yet been received.
 * <p>
 * Once the responders whose proposal has been accepted (i.e. those that have
 * received a <code>ACCEPT-PROPOSAL</code> message) have completed
 * the task, they can, finally, 
 * respond with an 
 * <code>INFORM</code> of the result of the action (eventually just that the 
 * action has been done) or with a <code>FAILURE</code> if anything went wrong.
 * This second category of reply messages has been here identified as a
 * result notification and can be handled via the 
 * <code>handleAllResultNotification</code> callback method.
 * Again, specific handle callback
 * methods for each type of communicative act are also
 * available when the programmer wishes to handle them separately:
 * <code>handleRefuse, handleInform</code>.
 * <p>
 * If a message were received, with the same value of this 
 * <code>conversation-id</code>, but that does not comply with the FIPA 
 * protocol, than the method <code>handleOutOfSequence</code> would be called.
 * <p>
 * This class can be extended by the programmer by overriding all the needed
 * handle methods or, in alternative, appropriate behaviours can be
 * registered for each handle via the <code>registerHandle</code>-type
 * of methods. This last case is more difficult to use and proper
 * care must be taken to properly use the <code>DataStore</code> of the
 * <code>Behaviour</code> as a shared memory mechanism with the
 * registered behaviour.
 * <p>
 * Read carefully the section of the 
 * <a href="..\..\..\programmersguide.pdf"> JADE programmer's guide </a>
 * that describes
 * the usage of this class.

 * @author Giovanni Caire - TILab
 * @author Fabio Bellifemine - TILab
 * @author Tiziana Trucco - TILab
 * @author Marco Monticone - TILab
 * @version $Date: 2005-09-16 15:54:46 +0200 (ven, 16 set 2005) $ $Revision: 5780 $
 * @since JADE2.5
 * @see ContractNetResponder
 * @see AchieveREInitiator
 * @see <a href=http://www.fipa.org/specs/fipa00029/XC00029F.html>FIPA Spec</a>
 **/
public class ContractNetInitiator extends Initiator {
	
  // Private data store keys (can't be static since if we register another instance of this class as state of the FSM 
  // using the same data store the new values overrides the old one. 
  /** 
   * key to retrieve from the DataStore of the behaviour the ACLMessage 
   *	object passed in the constructor of the class.
   **/
  public final String CFP_KEY = INITIATION_K;
  /** 
   * key to retrieve from the DataStore of the behaviour the vector of
   * CFP ACLMessage objects that have to be sent.
   **/
  public final String ALL_CFPS_KEY = ALL_INITIATIONS_K;
  /** 
   * key to retrieve from the DataStore of the behaviour the vector of
   * ACCEPT/REJECT_PROPOSAL ACLMessage objects that have to be sent 
   **/
  public final String ALL_ACCEPTANCES_KEY = "__all-acceptances" +hashCode();
  /** 
   * key to retrieve from the DataStore of the behaviour the last
   * ACLMessage object that has been received (null if the timeout
   * expired). 
   **/
  public final String REPLY_KEY = REPLY_K;
  /** 
   * key to retrieve from the DataStore of the behaviour the vector of
   * ACLMessage objects that have been received as response.
   **/
  public final String ALL_RESPONSES_KEY = "__all-responses" + hashCode();
  /** 
   * key to retrieve from the DataStore of the behaviour the vector of
   * ACLMessage objects that have been received as result notifications.
   **/
  public final String ALL_RESULT_NOTIFICATIONS_KEY = "__all-result-notifications" +hashCode();
 
  // FSM states names
  private static final String HANDLE_PROPOSE = "Handle-propose";
  private static final String HANDLE_REFUSE = "Handle-refuse";
  private static final String HANDLE_INFORM = "Handle-inform";
  private static final String HANDLE_ALL_RESPONSES = "Handle-all-responses";
  private static final String HANDLE_ALL_RESULT_NOTIFICATIONS = "Handle-all-result-notifications";
	
  // States exit values
  private static final int ALL_RESPONSES_RECEIVED = 1;
  private static final int ALL_RESULT_NOTIFICATIONS_RECEIVED = 2;
	
  // When step == 1 we deal with CFP and responses
  // When step == 2 we deal with ACCEPT/REJECT_PROPOSAL and result notifications
	private int step = 1;;
  // If set to true all responses not yet received are skipped
  private boolean skipNextRespFlag = false;
	
  /**
   * Constructor for the class that creates a new empty DataStore
   * @see #ContractNetInitiator(Agent, ACLMessage, DataStore)
   **/
  public ContractNetInitiator(Agent a, ACLMessage cfp){
		this(a, cfp, new DataStore());
  }

	/**
   * Constructs a <code>ContractNetInitiator</code> behaviour
   * @param a The agent performing the protocol
   * @param msg The message that must be used to initiate the protocol.
   * Notice that the default implementation of the 
   * <code>prepareCfps</code>
   * method returns
   * an array composed of that message only.
   * @param s The <code>DataStore</code> that will be used by this 
   * <code>ContractNetInitiator</code>
   */
  public ContractNetInitiator(Agent a, ACLMessage cfp, DataStore store) {
		super(a, cfp, store);
		
		// Register the FSM transitions specific to the ContractNet protocol
		registerTransition(CHECK_IN_SEQ, HANDLE_PROPOSE, ACLMessage.PROPOSE);		
		registerTransition(CHECK_IN_SEQ, HANDLE_REFUSE, ACLMessage.REFUSE);
		registerTransition(CHECK_IN_SEQ, HANDLE_INFORM, ACLMessage.INFORM);		
		registerDefaultTransition(HANDLE_PROPOSE, CHECK_SESSIONS);
		registerDefaultTransition(HANDLE_REFUSE, CHECK_SESSIONS);
		registerDefaultTransition(HANDLE_INFORM, CHECK_SESSIONS);
		registerTransition(CHECK_SESSIONS, HANDLE_ALL_RESPONSES, ALL_RESPONSES_RECEIVED);
		registerTransition(CHECK_SESSIONS, HANDLE_ALL_RESULT_NOTIFICATIONS, ALL_RESULT_NOTIFICATIONS_RECEIVED);
		registerDefaultTransition(HANDLE_ALL_RESPONSES, SEND_INITIATIONS, getToBeReset());
			
		// Create and register the states specific to the ContractNet protocol
		Behaviour b = null;
		// HANDLE_PROPOSE
		b = new OneShotBehaviour(myAgent) {
	  	private static final long     serialVersionUID = 3487495895819003L;
	  	
			public void action() {
				Vector acceptances = (Vector) getDataStore().get(ALL_ACCEPTANCES_KEY);
				ACLMessage propose = (ACLMessage) getDataStore().get(REPLY_K);
				handlePropose(propose, acceptances);
			}
		};
		b.setDataStore(getDataStore());		
		registerState(b, HANDLE_PROPOSE);
		
		// HANDLE_REFUSE
		b = new OneShotBehaviour(myAgent) {
	  	private static final long     serialVersionUID = 3487495895819004L;
	  	
			public void action() {
			    handleRefuse((ACLMessage) getDataStore().get(REPLY_K));
			}
		};
		b.setDataStore(getDataStore());		
		registerState(b, HANDLE_REFUSE);
		
		// HANDLE_INFORM
		b = new OneShotBehaviour(myAgent) {
	  	private static final long     serialVersionUID = 3487495895818006L;
	  	
			public void action() {
			    handleInform((ACLMessage) getDataStore().get(REPLY_K));
			}
		};
		b.setDataStore(getDataStore());		
		registerState(b, HANDLE_INFORM);
	
		// HANDLE_ALL_RESPONSES
		b = new OneShotBehaviour(myAgent) {
		
			public void action() {
				Vector responses = (Vector) getDataStore().get(ALL_RESPONSES_KEY);
				Vector acceptances = (Vector) getDataStore().get(ALL_ACCEPTANCES_KEY);
		    handleAllResponses(responses, acceptances);
			}
	  };
		b.setDataStore(getDataStore());		
		registerState(b, HANDLE_ALL_RESPONSES);
	
		// HANDLE_ALL_RESULT_NOTIFICATIONS
		b = new OneShotBehaviour(myAgent) {
		
			public void action() {
		    handleAllResultNotifications((Vector) getDataStore().get(ALL_RESULT_NOTIFICATIONS_KEY));
			}
		};
		b.setDataStore(getDataStore());		
		registerLastState(b, HANDLE_ALL_RESULT_NOTIFICATIONS);	
	}

  //#APIDOC_EXCLUDE_BEGIN
  /**
   */    
  protected Vector prepareInitiations(ACLMessage initiation) {
  	return prepareCfps(initiation);
  }
  
  /**
     Create and initialize the Sessions and sends the initiation messages
   */    
  protected void sendInitiations(Vector initiations) {
		// By default the initiations parameter points to the Vector of the CFPs. 
		// However at step 2 we need to deal with the acceptances
		if (step == 2) {
			initiations = (Vector) getDataStore().get(ALL_ACCEPTANCES_KEY);
		}
		
		super.sendInitiations(initiations);
  }
  
  /**
     Create and initialize the Sessions and sends the initiation messages
   *    
  protected void sendInitiations(Vector initiations) {
		long currentTime = System.currentTimeMillis();
		long minTimeout = -1;
		long deadline = -1;

		// By default the initiations parameter points to the Vector of the CFPs. 
		// However at step 2 we need to deal with the acceptances
		if (step == 2) {
			initiations = (Vector) getDataStore().get(ALL_ACCEPTANCES_KEY);
		}

		String conversationID = createConvId(initiations);
		replyTemplate = MessageTemplate.MatchConversationId(conversationID);
	  int cnt = 0; // counter of sessions
	  for (Enumeration e=initiations.elements(); e.hasMoreElements(); ) {
			ACLMessage msg = (ACLMessage) e.nextElement();
			if (msg != null) {
		    // Update the list of sessions on the basis of the receivers
		    // FIXME: Maybe this should take the envelope into account first
		    
		    ACLMessage toSend = (ACLMessage)msg.clone();
		    toSend.setConversationId(conversationID);
		    for (Iterator receivers = msg.getAllReceiver(); receivers.hasNext(); ) {
					toSend.clearAllReceiver();
					AID r = (AID)receivers.next();
					toSend.addReceiver(r);
					if (step == 1 || toSend.getPerformative() == ACLMessage.ACCEPT_PROPOSAL) {
						String sessionKey = "R" + hashCode()+  "_" + Integer.toString(cnt);
						toSend.setReplyWith(sessionKey);
						sessions.put(sessionKey, new Session(step));
						adjustReplyTemplate(toSend);
						cnt++;
					}
					myAgent.send(toSend);
		    }
		  
	    	// Update the timeout (if any) used to wait for replies according
	    	// to the reply-by field
	    	// Get the miminum (if we are in step 2 only consider ACCEPT_PROPOSALs) 
				if (step == 1 || toSend.getPerformative() == ACLMessage.ACCEPT_PROPOSAL) {
	    		Date d = msg.getReplyByDate();
	    		if (d != null) {
						long timeout = d.getTime()- currentTime;
						if (timeout > 0 && (timeout < minTimeout || minTimeout <= 0)) {
		    			minTimeout = timeout;
		    			deadline = d.getTime();
						}
	    		}
				}
			}
	  }
	  // Finally set the MessageTemplate and timeout used in the RECEIVE_REPLY 
	  // state to accept replies
	  replyReceiver.setTemplate(replyTemplate);
	  replyReceiver.setDeadline(deadline);
  }*/
  
  /**
     Check whether a reply is in-sequence and update the appropriate Session
   */    
  protected boolean checkInSequence(ACLMessage reply) {
  	boolean ret = false;
		String inReplyTo = reply.getInReplyTo();
		Session s = (Session) sessions.get(inReplyTo);
		if (s != null) {
			int perf = reply.getPerformative();