clientsubscriptions.java

java开源的企业总线.xmlBlaster

/*------------------------------------------------------------------------------
Name:      ClientSubscriptions.java
Project:   xmlBlaster.org
Copyright: xmlBlaster.org, see xmlBlaster-LICENSE file
------------------------------------------------------------------------------*/
package org.xmlBlaster.engine;

import org.xmlBlaster.engine.ServerScope;
import java.util.logging.Logger;
import java.util.logging.Level;
import org.xmlBlaster.util.key.KeyData;
import org.xmlBlaster.util.key.QueryKeyData;
import org.xmlBlaster.util.XmlBlasterException;
import org.xmlBlaster.authentication.Authenticate;
import org.xmlBlaster.authentication.I_ClientListener;
import org.xmlBlaster.authentication.ClientEvent;
import org.xmlBlaster.authentication.SessionInfo;
import org.xmlBlaster.util.def.Constants;

import java.util.*;


/**
 * Handling subscriptions, collected for each session of each client. 
 * <p />
 * There exists exactly one instance of this class for each xmlBlaster server,
 * the instance is handled by RequestBroker.
 * <p />
 * The interface SubscriptionListener informs about subscribe/unsubscribe events
 * @author Marcel Ruff
 */
public class ClientSubscriptions implements I_ClientListener, I_SubscriptionListener
{
   private static Logger log = Logger.getLogger(ClientSubscriptions.class.getName());

   /**
    * All exact subscriptions of a Client are in this map.
    * <p>
    * These are the subscriptions which are referenced from a TopicHandler<br>
    * including those with a query (XPATH).
    * <p>
    * A multimap would be appropriate, but since this is not supported
    * by the Collections API, a map with a set as value is used.
    * <br>
    * Used for performing logout.
    * <p>
    * key   = client.getSessionName().getRelativeName()
    * value = subMap (Collections.synchronizedMap(new HashMap());)
    *         with SubscriptionInfo objects
    */
   final private Map clientSubscriptionMap = Collections.synchronizedMap(new HashMap());


   /**
    * All generic subscriptions are collected here.
    * Generic are all subscriptions who don't subscribe a precise key-oid,
    * but rather subscribe all MsgUnit matching a XPath query match.
    * <br>
    * If new MsgUnit are published, this set is consulted to check
    * if some older subscriptions would match as well
    * <p>
    * value = SubscriptionInfo objects with generic subscriptions, but not
    *         those, who subscribed a MsgUnit exactly by a known oid
    */
   final private Set querySubscribeRequestsSet = Collections.synchronizedSet(new HashSet());


   /**
    * Exactly one instance for each xmlBlaster server.
    * <p />
    * (no singleton pattern to allow multiple servers)
    * @param requestBroker my master (singleton)
    * @param authenticate another master
    */
   ClientSubscriptions(ServerScope glob, RequestBroker requestBroker, Authenticate authenticate) throws XmlBlasterException
   {
      requestBroker.addSubscriptionListener(this);
      authenticate.addClientListener(this);
   }


   /**
    * All known subscriptions which match a query,
    * but not those subscriptions which address exactly via key-oid
    */
   public Set getQuerySubscribeRequestsSet()
   {
      return querySubscribeRequestsSet;
   }

   /**
    * If you have the unique id of a subscription, you may access the
    * SubscriptionInfo object here.
    * <p />
    * You can access XPATH or EXACT subscription objects through this method
    * @param sessionInfo All infos about the client
    * @param subscriptionInfoUniqueKey
    * @return corresponding subscriptionInfo object<br />
    *         or null if not found
    */
   public SubscriptionInfo getSubscription(SessionInfo sessionInfo, String subscriptionInfoUniqueKey) throws XmlBlasterException
   {
      Object obj;
      Map subMap;
      synchronized(clientSubscriptionMap) {
         obj = clientSubscriptionMap.get(sessionInfo.getSessionName().getRelativeName());
         if (obj == null)
            return null;
         subMap = (Map)obj;
      }

      SubscriptionInfo subs = (SubscriptionInfo)subMap.get(subscriptionInfoUniqueKey);
      if (log.isLoggable(Level.FINE)) log.fine("Looking for subscriptionId=" + subscriptionInfoUniqueKey + " found " + subs);
      return subs;
   }

