initiator.java

java实现的P2P多agent中间件

		String conversationID = createConvId(initiations);
		replyTemplate = MessageTemplate.MatchConversationId(conversationID);
		int cnt = 0; // counter of sessions
		Vector sentMessages = new Vector();
		for (Enumeration e=initiations.elements(); e.hasMoreElements(); ) {
			ACLMessage initiation = (ACLMessage) e.nextElement();
			if (initiation != null) {
				// Update the list of sessions on the basis of the receivers
				// FIXME: Maybe this should take the envelope into account first
				
				for (Iterator receivers = initiation.getAllReceiver(); receivers.hasNext(); ) {
					ACLMessage toSend = (ACLMessage)initiation.clone();
					toSend.setConversationId(conversationID);
					toSend.clearAllReceiver();
					AID r = (AID)receivers.next();
					toSend.addReceiver(r);
					ProtocolSession ps = getSession(toSend, cnt);
					if (ps != null) {
						String sessionKey = ps.getId();
						if (sessionKey == null) {
							sessionKey = "R" + System.currentTimeMillis()+  "_" + Integer.toString(cnt);
						}
						toSend.setReplyWith(sessionKey);
						sessions.put(sessionKey, ps);
						adjustReplyTemplate(toSend);
						cnt++;
					}
					myAgent.send(toSend);
					sentMessages.addElement(toSend);
				}
				
				// Update the timeout (if any) used to wait for replies according
				// to the reply-by field: get the miminum.  
				Date d = initiation.getReplyByDate();
				if (d != null) {
					long timeout = d.getTime()- currentTime;
					if (timeout > 0 && (timeout < minTimeout || minTimeout <= 0)) {
						minTimeout = timeout;
						deadline = d.getTime();
					}
				}
			}
		}
		// Replace the initiations Vector with that of actually sent messages
		getDataStore().put(ALL_INITIATIONS_K, sentMessages);
		
		// Finally set the MessageTemplate and timeout used in the RECEIVE_REPLY 
		// state to accept replies
		replyReceiver.setTemplate(replyTemplate);
		replyReceiver.setDeadline(deadline);
	}    
	//#APIDOC_EXCLUDE_END
	
	/**
	 * This method is called every time a <code>not-understood</code>
	 * message is received, which is not out-of-sequence according
	 * to the protocol rules.
	 * This default implementation does nothing; programmers might
	 * wish to override the method in case they need to react to this event.
	 * @param notUnderstood the received not-understood message
	 **/
	protected void handleNotUnderstood(ACLMessage notUnderstood) {
	}
	
	/**
	 * This method is called every time a <code>failure</code>
	 * message is received, which is not out-of-sequence according
	 * to the protocol rules.
	 * This default implementation does nothing; programmers might
	 * wish to override the method in case they need to react to this event.
	 * @param failure the received failure message
	 **/
	protected void handleFailure(ACLMessage failure) {
	}
	
	/**
	 * This method is called every time a 
	 * message is received, which is out-of-sequence according
	 * to the protocol rules.
	 * This default implementation does nothing; programmers might
	 * wish to override the method in case they need to react to this event.
	 * @param msg the received message
	 **/
	protected void handleOutOfSequence(ACLMessage msg) {
	}
	
	//#APIDOC_EXCLUDE_BEGIN
	/**
	 Attach a behaviour to the <code>Prepare-initiations</code>
	 protocol state.
	 @param b The behaviour object to be executed in the
	 <code>Prepare-initiations</code> state.
	 */
	protected void registerPrepareInitiations(Behaviour b) {
		registerState(b, PREPARE_INITIATIONS);
		b.setDataStore(getDataStore());
	}
	//#APIDOC_EXCLUDE_END
	
	/**
	 This method allows to register a user defined <code>Behaviour</code>
	 in the HANDLE_NOT_UNDERSTOOD 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>not-understood</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 registerHandleNotUnderstood(Behaviour b) {
		registerState(b, HANDLE_NOT_UNDERSTOOD);
		b.setDataStore(getDataStore());
	}
	
	/**
	 This method allows to register a user defined <code>Behaviour</code>
	 in the HANDLE_FAILURE 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</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 registerHandleFailure(Behaviour b) {
		registerState(b, HANDLE_FAILURE);
		b.setDataStore(getDataStore());
	}
	
	/**
	 This method allows to register a user defined <code>Behaviour</code>
	 in the HANDLE_OUT_OF_SEQ 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>out of sequence</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 registerHandleOutOfSequence(Behaviour b) {
		registerState(b, HANDLE_OUT_OF_SEQ);
		b.setDataStore(getDataStore());
	}
	
	/**
	 * reset this behaviour by putting a null ACLMessage as message
	 * to be sent
	 **/
	public void reset(){
		reset(null);
	}
	
	/**
	 * reset this behaviour
	 * @param msg is the ACLMessage to be sent
	 **/
	public void reset(ACLMessage msg){
		initiation = msg;
		reinit();
		super.reset();
	}
	
	/**
	 Re-initialize the internal state without performing a complete reset.
	 */
	protected void reinit() {
		replyReceiver.reset(null, MsgReceiver.INFINITE, getDataStore(),REPLY_K);
		sessions.clear(); 
		DataStore ds = getDataStore();
		ds.remove(INITIATION_K);
		ds.remove(ALL_INITIATIONS_K);
		ds.remove(REPLY_K);
	}
	
	/** 
	 Override the onStart() method to initialize the vectors that
	 will keep all the replies in the data store.
	 */
	public void onStart() {
		initializeDataStore(initiation);
	}
	
	/** 
	 Override the setDataStore() method to propagate this
	 setting to all children.
	 */
	public void setDataStore(DataStore ds) {
		super.setDataStore(ds);
		Iterator it = getChildren().iterator();
		while (it.hasNext()) {
			Behaviour b = (Behaviour) it.next();
			b.setDataStore(ds);
		}
	}
	
	//#APIDOC_EXCLUDE_BEGIN
	/**
	 Initialize the data store. 
	 **/
	protected void initializeDataStore(ACLMessage initiation){
		getDataStore().put(INITIATION_K, initiation);
	}
	//#APIDOC_EXCLUDE_END
	
	/**
	 Create a new conversation identifier to begin a new
	 interaction.
	 @param msgs A vector of ACL messages. If the first one has a
	 non-empty <code>:conversation-id</code> slot, its value is
	 used, else a new conversation identifier is generated.
	 */
	protected String createConvId(Vector msgs) {
		// If the conversation-id of the first message is set --> 
		// use it. Otherwise create a default one
		String convId = null;
		if (msgs.size() > 0) {
			ACLMessage msg = (ACLMessage) msgs.elementAt(0);
			if ((msg == null) || (msg.getConversationId() == null)) {
				convId = "C"+hashCode()+"_"+System.currentTimeMillis();
			}
			else {
				convId = msg.getConversationId();
			}
		}
		return convId;
	}
	
	//#APIDOC_EXCLUDE_BEGIN
	protected void adjustReplyTemplate(ACLMessage msg) {
		// If myAgent is among the receivers (strange case, but can happen)
		// then modify the replyTemplate to avoid intercepting the initiation
		// message msg as if it was a reply
		AID r = (AID) msg.getAllReceiver().next();
		if (myAgent.getAID().equals(r)) {
			replyTemplate = MessageTemplate.and(
					replyTemplate,
					MessageTemplate.not(MessageTemplate.MatchCustom(msg, true)));
		}
	}
	
	
	/**
	 Inner interface Session
	 */
	protected interface ProtocolSession {
		String getId();
		boolean update(int perf);
		int getState();
		boolean isCompleted();
	}
	//#APIDOC_EXCLUDE_END
}