twoph2initiator.java

java实现的P2P多agent中间件

    }

    /** This method allows to register a user-defined <code>Behaviour</code> in the
     * PREPARE_ACCEPTANCES 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. It is responsibility of the registered
     * behaviour to put the <code>Vector</code> of ACLMessage objects to be sent into
     * the datastore at the <code>ALL_ACCEPTANCES_KEY</code> key.
     * @param b the Behaviour that will handle this state
     */
    public void registerPrepareProposals(Behaviour b) {
        registerPrepareInitiations(b);
    }

    /**
     * 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_OLD_RESPONSE 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>failure, disconfirm or 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 registerHandleOldResponse(Behaviour b) {
        registerState(b, HANDLE_OLD_RESPONSE);
        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 received from the datastore at
     * <code>ALL_RESPONSES_RECEIVED_KEY</code>.
     * @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

		private String[] toBeReset = null;
			
	  /**
	   */
	  protected String[] getToBeReset() {
	  	if (toBeReset == null) {
				toBeReset = new String[] {
					HANDLE_INFORM,
					HANDLE_OLD_RESPONSE, 
					HANDLE_NOT_UNDERSTOOD,
					HANDLE_FAILURE,
					HANDLE_OUT_OF_SEQ
				};
	  	}
	  	return toBeReset;
	  }
	  
    /**
     * Returns vector of accept/reject-proposal stored in the data store at
     * key <code>inputKey</code> from previouse phase.
     * @param initiation ignored
     * @return Vector of accept/reject-proposal
     */
    protected final Vector prepareInitiations(ACLMessage initiation) {
        return prepareAcceptances(initiation);
    }

    /**
     * This method sets for all prepared accept/reject-proposal
     * <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. 2). After that it sends all accept/reject-proposal.
     * @param initiations vector prepared in PREPARE_ACCEPTANCES state
     */
    protected final void sendInitiations(Vector initiations) {
      getDataStore().put(ALL_PENDINGS_KEY, new Vector());
      
      super.sendInitiations(initiations);
    }
    
    /**
     * Check whether a reply is in-sequence and than update the appropriate Session
     * and removes corresponding accept/reject-proposal 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);
			          if (perf == ACLMessage.INFORM) {
			        		((Vector) getDataStore().get(ALL_INFORMS_KEY)).add(reply);
			          }
                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;
          }
      }
    }
    	
    private void oldResponse(ACLMessage reply) {
      String inReplyTo = reply.getInReplyTo();
      String sessionKey = inReplyTo.substring(0, inReplyTo.length() - 3) + "PH2";
			int perf = reply.getPerformative();
			if (perf == ACLMessage.FAILURE || perf == ACLMessage.NOT_UNDERSTOOD || perf == ACLMessage.DISCONFIRM) {
				sessions.remove(sessionKey);
				updatePendings(sessionKey);
			}
    }

    /**
     * Check if there are still active sessions or if timeout is expired.
     * @param reply last message received
     * @return ALL_RESPONSES_RECEIVED, -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 
    		return ALL_RESPONSES_RECEIVED;
    	}
    	else {
    		// We are still waiting for some responses
    		return -1;
    	}
    }

    /**
     * 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_INFORMS_KEY, 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.PH2);
  }
  
    /**
     * Inner class Session
     */
    class Session implements ProtocolSession, Serializable {
        // Possible 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.INFORM:
                    case ACLMessage.FAILURE:
                    case ACLMessage.NOT_UNDERSTOOD:
                      state = REPLY_RECEIVED;
	                    return true;
                    default: 
                    	return false;
                }
            }
            else {
                return false;
            }
        }

        public int getState() {
            return state;
        }

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