   /**
    * If you have the unique id of a subscription, you may access the
    * SubscriptionInfo object here.
    * <p />
    * You can access XPATH or EXACT subscription objects through this method
    * @param subscriptionInfoUniqueKey
    * @return corresponding subscriptionInfo object<br />
    *         or null if not found
    */
   public SubscriptionInfo getSubscription(String subscriptionInfoUniqueKey) {
      synchronized(this.clientSubscriptionMap) {
         Iterator iterator = this.clientSubscriptionMap.values().iterator();
         while (iterator.hasNext()) {
            Map subMap = (Map)iterator.next();
            synchronized(subMap) {
               Object obj = subMap.get(subscriptionInfoUniqueKey); 
               if (obj != null) {
                  return (SubscriptionInfo)obj;
               }
            }
         }
      }

      if (log.isLoggable(Level.FINE)) log.fine("subscriptionId=" + subscriptionInfoUniqueKey + " not found");
      return null;
   }

   /**
    * @return The number of all subscriptions in this cluster node.
    */
   public int getNumSubscriptions() {
      int num = 0;
      synchronized(this.clientSubscriptionMap) {
         Iterator iterator = this.clientSubscriptionMap.values().iterator();
         while (iterator.hasNext()) {
            Map subMap = (Map)iterator.next();
            synchronized(subMap) {
               num += subMap.size();
            }
         }
      }
      return num;
   }

   /**
    * @return All subscriptionId in a comma separated string
    */
   public String getSubscriptionList() {
      StringBuffer sb = new StringBuffer(getNumSubscriptions()*20);
      synchronized(this.clientSubscriptionMap) {
         Iterator iterator = this.clientSubscriptionMap.values().iterator();
         while (iterator.hasNext()) {
            Map subMap = (Map)iterator.next();
            synchronized(subMap) {
               Iterator iterator2 = subMap.values().iterator();
               while (iterator2.hasNext()) {
                  SubscriptionInfo sub = (SubscriptionInfo)iterator2.next();
                  if (sb.length() > 0)
                     sb.append(",");
                  sb.append(sub.getSubscriptionId());
               }
            }
         }
      }
      return sb.toString();
   }
   public String[] getSubscriptions() {
      ArrayList list = new ArrayList(10+this.clientSubscriptionMap.size());
      synchronized(this.clientSubscriptionMap) {
         Iterator iterator = this.clientSubscriptionMap.values().iterator();
         while (iterator.hasNext()) {
            Map subMap = (Map)iterator.next();
            synchronized(subMap) {
               Iterator iterator2 = subMap.values().iterator();
               while (iterator2.hasNext()) {
                  SubscriptionInfo sub = (SubscriptionInfo)iterator2.next();
                  list.add(sub.getSubscriptionId());
               }
            }
         }
      }
      return (String[])list.toArray(new String[list.size()]);
   }

   /**
    * Access all subscriptions of a client
    * @return never null
    */
   public SubscriptionInfo[] getSubscriptions(SessionInfo sessionInfo) {
      synchronized(clientSubscriptionMap) {
         Object obj = clientSubscriptionMap.get(sessionInfo.getSessionName().getRelativeName());
         if (obj == null)
            return new SubscriptionInfo[0];
         java.util.Collection c = ((Map)obj).values();
         return (SubscriptionInfo[])c.toArray(new SubscriptionInfo[c.size()]);
      }
   }


