proposeinitiator.java

java实现的P2P多agent中间件

          }
        }
        public int onEnd() {
          return ret;
        }
      };
    b.setDataStore(getDataStore());		
    registerState(b, CHECK_IN_SEQ);

    // HANDLE_ALL_RESPONSES
    b = new OneShotBehaviour(myAgent) {
        public void action() {
          handleAllResponses((Vector) getDataStore().get(ALL_RESPONSES_KEY));
        }
      };
    b.setDataStore(getDataStore());		
    registerState(b, HANDLE_ALL_RESPONSES);
	
    // HANDLE_ACCEPT_PROPOSAL
    b = new OneShotBehaviour(myAgent) {
        public void action() {
          handleAcceptProposal((ACLMessage) getDataStore().get(REPLY_KEY));
        }
      };
    b.setDataStore(getDataStore());		
    registerState(b, HANDLE_ACCEPT_PROPOSAL);
		
    // HANDLE_REJECT_PROPOSAL
    b = new OneShotBehaviour(myAgent) {
        public void action() {
          handleRejectProposal((ACLMessage) getDataStore().get(REPLY_KEY));
        }
      };
    b.setDataStore(getDataStore());		
    registerState(b, HANDLE_REJECT_PROPOSAL);
	
    // HANDLE_NOT_UNDERSTOOD
    b = new OneShotBehaviour(myAgent) {
        private static final long     serialVersionUID = 3487495895818005L;
  	
        public void action() {
          handleNotUnderstood((ACLMessage) getDataStore().get(REPLY_KEY));
        }
      };
    b.setDataStore(getDataStore());		
    registerState(b, HANDLE_NOT_UNDERSTOOD);
	
    // HANDLE_OUT_OF_SEQ
    b = new OneShotBehaviour(myAgent) {
        private static final long     serialVersionUID = 3487495895818008L;
  	
        public void action() {
          handleOutOfSequence((ACLMessage) getDataStore().get(REPLY_KEY));
        }
      };
    b.setDataStore(getDataStore());		
    registerState(b, HANDLE_OUT_OF_SEQ);
	
    // CHECK_SESSIONS
    b = new OneShotBehaviour(myAgent) {
        int ret;
        private static final long     serialVersionUID = 3487495895818009L;
  	
        public void action() {
          ACLMessage reply = (ACLMessage) getDataStore().get(REPLY_KEY);
          ret = checkSessions(reply);
        }		
        public int onEnd() {
          return ret;
        }
      };
    b.setDataStore(getDataStore());		
    registerState(b, CHECK_SESSIONS);
	
    // CHECK_AGAIN
    
    b = new OneShotBehaviour(myAgent) {
        public void action() {
        }
        public int onEnd() {
          return sessions.size();
        }
      };
    b.setDataStore(getDataStore());		
    registerState(b, CHECK_AGAIN);

    // DUMMY_FINAL
    b = new OneShotBehaviour(myAgent) {
        private static final long     serialVersionUID = 3487495895818010L;
  	
        public void action() {}
      };
    registerLastState(b, DUMMY_FINAL);
  }

  /**
   * Initialize the data store. 
   */
  protected void initializeDataStore(ACLMessage msg){
    getDataStore().put(INITIATION_K, initiation);
    Vector l = new Vector();
    getDataStore().put(ALL_RESPONSES_KEY, l);
  }

  /**
   * Create and initialize the Sessions and sends the initiation messages.
   * This method is called internally by the framework and is not intended 
   * to be called by the user
   */    
  protected void sendInitiations(Vector initiations) {
    long currentTime = System.currentTimeMillis();
    long minTimeout = -1;
    long deadline = -1;

    String conversationID = createConvId(initiations);
    replyTemplate = MessageTemplate.MatchConversationId(conversationID);
    int cnt = 0; // counter of sessions
    for (Enumeration e=initiations.elements(); e.hasMoreElements(); ) {
      ACLMessage proposal = (ACLMessage) e.nextElement();
      if (proposal != 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)proposal.clone();
        toSend.setProtocol(InteractionProtocol.FIPA_PROPOSE);
        toSend.setConversationId(conversationID);
        for (Iterator receivers = proposal.getAllReceiver(); receivers.hasNext(); ) {
          toSend.clearAllReceiver();
          AID r = (AID)receivers.next();
          toSend.addReceiver(r);
          String sessionKey = "R" + hashCode()+  "_" + Integer.toString(cnt);
          toSend.setReplyWith(sessionKey);
          sessions.put(sessionKey, new Session());
          adjustReplyTemplate(toSend);
          myAgent.send(toSend);
          // Store the propose message actually sent. It can 
          // be useful to retrieve it to create the CANCEL message
          getDataStore().put(r, toSend);
          cnt++;
        }
			  
        // Update the timeout (if any) used to wait for replies according
        // to the reply-by field. Get the miminum  
        Date d = proposal.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.
   * This method is called internally by the framework and is not intended 
   * to be called by the user       
   */    
  protected boolean checkInSequence(ACLMessage reply) {
    String inReplyTo = reply.getInReplyTo();
    Session s = (Session) sessions.get(inReplyTo);
    if (s != null) {
      int perf = reply.getPerformative();
      if (s.update(perf)) {
        // The reply is compliant to the protocol 
        switch (s.getState()) {
        case Session.RESULT_NOTIFICATION_RECEIVED:
        case Session.NEGATIVE_RESPONSE_RECEIVED:
          // The reply is a response
          Vector allRsp = (Vector) getDataStore().get(ALL_RESPONSES_KEY);
          allRsp.addElement(reply);
          break;
        default:
          // Something went wrong. Return false --> we will go to the HANDLE_OUT_OF_SEQ state
          return false;
        }
        // If the session is completed then remove it.
        if (s.isCompleted()) {
          sessions.remove(inReplyTo);
        }
        return true;
      }
    }
    return false;
  }
        
  /**
   * Check the status of the sessions after the reception of the last reply
   * or the expiration of the timeout.
   * This method is called internally by the framework and is not intended 
   * to be called by the user       
   */    
  protected int checkSessions(ACLMessage reply) {
    int ret = -1;
    if (getLastExitValue() == MsgReceiver.TIMEOUT_EXPIRED) {
      if (!allResponsesReceived) {
        // Special case 1: Timeout has expired
        // Remove all the sessions for which no response has been received yet
        List sessionsToRemove = new ArrayList(sessions.size());
        for (Iterator i=sessions.keySet().iterator(); i.hasNext(); ) {
          Object key = i.next();
          Session s = (Session)sessions.get(key);
          if ( s.getState() == Session.INIT ) {
            sessionsToRemove.add(key);
          }
        }
        for (Iterator i=sessionsToRemove.iterator(); i.hasNext(); ) {
          sessions.remove(i.next());
        }
        sessionsToRemove=null;  //frees memory
      }
      else {
        // Special case 2: All responses have already been received 
        // and an additional timeout (set e.g. through replyReceiver.setDeadline())
        // expired. Remove all sessions
        sessions.clear();
      }
    }
	  	
    if (!allResponsesReceived) {
      // Check whether all responses have been received (this is the 
      // case when no active session is still in the INIT state).
      allResponsesReceived = true;
      Iterator it = sessions.values().iterator();
      while (it.hasNext()) {
        Session s = (Session) it.next();
        if (s.getState() == Session.INIT) {
          allResponsesReceived = false;
          break;
        }
      }
      if (allResponsesReceived) {
        // Set an infite timeout to the replyReceiver.
        replyReceiver.setDeadline(MsgReceiver.INFINITE);
        ret = ALL_RESPONSES_RECEIVED;
      }
    }
    else {
      // Note that this check must be done only if the HANDLE_ALL_RESPONSES
      // has already been visited.
      if (sessions.size() == 0) {
        // There are no more active sessions --> Terminate
        ret = TERMINATED;
      }
    }
    return ret;
  }

  /**
   * This method must return the vector of ACLMessage objects to be
   * sent. It is called in the first state of this protocol.
   * This default implementation just returns the ACLMessage object (a PROPOSE)
   * passed in the constructor. Programmers might prefer to override
   * this method in order to return a vector of PROPOSE objects
   * for 1:N conversations
   * or also to prepare the messages during the execution of the behaviour.
   * @param propose the ACLMessage object passed in the constructor
   * @return a Vector of ACLMessage objects. The value of the slot
   * <code>reply-with</code> is ignored and regenerated automatically
   * by this class.
   **/    
  protected Vector prepareInitiations(ACLMessage propose) {
		Vector v = new Vector(1);
		v.addElement(propose);
		return v;
  }

  /**