df.java

java实现的P2P多agent中间件

			try {
				n = Integer.parseInt(sValue);
			} catch (Exception e) {
				logger.log(Logger.WARNING, "\""+sValue+"\" is not a valid value for integer parameter" + name, e);
			}
		}
		return n;
	}
	
	private String getValue(String s) {
		return (s != null ? s : "null");
	}
	
	
	/**
	 Cleanup <em>DF</em> on exit. This method performs all necessary
	 cleanup operations during agent shutdown.
	 */
	protected void takeDown() {
		//#PJAVA_EXCLUDE_BEGIN
		if (amsSubscriber != null) {
			// Unsubscribe from the AMS
			send(amsSubscriber.getCancel());
		}
		
		if (tbf != null) {
			tbf.interrupt();
		}
		//#PJAVA_EXCLUDE_END
		
		if(gui != null) {
			gui.disposeAsync();
		}
		DFAgentDescription dfd = new DFAgentDescription();
		dfd.setName(getAID());
		Iterator it = parents.iterator();
		while(it.hasNext()) {
			AID parentName = (AID)it.next();
			try {
				DFService.deregister(this, parentName, dfd);
			}
			catch(FIPAException fe) {
				fe.printStackTrace();
			}
		}
	}
	
	
	////////////////////////////////////
	// Recursive search
	////////////////////////////////////
	
	/**
	 Add the behaviour handling a recursive search.
	 If constraints contains a null search_id, then a new one is generated and
	 the new search_id is stored into searchIdCache 
	 for later check (i.e. to avoid search loops). 
	 */
	private void performRecursiveSearch(List localResults, DFAgentDescription dfd, SearchConstraints constraints, Search action){
		int maxRes = getActualMaxResults(constraints);
		int maxDep = constraints.getMaxDepth().intValue();
		
		// Create the new SearchConstraints.
		SearchConstraints newConstr = new SearchConstraints();
		// Max-depth decreased by 1
		newConstr.setMaxDepth(new Long ((long) (maxDep - 1)));
		// Max-results decreased by the number of items found locally
		newConstr.setMaxResults(new Long((long) (maxRes - localResults.size())));
		// New globally unique search-id unless already present
		String searchId = constraints.getSearchId();
		if (searchId == null) {
			searchId = getName() + String.valueOf(searchIdCnt++) + System.currentTimeMillis();
			if (searchIdCnt >= SEARCH_ID_CACHE_SIZE) {
				searchIdCnt = 0;
			}
			searchIdCache.add(searchId);
		}
		newConstr.setSearchId(searchId);
		
		if(logger.isLoggable(Logger.CONFIG))
			logger.log(Logger.CONFIG,"Activating recursive search: "+localResults.size()+" item(s) found locally. "+maxRes+" expected. Search depth is "+maxDep+". Search ID is "+searchId+". Propagating search to "+children.size()+" federated DF(s)");
		
		// Add the behaviour handling the search on federated DFs
		addBehaviour(new RecursiveSearchHandler(localResults, dfd, newConstr, action));
	}
	
	/** 
	 Inner class RecursiveSearchHandler.
	 This is a behaviour handling recursive searches i.e. searches that
	 must be propagated to children (federated) DFs.
	 */
	private class RecursiveSearchHandler extends AchieveREInitiator {
		private static final long DEFAULTTIMEOUT = 300000; // 5 minutes
		
		private List results;
		private DFAgentDescription template;
		private SearchConstraints constraints;
		private Search action;
		private int maxExpectedResults;
		private int receivedResults;
		
		/**
		 Construct a new RecursiveSearchHandler.
		 @param results The search results. Initially this includes the items found
		 locally.
		 @param template The DFAgentDescription used as tamplate for the search.
		 @param constraints The constraints for the search to be propagated.
		 @param action The original Search action. This is used as a key to retrieve
		 the incoming REQUEST message.
		 */
		private RecursiveSearchHandler(List results, DFAgentDescription template, SearchConstraints constraints, Search action) {
			super(df.this, null);
			
			this.results = results;
			this.template = template;
			this.constraints = constraints;
			this.action = action;
			
			maxExpectedResults = constraints.getMaxResults().intValue();
			receivedResults = 0;
		}
		
		/**
		 We broadcast the search REQUEST to all children (federated) DFs in parallel.
		 */ 
		protected Vector prepareRequests(ACLMessage request) {
			Vector requests = null;
			ACLMessage incomingRequest = (ACLMessage) pendingRequests.get(action);
			if (incomingRequest != null) {
				Date deadline = incomingRequest.getReplyByDate();
				if (deadline == null) {
					deadline = new Date(System.currentTimeMillis() + DEFAULTTIMEOUT);
				}
				requests = new Vector(children.size());
				Iterator it = children.iterator();
				while (it.hasNext()) {
					AID childDF = (AID) it.next();
					ACLMessage msg = DFService.createRequestMessage(myAgent, childDF, FIPAManagementVocabulary.SEARCH, template, constraints);
					msg.setReplyByDate(deadline);
					requests.addElement(msg);
				}
			}
			return requests;
		}
		
		/** 
		 As long as we receive the replies we update the results. If we reach the
		 max-results we send back the notification to the requester and discard 
		 successive replies.
		 */
		protected void handleInform(ACLMessage inform) {
			if(logger.isLoggable(Logger.CONFIG))
				logger.log(Logger.CONFIG,"Recursive search result received from "+inform.getSender().getName()+"."); 
			int cnt = 0;
			if (receivedResults < maxExpectedResults) {
				try {
					DFAgentDescription[] dfds = DFService.decodeResult(inform.getContent());
					for (int i = 0; i < dfds.length; ++i) {
						// We add the item only if not already present
						if (addResult(dfds[i])) {
							receivedResults++;
							cnt++;
							if (receivedResults >= maxExpectedResults) {
								sendPendingNotification(action, results);
							}
						}
					}
				}
				catch (Exception e) {
					if(logger.isLoggable(Logger.SEVERE))
						logger.log(Logger.SEVERE,"WARNING: Error decoding reply from federated DF "+inform.getSender().getName()+" during recursive search ["+e.toString()+"]."); 
				}
			}
			if(logger.isLoggable(Logger.CONFIG))
				logger.log(Logger.CONFIG,cnt+" new items found in recursive search."); 
		}
		
		protected void handleRefuse(ACLMessage refuse) {
			if(logger.isLoggable(Logger.WARNING))
				logger.log(Logger.WARNING,"REFUSE received from federated DF "+refuse.getSender().getName()+" during recursive search.");
		}
		protected void handleFailure(ACLMessage failure) {
			// FIXME: In general this is due to a federation loop (search-id already used)
			// In this case no warning must be printed --> We should use FINE log level in that case
			if(logger.isLoggable(Logger.WARNING))
				logger.log(Logger.WARNING,"FAILURE received from federated DF "+failure.getSender().getName()+" during recursive search.");
		}
		protected void handleNotUnderstood(ACLMessage notUnderstood) {
			if(logger.isLoggable(Logger.WARNING))
				logger.log(Logger.WARNING,"NOT_UNDERSTOOD received from federated DF "+notUnderstood.getSender().getName()+" during recursive search.");
		}
		protected void handleOutOfSequence(ACLMessage msg) {
			if(logger.isLoggable(Logger.WARNING))
				logger.log(Logger.WARNING,"Out of sequence message "+ACLMessage.getPerformative(msg.getPerformative())+" received from "+msg.getSender().getName()+" during recursive search.");
		}
		
		public int onEnd() {
			// Send back the notification to the originator of the 
			// search (unless already sent)
			if (receivedResults < maxExpectedResults) {
				sendPendingNotification(action, results);
			}
			return super.onEnd();
		}
		
		private boolean addResult(DFAgentDescription newDfd) {
			Iterator it = results.iterator();
			while (it.hasNext()) {
				DFAgentDescription dfd = (DFAgentDescription) it.next();
				if (dfd.getName().equals(newDfd.getName())) {
					return false;
				}
			}
			results.add(newDfd);
			return true;
		}    
	} // END of inner class RecursiveSearchHandler
	
	
	
	//////////////////////////////////////////////////////
	// Methods actually accessing the DF Knowledge base
	//////////////////////////////////////////////////////
	
	void DFRegister(DFAgentDescription dfd) throws AlreadyRegistered {
		
		//checkMandatorySlots(FIPAAgentManagementOntology.REGISTER, dfd);
		Object old = agentDescriptions.register(dfd.getName(), dfd);
		if(old != null)
			throw new AlreadyRegistered();
		
		if(isADF(dfd)) {
			if(logger.isLoggable(Logger.INFO))
				logger.log(Logger.INFO,"Added federation "+dfd.getName().getName()+" --> "+getName());
			children.add(dfd.getName());
			try {
				gui.addChildren(dfd.getName());
			} catch (Exception ex) {}
		}
		// for subscriptions
		subManager.handleChange(dfd, null);
		
		try{ //refresh the GUI if shown, exception thrown if the GUI was not shown
			gui.addAgentDesc(dfd.getName());
			gui.showStatusMsg("Registration of agent: " + dfd.getName().getName() + " done.");
		}catch(Exception ex){}
		
	}
	
	//this method is called into the prepareResponse of the DFFipaAgentManagementBehaviour to perform a Deregister action
	void DFDeregister(DFAgentDescription dfd) throws NotRegistered {
		//checkMandatorySlots(FIPAAgentManagementOntology.DEREGISTER, dfd);
		Object old = agentDescriptions.deregister(dfd.getName());
		
		if(old == null)
			throw new NotRegistered();
		
		if (children.remove(dfd.getName()))
			try {
				gui.removeChildren(dfd.getName());
			} catch (Exception e) {}
			// for subscriptions
			dfd.clearAllServices(); //clear all services since we are deregistering
			subManager.handleChange(dfd, (DFAgentDescription)old);
			try{
				// refresh the GUI if shown, exception thrown if the GUI was not shown
				// this refresh must be here, otherwise the GUI is not synchronized with
				// registration/deregistration made without using the GUI
				gui.removeAgentDesc(dfd.getName(),df.this.getAID());
				gui.showStatusMsg("Deregistration of agent: " + dfd.getName().getName() +" done.");
			}catch(Exception e1){}
	}
	
	
	void DFModify(DFAgentDescription dfd) throws NotRegistered {
		//checkMandatorySlots(FIPAAgentManagementOntology.MODIFY, dfd);
		Object old = agentDescriptions.register(dfd.getName(), dfd);
		if(old == null) {
			// Rollback
			agentDescriptions.deregister(dfd.getName());
			throw new NotRegistered();
		}
		// for subscription
		subManager.handleChange(dfd, (DFAgentDescription)old);
		try{
			gui.removeAgentDesc(dfd.getName(), df.this.getAID());
			gui.addAgentDesc(dfd.getName());
			gui.showStatusMsg("Modify of agent: "+dfd.getName().getName() + " done.");
		}catch(Exception e){}
		
	}
	
	List DFSearch(DFAgentDescription dfd, int maxResults) {
		return agentDescriptions.search(dfd, maxResults);
	}
	
	KBIterator DFIteratedSearch(DFAgentDescription dfd) {
		return agentDescriptions.iterator(dfd);
	}
	
	
	////////////////////////////////////////////////////////////////
	// Methods serving the actions of the FIPA Management ontology
	////////////////////////////////////////////////////////////////
	
	/**
	 Serve the Register action of the FIPA management ontology.
	 Package scoped since it is called by DFFipaAgentManagementBehaviour.
	 */
	void registerAction(Register r, AID requester) throws FIPAException {
		DFAgentDescription dfd = (DFAgentDescription) r.getDescription();
		
		// Check mandatory slots
		DFService.checkIsValid(dfd, true);
		if(logger.isLoggable(Logger.CONFIG))
			logger.log(Logger.CONFIG,"Agent "+requester.getName()+" requesting action Register for "+dfd.getName());
		
		// Avoid autoregistration
		if (dfd.getName().equals(getAID())) {
			throw new Unauthorised();
		}
		
		// Do it
		DFRegister(dfd);
		
	}
	
	/**
	 Serve the Deregister action of the FIPA management ontology.
	 Package scoped since it is called by DFFipaAgentManagementBehaviour.
	 */
	void deregisterAction(Deregister d, AID requester) throws FIPAException {
		DFAgentDescription dfd = (DFAgentDescription) d.getDescription();
		
		// Check mandatory slots
		DFService.checkIsValid(dfd, false);
		if(logger.isLoggable(Logger.CONFIG))
			logger.log(Logger.CONFIG,"Agent "+requester.getName()+" requesting action Deregister for "+dfd.getName());