   /*
    * If you have the key oid, the xpath query or domain of a message and a session it belongs to,
    * you may access the SubscriptionInfo object here.
    * <p />
    * Note that TopicHandler.findSubscriber() will not return a SubscriptionInfo
    * if never a message arrived for such a subscription, so prefer this method.
    *
    * @param sessionInfo All infos about the client
    * @param queryKey The query to look for
    * @return Vector containing corresponding subscriptionInfo objects<br />
    *         is > 1 if this session has subscribed multiple times on the
    *         same message, or null if this session has not subscribed it
    */
   public Vector getSubscription(SessionInfo sessionInfo, QueryKeyData queryKey) throws XmlBlasterException {
      if (queryKey == null || sessionInfo==null) return null;
      Object obj;
      Map subMap;
      synchronized(clientSubscriptionMap) {
         obj = clientSubscriptionMap.get(sessionInfo.getSessionName().getRelativeName());
         if (obj == null) {
            if (log.isLoggable(Level.FINE)) log.fine("Session '" + sessionInfo.getId() + "' is unknown, there was no subscription of this session yet.");
            return null;
         }
         subMap = (Map)obj;
      }

      // Slow linear search of all subscribes of a session
      // Don't use for performance critical tasks
      Vector vec = null;
      synchronized (subMap) {
         Iterator iterator = subMap.values().iterator();
         while (iterator.hasNext()) {
            SubscriptionInfo sub = (SubscriptionInfo)iterator.next();
            if (queryKey.equals(sub.getKeyData())) {
               if (vec == null) vec = new Vector();
               vec.addElement(sub);
            }
         }
      }
      return vec;
   }


   /*
    * If you have the key oid of a message and a session it belongs to, you may access the
    * SubscriptionInfo object here.
    * <p />
    * Note that TopicHandler.findSubscriber() will not return a SubscriptionInfo
    * if never a message arrived for such a subscription, so prefer this method.
    *
    * @param sessionInfo All infos about the client
    * @param keyOid The unique message oid
    * @param exact true Access only EXACT subscription objects through this method.<br />
    *              false All subscriptions
    * @return Vector containing corresponding subscriptionInfo objects<br />
    *         is > 1 if this session has subscribed multiple times on the
    *         same message, or null if this session has not subscribed it
    */
   public Vector getSubscriptionByOid(SessionInfo sessionInfo, String keyOid, boolean exactOnly) throws XmlBlasterException {
      if (keyOid == null || sessionInfo==null) return null;
      Object obj;
      Map subMap;
      synchronized(clientSubscriptionMap) {
         obj = clientSubscriptionMap.get(sessionInfo.getSessionName().getRelativeName());
         if (obj == null) {
            if (log.isLoggable(Level.FINE)) log.fine("Session '" + sessionInfo.getId() + "' is unknown, there was no subscription of this session yet. Oid '" + keyOid + "' is not existing");
            return null;
         }
         subMap = (Map)obj;
      }

      //if (log.isLoggable(Level.FINE)) log.trace(ME, "Found subscription map with " + subMap.size() + " entries for '" + sessionInfo.getId() + "'");

      // Slow linear search of all subscribes of a session
      // Don't use for performance critical tasks
      Vector vec = null;
      synchronized (subMap) {
         Iterator iterator = subMap.values().iterator();
         while (iterator.hasNext()) {
            SubscriptionInfo sub = (SubscriptionInfo)iterator.next();
            if (exactOnly && sub.getKeyData().isQuery()) {
               if (log.isLoggable(Level.FINE)) log.fine("Ignoring subscription " + sub.getSubscriptionId() +
                      " for session '" + sessionInfo.getId() + "' for message '" + keyOid + "'");
               continue;
            }
            if (keyOid.equals(sub.getKeyOid())) {
               if (log.isLoggable(Level.FINE)) log.fine("Found subscription " + sub.getSubscriptionId() +
                      " for session '" + sessionInfo.getId() + "' for message '" + keyOid + "'");
               if (vec == null) vec = new Vector();
               vec.addElement(sub);
            }
         }
      }
      return vec;
   }


   /*
    * If you have the key oid of a message, you can access all SubscriptionInfo objects
    * of all sessions here.
    * <p />
    * Note that TopicHandler.findSubscriber() will not return a SubscriptionInfo
    * if never a message arrived for such a subscription, so prefer this method.
    *
    * @param keyOid The unique message oid
    * @param exact true Access only EXACT subscription objects through this method.<br />
    *              false All subscriptions
    * @return Vector containing corresponding subscriptionInfo objects or null
    */
   public Vector getSubscriptionByOid(String keyOid, boolean exactOnly) throws XmlBlasterException {
      Vector vec = null;
      // Slow linear search of SubscriptionInfos
      // Don't use for performance critical tasks
      synchronized(clientSubscriptionMap) {
         Iterator iterator = clientSubscriptionMap.values().iterator();