subjectinfo.java

java开源的企业总线.xmlBlaster

      }
      if (log.isLoggable(Level.FINE)) log.fine(ME+": Accessing " + this.callbackAddressCache.length + " callback addresses from " + getSessions().length + " sessions for '" + getLoginName() + "' queue");
      return this.callbackAddressCache;
   }

   /**
    * If you have a callback address and want to know to which session it belongs.
    * @param addr The address object
    * @return the sessionInfo or null
    */
   public final SessionInfo findSessionInfo(AddressBase addr) {
      SessionInfo[] sessions = getSessions();
      for (int i=0; i<sessions.length; i++) {
         SessionInfo ses = sessions[i];
         if (ses.hasAddress(addr))
            return ses;
      }
      return null;
   }

   /**
    * @exception Throws XmlBlasterException if max. sessions is exhausted
    */
   public final void checkNumberOfSessions(ConnectQosData qos) throws XmlBlasterException {
      if (SessionQos.DEFAULT_maxSessions != qos.getSessionQos().getMaxSessions())
         this.maxSessions = qos.getSessionQos().getMaxSessions();

      if (getSessions().length >= this.maxSessions) {
         log.warning(ME+": Max sessions = " + this.maxSessions + " for user " + getLoginName() + "@" + qos.getSecurityQos().getClientIp() + " exhausted, login denied.");
         throw new XmlBlasterException(glob, ErrorCode.USER_CONFIGURATION_MAXSESSION, ME, "Max sessions = " + this.maxSessions + " exhausted, login denied.");
      }
   }

   /**
    * Get notification that the client did a login.
    * <p />
    * This instance may exist before a login was done, for example
    * when some messages where directly addressed to this client.<br />
    * This notifies about a client login.
    */
   public final void notifyAboutLogin(SessionInfo sessionInfo) throws XmlBlasterException {
      if (!isAlive()) { // disconnect() and connect() are not synchronized, so this can happen
         throw new XmlBlasterException(glob, ErrorCode.INTERNAL_UNKNOWN, ME, "SubjectInfo is shutdown, try to login again");
      }

      if (log.isLoggable(Level.FINER)) log.finer(ME+": notifyAboutLogin(" + sessionInfo.getSecretSessionId() + ")");
      synchronized (this.sessionMap) {
         this.sessionMap.put(sessionInfo.getId(), sessionInfo);
         this.sessionArrCache = null;
         this.callbackAddressCache = null;
      }
      if (log.isLoggable(Level.FINEST)) log.finest(ME+": "+this.subjectQueue.toXml(""));

      if (this.subjectQueue.getNumOfEntries() > 0) {
         if (log.isLoggable(Level.FINE)) log.fine(ME+": Flushing " + this.subjectQueue.getNumOfEntries() + " messages");
         this.glob.getSubjectInfoShuffler().shuffle(this); // Background thread
      }
   }

   /**
    * Get notification that the client did a logout.
    * <br />
    * Note that the loginName is not reset.
    * @param absoluteSessionName == sessionInfo.getId()
    * @param clearQueue Shall the message queue of the client be cleared&destroyed as well (e.g. disconnectQos.deleteSubjectQueue())?
    * @param forceShutdownEvenIfEntriesExist on last session
    */
   public final void notifyAboutLogout(String absoluteSessionName, boolean clearQueue, boolean forceShutdownEvenIfEntriesExist) throws XmlBlasterException {
      if (!isAlive()) { // disconnect() and connect() are not synchronized, so this can happen
         throw new XmlBlasterException(glob, ErrorCode.INTERNAL_UNKNOWN, ME, "SubjectInfo is shutdown, no logout");
      }
      if (log.isLoggable(Level.FINER)) log.finer(ME+": Entering notifyAboutLogout(" + absoluteSessionName + ", " + clearQueue + ")");
      SessionInfo sessionInfo = null;
      synchronized (this.sessionMap) {
         sessionInfo = (SessionInfo)sessionMap.remove(absoluteSessionName);
         this.sessionArrCache = null;
         this.callbackAddressCache = null;
      }
      if (sessionInfo != null) {
         this.dispatchStatistic.incrNumUpdate(sessionInfo.getNumUpdate());
      }
      else {
         log.warning(ME+": Lookup of session with absoluteSessionName=" + absoluteSessionName + " failed");
      }

      if (log.isLoggable(Level.FINEST)) log.finest(this.subjectQueue.toXml(null));

      //if (!isLoggedIn()) {
      //   if (clearQueue || getSubjectQueue().getNumOfEntries() < 1) {
            shutdown(clearQueue, forceShutdownEvenIfEntriesExist); // Does shutdown only on last session
      //   }
      //}
   }

   /**
    * Access the unique login name of a client.
    * <br />
    * If not known, its unique key (subjectId) is delivered
    * @return The SessionName object specific for a subject (pubSessionId is null)
    */
   public final SessionName getSubjectName() {
      return this.subjectName;
   }

   /**
    * Cluster wide unique identifier "/node/heron/client/<loginName>" e.g. for logging
    * <p />
    * @return e.g. "client/joe
    */
   public final String getId() {
      return this.subjectName.getAbsoluteName();
   }

   /**
    * @see #getId
    */
   public final String toString() {
      return this.subjectName.getAbsoluteName();
   }

   /**
    * Access the unique login name of a client.
    * <br />
    * If not known, its unique key (subjectId) is delivered
    * @return loginName
    */
   public final String getLoginName() {
      return this.subjectName.getLoginName();
   }

   /**
    * Dump state of this object into a XML ASCII string.
    * <br>
    * @return internal state of SubjectInfo as a XML ASCII string
    */
   public final String toXml() {
      return toXml((String)null);
   }

   /**
    * Dump state of this object into a XML ASCII string.
    * <br>
    * @param extraOffset indenting of tags for nice output
    * @return internal state of SubjectInfo as a XML ASCII string
    */
   public final String toXml(String extraOffset) {
      StringBuffer sb = new StringBuffer(256);
      if (extraOffset == null) extraOffset = "";
      String offset = Constants.OFFSET + extraOffset;

      sb.append(offset).append("<SubjectInfo id='").append(this.subjectName.getAbsoluteName()).append("'>");
      sb.append(offset).append(" <state>").append(getStateStr()).append("</state>");
      if (isAlive()) {
         sb.append(offset).append(" <subjectId>").append(getLoginName()).append("</subjectId>");
         sb.append(subjectQueue.toXml(extraOffset+Constants.INDENT));
         SessionInfo[] sessions = getSessions();
         for (int i=0; i<sessions.length; i++) {
            SessionInfo sessionInfo = sessions[i];
            sb.append(sessionInfo.toXml(extraOffset+Constants.INDENT, (Properties)null));
         }
      }
      sb.append(offset).append("</SubjectInfo>");

      return sb.toString();
   }

   /**
    * Get the SessionInfo with its public session identifier e.g. "5"
    * @return null if not found
    */
   public final SessionInfo getSessionByPublicId(long publicSessionId) {
      if (publicSessionId == 0L) {
         return null;
      }
      SessionName sessionName = new SessionName(glob, subjectName, publicSessionId);
      synchronized (this.sessionMap) {
         return (SessionInfo)this.sessionMap.get(sessionName.getAbsoluteName());
      }
   }

   public final boolean isUndef() {
      return this.state == UNDEF;
   }

   public final boolean isAlive() {
      return this.state == ALIVE;
   }

   public final boolean isDead() {
      return this.state == DEAD;
   }

   public final String getStateStr() {
      if (isAlive()) {
         return "ALIVE";
      }
      else if (isDead()) {
         return "DEAD";
      }
      else if (isUndef()) {
         return "UNDEF";
      }
      else {
         return "INTERNAL_ERROR";
      }
   }

   /**
    * Query the subject queue, can be peeking or consuming.
    * @param querySpec Can be configured to be consuming
    * @see <a href="http://www.xmlBlaster.org/xmlBlaster/doc/requirements/engine.qos.queryspec.QueueQuery.html">The engine.qos.queryspec.QueueQuery requirement</a>
    */
   public MsgUnit[] getSubjectQueueEntries(String querySpec) throws XmlBlasterException {
      if (this.queueQueryPlugin == null) {
         synchronized (this) {
            if (this.queueQueryPlugin == null) {
               this.queueQueryPlugin = new QueueQueryPlugin(this.glob);
            }
         }
      }
      return this.queueQueryPlugin.query(this.subjectQueue, querySpec);
   }



   //=========== Enforced by I_AdminSubject and SubjectInfoProtector.java ================
   /**
    * @return startupTime in seconds
    */
   long getUptime() {
      return (System.currentTimeMillis() - this.startupTime)/1000L;
   }

   public final String getCreationDate() {
      long ll = this.startupTime;
      java.sql.Timestamp tt = new java.sql.Timestamp(ll);
      return tt.toString();
   }

   /**
    * How many update where sent for this client, the sum of all session and
    * subject queues of this clients.
    */
   long getNumUpdate() {
      long numUpdates = this.dispatchStatistic.getNumUpdate(); // The sessions which disappeared already are remembered here
      SessionInfo[] sessions = getSessions();
      for (int i=0; i<sessions.length; i++) {
         SessionInfo sessionInfo = sessions[i];
         numUpdates += sessionInfo.getNumUpdate();
      }
      return numUpdates;
   }

   long getSubjectQueueNumMsgs() {
      return subjectQueue.getNumOfEntries();
   }

   long getSubjectQueueMaxMsgs() {
      return subjectQueue.getMaxNumOfEntries();
   }

   /**
    * Access the number of sessions of this user.
    * @return The number of sessions of this user
    */
   int getNumSessions() {
      return getSessions().length;
   }

   /**
    * @return The max allowed simultaneous logins of this user
    */
   int getMaxSessions() {
      return this.maxSessions;
   }

   /**
    * JMX access.
    * @param Change the max allowed simultaneous logins of this user
    */
   void setMaxSessions(int max) {
      this.maxSessions = max;
   }

   /**
    * Access a list of public session identifier e.g. "1,5,7,12"
    * @return An empty string if no sessions available
    */
   String getSessionList() {
      int numSessions = getNumSessions();
      if (numSessions < 1)
         return "";
      StringBuffer sb = new StringBuffer(numSessions * 30);
      SessionInfo[] sessions = getSessions();
      for (int i=0; i<sessions.length; i++) {
         if (sb.length() > 0)
            sb.append(",");
         sb.append(sessions[i].getPublicSessionId());
      }
      return sb.toString();
   }

   /**
    * Find a session by its public session ID.
    * @param pubSessionId e.g. "-2"
    * @return I_AdminSession or null if not found
    */
   I_AdminSession getSessionByPubSessionId(long pubSessionId) {
      SessionInfo sessionInfo = getSessionByPublicId(pubSessionId);
      return (sessionInfo == null) ? null : sessionInfo.getSessionInfoProtector();
   }

   /**
    * Kills all sessions of this client
    * @return The list of killed sessions (public session IDs), in a human readable string
    */
   String killClient() throws XmlBlasterException {
      int numSessions = getNumSessions();
      long num = getSubjectQueueNumMsgs();
      if (numSessions < 1) {
         shutdown(true, true);
         return getId() + " killed. No sessions where available, but destroyed " + num + " subject queue messages";
      }
      String sessionList = getSessionList();
      while (true) {
         SessionInfo sessionInfo = null;
         synchronized (sessionMap) {
            Iterator iterator = sessionMap.values().iterator();
            if (!iterator.hasNext())
               break;
            sessionInfo = (SessionInfo)iterator.next();
         }
         sessionInfo.killSession();
      }
      /* The upper form is probably better
      SessionInfo[] sessions = getSessions();
      for (int ii=0; ii<sessions.length; ii++) {
         sessions[ii].killSession();
      }
      */
     String post = (num == 0) ? "" : " Destroyed " + num + " subject queue messages";
     return getId() + " sessions " + sessionList + " killed." + post;
   }

   public String[] peekSubjectMessages(int numOfEntries) throws XmlBlasterException {
      return this.glob.peekMessages(this.subjectQueue, numOfEntries, "subject");
   }

   public String[] peekSubjectMessagesToFile(int numOfEntries, String path) throws Exception {
      try {
         return this.glob.peekQueueMessagesToFile(this.subjectQueue, numOfEntries, path, "subject");
      }
      catch (XmlBlasterException e) {
         throw new Exception(e.toString());
      }
   }

   /** 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) {}

   /**
    * JMX: Enforced by interface NotificationBroadcasterSupport
    */
   public MBeanNotificationInfo[] getNotificationInfo() {
      String[] types = new String[] {
         AttributeChangeNotification.ATTRIBUTE_CHANGE
      };
      String name = AttributeChangeNotification.class.getName();
      String description = "TODO: An attribute of this MBean has changed";
      MBeanNotificationInfo info =
         new MBeanNotificationInfo(types, name, description);
      return new MBeanNotificationInfo[] {info};
   }
}