twoph1initiator.java

java实现的P2P多agent中间件

        registerPrepareInitiations(b);
    }

    /**
     * This method allows to register a user defined <code>Behaviour</code> in the
     * HANDLE_CONFIRM state. This behaviour would override the homonymous method.
     * This method also set the data store of the registered <code>Behaviour</code>
     * to the DataStore of this current behaviour. The registered behaviour can retrieve
     * the <code>confirm</code> ACLMessage object received from the datastore at the
     * <code>REPLY_KEY</code> key.
     * @param b the Behaviour that will handle this state
     */
    public void registerHandleConfirm(Behaviour b) {
        registerState(b, HANDLE_CONFIRM);
        b.setDataStore(getDataStore());
    }

    /**
     * This method allows to register a user defined <code>Behaviour</code> in the
     * HANDLE_DISCONFIRM state. This behaviour would override the homonymous method.
     * This method also set the data store of the registered <code>Behaviour</code>
     * to the DataStore of this current behaviour. The registered behaviour can retrieve
     * the <code>disconfirm</code> ACLMessage object received from the datastore at the
     * <code>REPLY_KEY</code> key.
     * @param b the Behaviour that will handle this state
     */
    public void registerHandleDisconfirm(Behaviour b) {
        registerState(b, HANDLE_DISCONFIRM);
        b.setDataStore(getDataStore());
    }

    /**
     * This method allows to register a user defined <code>Behaviour</code> in the
     * HANDLE_INFORM state. This behaviour would override the homonymous method.
     * This method also set the data store of the registered <code>Behaviour</code>
     * to the DataStore of this current behaviour. The registered behaviour can retrieve
     * the <code>inform</code> ACLMessage object received from the datastore at the
     * <code>REPLY_KEY</code> key.
     * @param b the Behaviour that will handle this state
     */
    public void registerHandleInform(Behaviour b) {
        registerState(b, HANDLE_INFORM);
        b.setDataStore(getDataStore());
    }

    /**
     * This method allows to register a user defined <code>Behaviour</code> in the
     * HANDLE_ALL_RESPONSES state. This behaviour would override the homonymous method.
     * This method also set the data store of the registered <code>Behaviour</code> to
     * the DataStore of this current behaviour. The registered behaviour can retrieve
     * the vector of ACLMessage confirms, disconfirms, informs, pending and responses
     * from the datastore at <code>ALL_CONFIRMS_KEY</code>, <code>ALL_DISCONFIRMS_KEY</code>,
     * <code>ALL_INFORMS_KEY</code>, <code>ALL_PH1_PENDINGS_KEY</code> and
     * <code>output</code> field.
     * @param b the Behaviour that will handle this state
     */
    public void registerHandleAllResponses(Behaviour b) {
        registerState(b, HANDLE_ALL_RESPONSES);
        b.setDataStore(getDataStore());
    }

    /* User CAN'T override these methods */
    //#APIDOC_EXCLUDE_BEGIN

  /**
   */
  protected String[] getToBeReset() {
  	if (toBeReset == null) {
			toBeReset = new String[] {
				HANDLE_CONFIRM, 
				HANDLE_DISCONFIRM, 
				HANDLE_INFORM, 
				HANDLE_NOT_UNDERSTOOD,
				HANDLE_FAILURE,
				HANDLE_OUT_OF_SEQ
			};
  	}
  	return toBeReset;
  }
    
    /**
     * Prepare vector containing queryIfs.
     * @param initiation queryIf passed in the constructor
     * @return Vector of queryIfs
     */
    protected final Vector prepareInitiations(ACLMessage initiation) {
      return prepareQueryIfs(initiation);
    }

    /**
     * This method sets for all prepared queryIfs <code>conversation-id</code> slot (with
     * value passed in the constructor), <code>protocol</code> slot and
     * <code>reply-with</code> slot with a unique value constructed by concatenating
     * receiver's agent name and phase number (i.e. 1). After that it sends all cfps.
     * @param initiations vector prepared in PREPARE_QUERYIFS state
     */
    protected final void sendInitiations(Vector initiations) {
      getDataStore().put(ALL_PENDINGS_KEY, new Vector());
      
      super.sendInitiations(initiations);
      
      totSessions = sessions.size();
    }
    
    /**
     * Check whether a reply is in-sequence and than update the appropriate Session
     * and removes corresponding queryif from vector of pendings.
     * @param reply message received
     * @return true if reply is compliant with flow of protocol, false otherwise
     */
    protected final boolean checkInSequence(ACLMessage reply) {
      boolean ret = false;
      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 
          ((Vector) getDataStore().get(ALL_RESPONSES_KEY)).add(reply);
          
          switch(perf) {
              case ACLMessage.CONFIRM: {
                  ((Vector) getDataStore().get(ALL_CONFIRMS_KEY)).add(reply);
                  break;
              }
              case ACLMessage.DISCONFIRM: {
                  ((Vector) getDataStore().get(ALL_DISCONFIRMS_KEY)).add(reply);
                  break;
              }
              case ACLMessage.INFORM: {
                  ((Vector) getDataStore().get(ALL_INFORMS_KEY)).add(reply);
                  break;
              }
          }
          updatePendings(inReplyTo);
          ret = true;
        }
        if (s.isCompleted()) {
            sessions.remove(inReplyTo);
        }
      } 
      return ret;
    }

    private void updatePendings(String key) {
    	Vector pendings = (Vector) getDataStore().get(ALL_PENDINGS_KEY);
      for(int i=0; i<pendings.size(); i++) {
          ACLMessage pendingMsg = (ACLMessage) pendings.get(i);
          if(pendingMsg.getReplyWith().equals(key)) {
              pendings.remove(i);
              break;
          }
      }
    }
    	
    /**
     * Check if there are still active sessions or if timeout is expired.
     * @param reply last message received
     * @return ALL_RESPONSES_RECEIVED or -1 (still active sessions)
     */
    protected final int checkSessions(ACLMessage reply) {
    	if (reply == null) {
    		// Timeout expired --> clear all sessions 
    		sessions.clear();
    	}
    	if (sessions.size() == 0) {
    		// We have finished --> fill the Vector of initiation messages for next 
    		// phase (unless already filled by the user)
        DataStore ds = getDataStore();
    		Vector nextPhMsgs = (Vector) ds.get(outputKey);
    		if (nextPhMsgs.size() == 0) {
	        Vector confirms = (Vector) ds.get(ALL_CONFIRMS_KEY);
	        Vector informs = (Vector) ds.get(ALL_INFORMS_KEY);
	        Vector pendings = (Vector) ds.get(ALL_PENDINGS_KEY);
	        fillNextPhInitiations(nextPhMsgs, confirms, informs, pendings);
    		}
        return ALL_RESPONSES_RECEIVED;
    	}
    	else {
    		// We are still waiting for some responses
    		return -1;
    	}
    }

    private void fillNextPhInitiations(Vector nextPhMsgs, Vector confirms, Vector informs, Vector pendings) {
    	if ((confirms.size()+informs.size()) == totSessions) {
    		// All responders replied with CONFIRM or INFORM --> Fill the vector 
    		// of initiation messages for next phase with ACCEPT_PROPOSAL
        for(int i=0; i<confirms.size(); i++) {
          ACLMessage msg = (ACLMessage) confirms.get(i);
          ACLMessage accept = msg.createReply();
          accept.setPerformative(ACLMessage.ACCEPT_PROPOSAL);
          nextPhMsgs.add(accept);
        }
    	}
    	else {
    		// At least one responder disconfirmed, failed or didn't reply --> Fill the vector 
    		// of initiation messages for next phase with REJECT_PROPOSALS
        for(int i=0; i<confirms.size(); i++) {
          ACLMessage msg = (ACLMessage) confirms.get(i);
          ACLMessage reject = msg.createReply();
          reject.setPerformative(ACLMessage.REJECT_PROPOSAL);
          nextPhMsgs.add(reject);
        }
        for(int i=0; i<pendings.size(); i++) {
          ACLMessage msg = (ACLMessage) pendings.get(i);
          ACLMessage reject = (ACLMessage) msg.clone();
          reject.setPerformative(ACLMessage.REJECT_PROPOSAL);
          nextPhMsgs.add(reject);
        }
    	}
    }
    
    /**
     * Initialize the data store.
     * @param msg Ignored
     */
    protected void initializeDataStore(ACLMessage msg) {
        super.initializeDataStore(msg);
        getDataStore().put(ALL_RESPONSES_KEY, new Vector());
        getDataStore().put(ALL_CONFIRMS_KEY, new Vector());
        getDataStore().put(ALL_DISCONFIRMS_KEY, new Vector());
        getDataStore().put(ALL_INFORMS_KEY, new Vector());
        getDataStore().put(outputKey, new Vector());
    }
    //#APIDOC_EXCLUDE_END
    		
    		
    
  protected ProtocolSession getSession(ACLMessage msg, int sessionIndex) {
    Vector pendings = (Vector) getDataStore().get(ALL_PENDINGS_KEY);
    pendings.add(msg);
		
  	return new Session("R" + hashCode()+  "_" + Integer.toString(sessionIndex) + "_" + TwoPhConstants.PH1);
  }
  
    /**
     * Inner class Session
     */
    class Session implements ProtocolSession, Serializable {
        // Session states 
        static final int INIT = 0;
        static final int REPLY_RECEIVED = 1;

        private int state = INIT;
        private String myId;

        public Session(String id) {
        	myId = id;
        }
        
				public String getId() {
					return myId;
				}
				

        /**
         * Return true if received ACLMessage is consistent with the protocol.
         * @param perf
         * @return Return true if received ACLMessage is consistent with the protocol
         */
        public boolean update(int perf) {
            if(state == INIT) {
                switch(perf) {
                    case ACLMessage.CONFIRM: ;
                    case ACLMessage.DISCONFIRM: ;
                    case ACLMessage.INFORM: ;
                    case ACLMessage.NOT_UNDERSTOOD:
                    case ACLMessage.FAILURE:
                      state = REPLY_RECEIVED;
                    	return true;
                    default: 
                    	return false;
                }
            }
            else {
                return false;
            }
        }

        public int getState() {
            return state;
        }

        public boolean isCompleted() {
            return (state == REPLY_RECEIVED);
        }
    }
}