subscriptionresponder.java

java实现的P2P多agent中间件

		if (s != null) {
			if (mySubscriptionManager != null) {
				mySubscriptionManager.deregister(s);
			}
			s.close();
		}
		return null;
	}
	
	/**
	 This method allows to register a user defined <code>Behaviour</code>
	 in the HANDLE_SUBSCRIPTION state.
	 This behaviour overrides the homonymous method.
	 This method also sets 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
	 response (if any) to be sent back into the datastore at the 
	 <code>RESPONSE_KEY</code> key.
	 The incoming subscription message can be retrieved from the 
	 datastore at the <code>SUBSCRIPTION_KEY</code> key
	 @param b the Behaviour that will handle this state
	 */
	public void registerHandleSubscription(Behaviour b) {
		registerState(b, HANDLE_SUBSCRIPTION);
		b.setDataStore(getDataStore());
	}
	
	/**
	 * @deprecated Use registerHandleSubscription() instead.
	 */
	public void registerPrepareResponse(Behaviour b) {
		registerHandleSubscription(b);
	}
	
	/**
	 This method allows to register a user defined <code>Behaviour</code>
	 in the HANDLE_CANCEL state.
	 This behaviour overrides the homonymous method.
	 This method also sets 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
	 response (if any) to be sent back into the datastore at the 
	 <code>RESPONSE_KEY</code> key.
	 The incoming CANCEL message can be retrieved from the 
	 datastore at the <code>CANCEL_KEY</code> key
	 @param b the Behaviour that will handle this state
	 */
	public void registerHandleCancel(Behaviour b) {
		registerState(b, HANDLE_CANCEL);
		b.setDataStore(getDataStore());
	}
	
	/**
	 Utility method to correctly create a new <code>Subscription</code> object 
	 managed by this <code>SubscriptionResponder</code>
	 */
	public Subscription createSubscription(ACLMessage subsMsg) {
		Subscription s = new Subscription(this, subsMsg);
		String convId = subsMsg.getConversationId();
		if (convId != null) {
			subscriptions.put(convId, s);
		}
		return s;
	}
	
	/**
	 Utility method to correctly retrieve the 
	 <code>Subscription</code> object that is related to the conversation
	 message <code>msg</code> belongs to.
	 @param msg The message whose <code>conversation-id</code> indicates the conversation
	 @return the <code>Subscription</code> object related to the conversation the given message belongs to
	 */
	public Subscription getSubscription(ACLMessage msg) {
		String convId = msg.getConversationId();
		return getSubscription(convId);
	}
	
	/**
	 Utility method to correctly retrieve the 
	 <code>Subscription</code> object that is related a given conversation.
	 @param convId The id of the conversation
	 @return the <code>Subscription</code> object related to the given conversation
	 */
	public Subscription getSubscription(String convId) {
		Subscription s = null;
		if (convId != null) {
			s = (Subscription) subscriptions.get(convId);
		}
		return s;
	}
	
	/**
	 * Utility method that retrieves all Subscription-s done by a given agent
	 * @param subscriber The AID of the agent whose subscriptions must be retrieved
	 * @return A <code>Vector</code> including all <code>Subscription</code>-s made by the given agent
	 */
	public Vector getSubscriptions(AID subscriber) {
		// Synchronization is needed to avoid concurrent modification exception in case this method is 
		// invoked from within a separate Thread
		synchronized (subscriptions) {
			Vector ss = new Vector();
			Enumeration en = subscriptions.elements();
			while (en.hasMoreElements()) {
				Subscription s = (Subscription) en.nextElement();
				if (s.getMessage().getSender().equals(subscriber)) {
					ss.addElement(s);
				}
			}
			return ss;
		}
	}
	
	/**
	 * Utility method that retrieves all Subscription-s managed by this <code>SubscriptionResponder</code>
	 * @return A <code>Vector</code> including all <code>Subscription</code>-s managed by this <code>SubscriptionResponder</code>
	 */
	public Vector getSubscriptions() {
		// Synchronization is needed to avoid concurrent modification exception in case this method is 
		// invoked from within a separate Thread
		synchronized (subscriptions) {
			Vector ss = new Vector();
			Enumeration en = subscriptions.elements();
			while (en.hasMoreElements()) {
				Subscription s = (Subscription) en.nextElement();
				ss.addElement(s);
			}
			return ss;
		}
	}
	
	/**
	 This is called by a Subscription object when a notification has
	 to be sent to the corresponding subscribed agent.
	 Executed in mutual exclusion with sendNotifications(). Note that this
	 synchronization is not needed in general, but we never know how users
	 manages Subscription objects (possibly in another thread)
	 */
	private synchronized void addNotification(ACLMessage notification, ACLMessage subscription) {
		ACLMessage[] tmp = new ACLMessage[] {notification, subscription};
		notifications.add(tmp);
		msgRecBehaviour.interrupt();
	}
	
	/**
	 This is called within the SEND_NOTIFICATIONS state.
	 Executed in mutual exclusion with addNotification(). Note that this
	 synchronization is not needed in general, but we never know how users
	 manages Subscription objects (possibly in another thread)
	 */
	private synchronized void sendNotifications() {
		Iterator it = notifications.iterator();
		while (it.hasNext()) {
			boolean receiversNull = true;
			boolean replyWithNull = true;
			ACLMessage[] tmp = (ACLMessage[]) it.next();
			if (tmp[0].getAllReceiver().hasNext()) {
				receiversNull = false;
			}
			if (tmp[0].getReplyWith() != null) {
				replyWithNull = false;
			}
			ReplySender.adjustReply(myAgent, tmp[0], tmp[1]);
			myAgent.send(tmp[0]);
			// If the message was modified --> restore it
			if (receiversNull) {
				tmp[0].clearAllReceiver();
			}
			if (replyWithNull) {
				tmp[0].setReplyWith(null);
			}
		}
		notifications.clear();
	}
	
	/**
	 Inner interface SubscriptionManager.
	 <p>
	 A <code>SubscriptionResponder</code>, besides enforcing and
	 controlling the sequence of messages in a subscription conversation, also stores current subscriptions
	 into an internal table. In many cases however it is desirable to manage Subscription objects in an application specific way 
	 (e.g. storing them to a persistent support such as a DB). To enable that, it is possible to pass a 
	 SubscriptionManager implementation to the SubscriptionResponder. The SubscriptionManager is notified 
	 about subscription and cancellation events by means of the register() and deregister() methods.     
	 <p>
	 */
	public static interface SubscriptionManager {
		/**
		 Register a new Subscription object
		 @param s The Subscription object to be registered
		 @return The boolean value returned by this method provides an 
		 indication to the <code>SubscriptionResponder</code> about whether
		 or not an AGREE message should be sent back to the initiator. The
		 default implementation of the <code>handleSubscription()</code> method
		 of the <code>SubscriptionResponder</code> ignores this indication,
		 but programmers can override it.
		 */
		boolean register(Subscription s) throws RefuseException, NotUnderstoodException;
		/**
		 Deregister a Subscription object
		 @return The boolean value returned by this method provides an 
		 indication to the <code>SubscriptionResponder</code> about whether
		 or not an INFORM message should be sent back to the initiator. The
		 default implementation of the <code>handleCancel()</code> method
		 of the <code>SubscriptionResponder</code> ignores this indication,
		 but programmers can override it.
		 */
		boolean deregister(Subscription s) throws FailureException;
	} // END of inner interface SubscriptionManager

	
	/**
	 Inner calss Subscription
	 <p>
	 This class represents a subscription. When a notification has to 
	 be sent to a subscribed agent the notification message should not 
	 be directly sent to the subscribed agent, but should be passed to the
	 <code>Subscription</code> object representing the subscription of that 
	 agent by means of its <code>notify()</code> method. This automatically 
	 handles sequencing and protocol fields appropriately.
	 <code>Subscription</code> objects must be created by means of the
	 <code>createSubscription()</code> method.
	 */
	public static class Subscription {
		
		private ACLMessage subscription;
		private SubscriptionResponder myResponder;
		
		/**
		 Private constructor. The <code>createSubscription()</code>
		 must be used instead.
		 @param r The <code>SubscriptionResponder</code> that received
		 the subscription message corresponding to this 
		 <code>Subscription</code>
		 @param s The subscription message corresponding to this 
		 <code>Subscription</code>
		 */
		private Subscription(SubscriptionResponder r, ACLMessage s){
			myResponder = r;
			subscription = s;
		}
		
		/**
		 Retrieve the ACL message with which this
		 subscription object was created.
		 @return the subscription message corresponding to this 
		 <code>Subscription</code>
		 */
		public ACLMessage getMessage() {
			return subscription;
		}
		
		/** 
		 This method allows sending back a notification message to the subscribed 
		 agent associated to this <code>Subscription</code> object. The user 
		 should call this method, instead of directly using the <code>send()</code>
		 method of the <code>Agent</code> class, as it automatically 
		 handles sequencing and protocol fields appropriately.
		 */			   
		public void notify(ACLMessage notification){
			myResponder.addNotification(notification, subscription);
		}
		
		/** 
		 This method removes the current Subscription object from the SubscriptionResponder internal tables.
		 */			   
		public void close(){
			String convId = subscription.getConversationId();
			if (convId != null) {
				myResponder.subscriptions.remove(convId);
			}
		}
		
		// FIXME: Add equals() and hashCode() methods based on the SUBSCRIBE message conversation-id 
	} // END of inner class Subscription
	
}