notificationservice.java

java实现的P2P多agent中间件

				}
			}
			else {
				// Do nothing for now, but could also route the command to the main slice, thus enabling e.g. AMS replication
			}
		}
		
		private void handleNotifyPosted(VerticalCommand cmd) {
			Object[] params = cmd.getParams();
			ACLMessage msg = (ACLMessage)params[0];
			AID receiver = (AID)params[1];
			
			firePostedMessage(msg, receiver);
		}
		
		private void handleNotifyReceived(VerticalCommand cmd) {
			Object[] params = cmd.getParams();
			ACLMessage msg = (ACLMessage)params[0];
			AID receiver = (AID)params[1];
			
			fireReceivedMessage(msg, receiver);
		}
		
		private void handleNotifyChangedAgentPrincipal(VerticalCommand cmd) {
			Object[] params = cmd.getParams();
			AID id = (AID)params[0];
			JADEPrincipal from = (JADEPrincipal)params[1];
			JADEPrincipal to = (JADEPrincipal)params[2];
			
			fireChangedAgentPrincipal(id, from, to);
		}
		
		private void handleNotifyAddedBehaviour(VerticalCommand cmd) {
			Object[] params = cmd.getParams();
			AID id = (AID)params[0];
			Behaviour b = (Behaviour)params[1];
			
			fireAddedBehaviour(id, b);
		}
		
		private void handleNotifyRemovedBehaviour(VerticalCommand cmd) {
			Object[] params = cmd.getParams();
			AID id = (AID)params[0];
			Behaviour b = (Behaviour)params[1];
			
			fireRemovedBehaviour(id, b);
		}
		
		private void handleNotifyChangedBehaviourState(VerticalCommand cmd) {
			Object[] params = cmd.getParams();
			AID id = (AID)params[0];
			Behaviour b = (Behaviour)params[1];
			String from = (String)params[2];
			String to = (String)params[3];
			
			fireChangedBehaviourState(id, b, from, to);
		}
	} // End of inner class NotificationSourceSink
	
	
	/**
	 * Inner class NotificationOutgoingFilter
	 */
	private class NotificationOutgoingFilter extends Filter {
		public boolean accept(VerticalCommand cmd) {		
			try {
				String name = cmd.getName();
				if(name.equals(jade.core.messaging.MessagingSlice.SEND_MESSAGE)) {
					handleSendMessage(cmd);
				}
				else if(name.equals(jade.core.management.AgentManagementSlice.INFORM_CREATED)) {
					handleInformCreated(cmd);
				}
				else if(name.equals(jade.core.management.AgentManagementSlice.INFORM_KILLED)) {
					handleInformKilled(cmd);
				}
				else if(name.equals(jade.core.management.AgentManagementSlice.INFORM_STATE_CHANGED)) {
					handleNotifyChangedAgentState(cmd);
				}
			}
			catch(Throwable t) {
				cmd.setReturnValue(t);
			}
			
			// Never veto a command
			return true;
		}
		
		private void handleSendMessage(VerticalCommand cmd) {
			Object[] params = cmd.getParams();
			AID sender = (AID)params[0];
			ACLMessage msg = ((GenericMessage)params[1]).getACLMessage();
			AID receiver = (AID)params[2];
			
			fireSentMessage(msg, sender, receiver);
		}
		
		private void handleInformCreated(VerticalCommand cmd) {
			Object[] params = cmd.getParams();
			AID agent = (AID)params[0];
			
			fireBornAgent(agent);
		}
		
		private void handleInformKilled(VerticalCommand cmd) {
			Object[] params = cmd.getParams();
			AID agent = (AID)params[0];
			
			fireDeadAgent(agent);
		}
		
		private void handleNotifyChangedAgentState(VerticalCommand cmd) {
			Object[] params = cmd.getParams();
			AID id = (AID)params[0];
			AgentState from = (AgentState)params[1];
			AgentState to = (AgentState)params[2];
			
			fireChangedAgentState(id, from, to);
		}
	} // END of inner class NotificationOutgoingFilter
	
	
	/**
	 * Inner class NotificationIncomingFilter
	 */
	private class NotificationIncomingFilter extends Filter {
		// Notify listeners about the REATTACHED event only when the reattachment procedure 
		// has been completed
		public void postProcess(VerticalCommand cmd) {		
			try {
				String name = cmd.getName();
				if(name.equals(jade.core.Service.REATTACHED)) {
					handleReattached(cmd);
				}
			}
			catch(Throwable t) {
				cmd.setReturnValue(t);
			}
		}
		
		private void handleReattached(VerticalCommand cmd) {
			fireReattached();
		}
	} // END of inner class NotificationIncomingFilter
	
	
	/**
	 * Inner class ServiceComponent
	 */
	private class ServiceComponent implements Service.Slice {
		public Service getService() {
			return NotificationService.this;
		}
		
		public Node getNode() throws ServiceException {
			try {
				return NotificationService.this.getLocalNode();
			}
			catch(IMTPException imtpe) {
				throw new ServiceException("Problem in contacting the IMTP Manager", imtpe);
			}
		}
		
		public VerticalCommand serve(HorizontalCommand cmd) {
			try {
				String cmdName = cmd.getName();
				Object[] params = cmd.getParams();
				
				if(cmdName.equals(NotificationSlice.H_SNIFFON)) {
					AID snifferName = (AID)params[0];
					AID targetName = (AID)params[1];
					
					sniffOn(snifferName, targetName);
				}
				else if(cmdName.equals(NotificationSlice.H_SNIFFOFF)) {
					AID snifferName = (AID)params[0];
					AID targetName = (AID)params[1];
					
					sniffOff(snifferName, targetName);
				}
				else if(cmdName.equals(NotificationSlice.H_DEBUGON)) {
					AID introspectorName = (AID)params[0];
					AID targetName = (AID)params[1];
					
					debugOn(introspectorName, targetName);
				}
				else if(cmdName.equals(NotificationSlice.H_DEBUGOFF)) {
					AID introspectorName = (AID)params[0];
					AID targetName = (AID)params[1];
					
					debugOff(introspectorName, targetName);
				}
			}
			catch(Throwable t) {
				cmd.setReturnValue(t);
			}
			
			return null;
		}
		
		private void sniffOn(AID snifferName, AID targetName) throws IMTPException {
			ToolNotifier tn = findNotifier(snifferName);
			if(tn == null) { // Need a new notifier 
				tn = new ToolNotifier(snifferName);
				AID id = new AID(snifferName.getLocalName() + "-on-" + myID().getName(), AID.ISLOCALNAME);
				try {
					myContainer.initAgent(id, tn, null, null); // FIXME: Modify to use a proper owner Principal
					myContainer.powerUpLocalAgent(id);
					helper.registerMessageListener(tn);
				}
				catch (Exception e) {
					e.printStackTrace();
				}
			}
			tn.addObservedAgent(targetName);
		}
		
		private void sniffOff(AID snifferName, AID targetName) throws IMTPException {
			ToolNotifier tn = findNotifier(snifferName);
			if(tn != null) { 
				tn.removeObservedAgent(targetName);
			}
		}
		
		private void debugOn(AID introspectorName, AID targetName) throws IMTPException {
			// AMS debug enabling must be done by a separated Thread to avoid
			// deadlock with ToolNotifier startup
			if (targetName.equals(myContainer.getAMS()) && !(Thread.currentThread().getName().equals(AMS_DEBUG_HELPER))) {
				final AID in = introspectorName;
				final AID tg = targetName;
				Thread helper = new Thread(new Runnable() {
					public void run() {
						try {
							debugOn(in, tg);
						}
						catch(IMTPException imtpe) {
							imtpe.printStackTrace();
						}
					}
				});
				helper.setName(AMS_DEBUG_HELPER);
				helper.start();
				return;
			}
			
			// Get the ToolNotifier for the indicated debugger (or create a new one
			// if not yet there)
			ToolNotifier tn = findNotifier(introspectorName);
			if(tn == null) { // Need a new notifier
				tn = new ToolNotifier(introspectorName);
				AID id = new AID(introspectorName.getLocalName() + "-on-" + myID().getName(), AID.ISLOCALNAME);
				try {
					myContainer.initAgent(id, tn, null, null); // FIXME: Modify to use a proper owner Principal
					myContainer.powerUpLocalAgent(id);
					if (targetName.equals(myContainer.getAMS())) {
						// If we are debugging the AMS, let's wait for the ToolNotifier 
						// to be ready to avoid deadlock problems. Note also that in 
						// this case this code is executed by the ams-debug-helper thread and not
						// by the AMS thread
						tn.waitUntilStarted();
					}
					// Wait a bit to let the ToolNotifier pass in ACTIVE_STATE
					try {Thread.sleep(1000);} catch (Exception e) {};
					helper.registerMessageListener(tn);
					helper.registerAgentListener(tn);
				}
				catch (Exception e) {
					e.printStackTrace();
				}
			}
			tn.addObservedAgent(targetName);
			
			// Update the map of debuggers currently debugging the targetName agent
			synchronized (debuggers) {
				List l = (List) debuggers.get(targetName);
				if (l == null) {
					l = new LinkedList();
					debuggers.put(targetName, l);
				}
				if (!l.contains(introspectorName)) {
					l.add(introspectorName);
				}
			}
			
			Agent a = myContainer.acquireLocalAgent(targetName);
			
			// Activate generation of behaviour-related events on the
			// target agent
			a.setGenerateBehaviourEvents(true);
			
			// Retrieve the current agent state
			AgentState as = a.getAgentState();
			
			// Retrieve the list of pending ACL messages
			List messages = new LinkedList();
			myContainer.fillListFromMessageQueue(messages, a);
			
			// Retrieve the list of ready and blocked agent behaviour IDs
			List readyBehaviours = new LinkedList();
			myContainer.fillListFromReadyBehaviours(readyBehaviours, a);
			List blockedBehaviours = new LinkedList();
			myContainer.fillListFromBlockedBehaviours(blockedBehaviours, a);
			
			myContainer.releaseLocalAgent(targetName);
			
			// Notify all the needed events	
			fireChangedAgentState(targetName, as, as);
			
			Iterator itReady = readyBehaviours.iterator();
			while(itReady.hasNext()) {
				BehaviourID bid = (BehaviourID)itReady.next();
				AgentEvent ev = new AgentEvent(AgentEvent.ADDED_BEHAVIOUR, targetName, bid, myContainer.getID());
				tn.addedBehaviour(ev);
			}
			
			Iterator itBlocked = blockedBehaviours.iterator();
			while(itBlocked.hasNext()) {
				BehaviourID bid = (BehaviourID)itBlocked.next();
				AgentEvent ev = new AgentEvent(AgentEvent.ADDED_BEHAVIOUR, targetName, bid, myContainer.getID());
				tn.addedBehaviour(ev);
				ev = new AgentEvent(AgentEvent.CHANGED_BEHAVIOUR_STATE, targetName, bid, Behaviour.STATE_READY, Behaviour.STATE_BLOCKED, myContainer.getID());