sessioninfo.java

java开源的企业总线.xmlBlaster

      return getDispatchStatistic().getNumUpdate();
   }

   public final long getCbQueueNumMsgs() {
      if (this.sessionQueue == null) return 0L;
      return this.sessionQueue.getNumOfEntries();
   }

   public final long getCbQueueBytes() {
      if (this.sessionQueue == null) return 0L;
      return this.sessionQueue.getNumOfBytes();
   }

   public final long getCbQueueBytesCache() {
      I_Queue sq = this.sessionQueue;
      if (sq == null) return 0L;
      if (sq instanceof CacheQueueInterceptorPlugin) {
         CacheQueueInterceptorPlugin cq = (CacheQueueInterceptorPlugin)sq;
         I_Queue tq = cq.getTransientQueue();
         if (tq != null) return tq.getNumOfBytes();
         return 0L;
      }
      return -1L;
   }

   public final long getCbQueueNumMsgsCache() {
      I_Queue sq = this.sessionQueue;
      if (sq == null) return 0L;
      if (sq instanceof CacheQueueInterceptorPlugin) {
         CacheQueueInterceptorPlugin cq = (CacheQueueInterceptorPlugin)sq;
         I_Queue tq = cq.getTransientQueue();
         if (tq != null) return tq.getNumOfEntries();
         return 0L;
      }
      return -1L;
   }

   public final long getCbQueueMaxMsgs() {
      if (this.sessionQueue == null) return 0L;
      return this.sessionQueue.getMaxNumOfEntries();
   }

   public final long getCbQueueMaxMsgsCache() {
      I_Queue sq = this.sessionQueue;
      if (sq == null) return 0L;
      if (sq instanceof CacheQueueInterceptorPlugin) {
          CacheQueueInterceptorPlugin cq = (CacheQueueInterceptorPlugin)sq;
          I_Queue tq = cq.getTransientQueue();
          if (tq != null) return tq.getMaxNumOfEntries();
          return 0L;
      }
      return -1L;
   }

   public String pingClientCallbackServer() {
      DispatchManager dispatchManager = this.dispatchManager;
      if (dispatchManager != null) {
         dispatchManager.pingCallbackServer(true);
         return "Ping done in " + getPingRoundTripDelay() + " millis, current state is " + dispatchManager.getDispatchConnectionsHandler().getState().toString();
      }
      return "No ping because of no callback";
   }

   public long getPingRoundTripDelay() {
      return getDispatchStatistic().getPingRoundTripDelay();
   }

   public long getRoundTripDelay() {
      return getDispatchStatistic().getRoundTripDelay();
   }

   public final String[] getSubscribedTopics() {
      SubscriptionInfo[] subs = glob.getRequestBroker().getClientSubscriptions().getSubscriptions(this);
      String[] arr = new String[subs.length];
      for (int i=0; i<arr.length; i++) {
         arr[i] = subs[i].getKeyOid();
      }
      return arr;
   }

   public final String subscribe(String url, String qos) throws XmlBlasterException {
      if (url == null) {
         return "Please pass a valid topic oid";
      }

      log.info(ME+": Administrative subscribe() of '" + url + "' for client '" + getId() + "' qos='" + qos + "'");
      SubscribeKey uk = new SubscribeKey(glob, url);
      SubscribeQos uq;
      if (qos == null || qos.length() == 0 || qos.equalsIgnoreCase("String")) {
         uq = new SubscribeQos(glob);
      }
      else {
         uq = new SubscribeQos(glob, glob.getQueryQosFactory().readObject(qos));
      }
      SubscribeQosServer uqs = new SubscribeQosServer(glob, uq.getData());

      String ret = glob.getRequestBroker().subscribe(this, uk.getData(), uqs);

      SubscribeReturnQos tmp = new SubscribeReturnQos(glob, ret);
      ret = "Subscribe '" + tmp.getSubscriptionId() + "' state is " + tmp.getState();
      if (tmp.getStateInfo() != null)
         ret += " " + tmp.getStateInfo();

      if (ret.length() == 0) {
         ret = "Unsubscribe of '" + url + "' for client '" + getId() + "' did NOT match any subscription";
      }

      return ret;
   }

   public String[] unSubscribeByIndex(int index, String qos) throws XmlBlasterException {
      SubscriptionInfo[] subs = glob.getRequestBroker().getClientSubscriptions().getSubscriptions(this);
      if (subs.length < 1)
         return new String[] { "Currently no topics are subscribed" };

      if (index < 0 || index >= subs.length) {
         return new String[] { "Please choose an index between 0 and " + (subs.length-1) + " (inclusiv)" };
      }

      return unSubscribe(subs[index].getSubscriptionId(), qos);
   }

   public final String[] unSubscribe(String url, String qos) throws XmlBlasterException {
      if (url == null)
         return new String[] { "Please pass a valid topic oid" };

      log.info(ME+": Administrative unSubscribe() of '" + url + "' for client '" + getId() + "'");
      UnSubscribeKey uk = new UnSubscribeKey(glob, url);

      UnSubscribeQos uq;
      if (qos == null || qos.length() == 0 || qos.equalsIgnoreCase("String"))
         uq = new UnSubscribeQos(glob);
      else
         uq = new UnSubscribeQos(glob, glob.getQueryQosFactory().readObject(qos));
      UnSubscribeQosServer uqs = new UnSubscribeQosServer(glob, uq.getData());

      String[] ret = glob.getRequestBroker().unSubscribe(this, uk.getData(), uqs);

      if (ret.length == 0)
         return new String[] { "Unsubscribe of '" + url + "' for client '" + getId() + "' did NOT match any subscription" };

      for (int i=0; i<ret.length; i++) {
         UnSubscribeReturnQos tmp = new UnSubscribeReturnQos(glob, ret[i]);
         ret[i] = "Unsubscribe '" + tmp.getSubscriptionId() + "' state is " + tmp.getState();
         if (tmp.getStateInfo() != null)
            ret[i] += " " + tmp.getStateInfo();
      }

      return ret;
   }

   public final String[] getSubscriptions() throws XmlBlasterException {
      SubscriptionInfo[] subs = glob.getRequestBroker().getClientSubscriptions().getSubscriptions(this);
      String[] arr = new String[subs.length];
      for (int i=0; i<arr.length; i++) {
         arr[i] = subs[i].getSubscriptionId();
      }
      return arr;
   }

   public final String getSubscriptionDump() throws XmlBlasterException {
      SubscriptionInfo[] subs = glob.getRequestBroker().getClientSubscriptions().getSubscriptions(this);
      if (subs.length < 1)
         return "";
      StringBuffer sb = new StringBuffer(subs.length * 300);
      sb.append("<SessionInfo id='").append(getId()).append("'>");
      for (int i=0; i<subs.length; i++) {
         sb.append(subs[i].toXml(" "));
      }
      sb.append("</SessionInfo>");
      return sb.toString();
   }

   public final String killSession() throws XmlBlasterException {
      glob.getAuthenticate().disconnect(getAddressServer(), securityCtx.getSecretSessionId(), "<qos/>");
      return getId() + " killed";
   }

   /**
    * Gets the uniqueId for the persistence of this session.
    * @return the uniqueId used to identify this session as an  entry
    * in the queue where it is stored  (for persistent subscriptions).
    * If the session is not persistent it returns -1L.
    *
    */
   public final long getPersistenceUniqueId() {
      return this.connectQos.getPersistenceUniqueId();
   }

   /**
    * Sets the uniqueId used to retrieve this session from the persistence
    * @param persistenceId
    */
   public final void setPersistenceUniqueId(long persistenceId) {
      this.connectQos.setPersistenceUniqueId(persistenceId);
   }

   /**
    * Sets the DispachManager belonging to this session to active or inactive.
    * It is initially active. Setting it to false temporarly inhibits dispatch of
    * messages which are in the callback queue. Setting it to true starts the
    * dispatch again.
    * @param dispatchActive
    */
   public void setDispatcherActive(boolean dispatcherActive) {
      if (this.dispatchManager != null) {
         this.dispatchManager.setDispatcherActive(dispatcherActive);
      }
   }

   public boolean getDispatcherActive() {
      if (this.dispatchManager != null) {
         return this.dispatchManager.isDispatcherActive();
      }
      return false;
   }

   public String[] peekCallbackMessages(int numOfEntries) throws XmlBlasterException {
      return this.glob.peekMessages(this.sessionQueue, numOfEntries, "callback");
   }

   /**
    * Peek messages from callback queue and dump them to a file, they are not removed.
    * @param numOfEntries The number of messages to peek, taken from the front
    * @param path The path to dump the messages to, it is automatically created if missing.
    * @return The file names of the dumped messages
    */
   public String[] peekCallbackMessagesToFile(int numOfEntries, String path) throws Exception {
      try {
         return this.glob.peekQueueMessagesToFile(this.sessionQueue, numOfEntries, path, "callback");
      }
      catch (XmlBlasterException e) {
         throw new Exception(e.toString());
      }
   }

   public long clearCallbackQueue() {
      I_Queue sessionQueue = this.sessionQueue;
      return (sessionQueue==null) ? 0L : sessionQueue.clear();
   }

   public long removeFromCallbackQueue(long numOfEntries) throws XmlBlasterException {
      I_Queue sessionQueue = this.sessionQueue;
      return (sessionQueue==null) ? 0L : sessionQueue.remove(numOfEntries, -1);
   }

   public MsgUnit[] getCallbackQueueEntries(String query) throws XmlBlasterException {
      if (this.queueQueryPlugin == null) {
         synchronized (this) {
            if (this.queueQueryPlugin == null) {
               this.queueQueryPlugin = new QueueQueryPlugin(this.glob);
            }
         }
      }
      return this.queueQueryPlugin.query(this.sessionQueue, query);
   }

   /** JMX Enforced by ConnectQosDataMBean interface. */
   public final void setSessionTimeout(long timeout) {
      getConnectQos().setSessionTimeout(timeout);
      try {
         refreshSession();
      } catch (XmlBlasterException e) {
         e.printStackTrace();
      }
   }


   /** JMX */
   public java.lang.String usage() {
      return ServerScope.getJmxUsageLinkInfo(this.getClass().getName(), null);
   }
   /** JMX */
   public java.lang.String getUsageUrl() {
      return ServerScope.getJavadocUrl(this.getClass().getName(), null);
   }
   /* JMX dummy to have a copy/paste functionality in jconsole */
   public void setUsageUrl(java.lang.String url) {}

   /**
    * @return Returns the remoteProperties or null
    */
   public ClientPropertiesInfo getRemoteProperties() {
      return this.remoteProperties;
   }

   /**
    * @return never null
    */
   public ClientProperty[] getRemotePropertyArr() {
      ClientPropertiesInfo tmp = this.remoteProperties;
      if (tmp == null) return new ClientProperty[0];
      return tmp.getClientPropertyArr();
   }

   /**
    * Set porperties send by our client.
    * @param remoteProperties The remoteProperties to set, pass null to reset.
    * The key is of type String and the value of type ClientProperty
    */
   public synchronized void setRemoteProperties(Map map) {
      if (map == null)
         this.remoteProperties = null;
      else
         this.remoteProperties = new ClientPropertiesInfo(map);
   }

   /**
    * Update porperties send by our client.
    * @param remoteProperties The remoteProperties to set,
    * if a property exists its value is overwritten, passing null does nothing
    * The key is of type String and the value of type ClientProperty
    */
   public synchronized void mergeRemoteProperties(Map map) {
      if (map == null || map.size() == 0) return;
      if (this.remoteProperties == null) {
          this.remoteProperties = new ClientPropertiesInfo(new HashMap());
          /*// Changed 2007-06-29 marcel: we now take a clone
         this.remoteProperties = new ClientPropertiesInfo(map);
         // remove, is only a hint:
         this.remoteProperties.put(Constants.CLIENTPROPERTY_REMOTEPROPERTIES, (ClientProperty)null);
         return;
         */
      }
      Iterator it = map.keySet().iterator();
      while (it.hasNext()) {
         String key = (String)it.next();
         if (Constants.CLIENTPROPERTY_REMOTEPROPERTIES.equals(key))
                continue; // Remove, is only a flag
         Object value = map.get(key);
         this.remoteProperties.put(key, (ClientProperty)value);
      }
   }

   /**
    * Add a remote property.
    * Usually this is done by a publish of a client, but for
    * testing reasons we can to it here manually.
    * If the key exists, its value is overwritten
    * @param key The unique key (no multimap)
    * @param value The value, it is assumed to be of type "String"
    * @return The old ClientProperty if existed, else null
    * @see <a href="http://www.xmlBlaster.org/xmlBlaster/doc/requirements/admin.events.html">The admin.events requirement</a>
    */
   public synchronized ClientProperty addRemoteProperty(String key, String value) {
      if (this.remoteProperties == null)
         this.remoteProperties = new ClientPropertiesInfo(null);
      ClientProperty old = (ClientProperty)this.remoteProperties.getClientPropertyMap().get(key);
      this.remoteProperties.put(key, value);
      return old;
   }

   public boolean isStalled() {
      return getDispatchStatistic().isStalled();
   }

   /**
    * Can be called when client connection is lost (NOT the callback connection).
    * Currently only detected by the SOCKET protocol plugin.
    * Others can only detect lost clients with their callback protocol pings
    */
   public void lostClientConnection() {
      if (log.isLoggable(Level.FINE)) log.fine(ME+": Protocol layer is notifying me about a lost connection");
      DispatchManager dispatchManager = this.dispatchManager;
      if (dispatchManager != null)
         dispatchManager.lostClientConnection();
   }

   /**
    * If the connection failed the reason is stored here, like this
    * cleanup code knows what happened.
    * @return the transportConnectFail
    */
   public XmlBlasterException getTransportConnectFail() {
      return this.transportConnectFail;
   }

   /**
    * @param transportConnectFail the transportConnectFail to set
    */
   public void setTransportConnectFail(XmlBlasterException transportConnectFail) {
      this.transportConnectFail = transportConnectFail;
   }

   /**
    * Can be optionally used by the current authorization plugin.
    */
   public Object getAuthorizationCache() {
      return authorizationCache;
   }

   public void setAuthorizationCache(Object authorizationCache) {
      this.authorizationCache = authorizationCache;
   }
}