sessioninfo.java

java开源的企业总线.xmlBlaster

         synchronized (this) {
            if (this.statistic == null) {
               DispatchManager dispatchManager = this.dispatchManager;
               if (dispatchManager != null)
                  this.statistic = dispatchManager.getDispatchStatistic();
               else
                  this.statistic = new DispatchStatistic();
            }
         }
      }
      return this.statistic;
   }

   /**
    * Call this to reactivate the session expiry to full value
    */
   public final void refreshSession() throws XmlBlasterException {
      if (connectQos.getSessionTimeout() > 0L) {
         synchronized (this.EXPIRY_TIMER_MONITOR) {
            Timeout expiryTimer = this.expiryTimer;
            if (expiryTimer != null) {
               this.timerKey = expiryTimer.addOrRefreshTimeoutListener(this, connectQos.getSessionTimeout(), null, this.timerKey);
            }
         }
      }
      else {
         removeExpiryTimer();
      }
   }

   /**
    * We are notified when this session expires.
    * @param userData You get bounced back your userData which you passed
    *                 with Timeout.addTimeoutListener()
    */
   public final void timeout(Object userData) {
      this.lock.lock();
      try {
         this.timerKey = null;
         log.warning(ME+": Session timeout for " + getLoginName() + " occurred, session '" + getSecretSessionId() + "' is expired, autologout");
         DisconnectQosServer qos = new DisconnectQosServer(glob);
         qos.deleteSubjectQueue(true);
         try {
            glob.getAuthenticate().disconnect(getAddressServer(), getSecretSessionId(), qos.toXml());
         } catch (XmlBlasterException e) {
            log.severe(ME+": Internal problem with disconnect: " + e.toString());
         }
      }
      finally {
         this.lock.release();
      }
   }

   /**
    * Is the given address the same as our?
    */
   public final boolean hasAddress(AddressBase addr) {
      if (addr == null) return false;
      I_Queue sessionQueue = getSessionQueue();
      if (sessionQueue == null) return false;
      CallbackAddress[] arr = ((CbQueueProperty)sessionQueue.getProperties()).getCallbackAddresses();
      for (int ii=0; arr!=null && ii<arr.length; ii++) {
         // if (arr[ii].isSameAddress(addr))
         if (arr[ii].equals(addr))
            return true;
      }
      return false;
   }

   /*
    * Put the given message into the queue
   public final void queueMessage(MsgUnit msgUnit) throws XmlBlasterException {
      if (log.isLoggable(Level.FINER)) log.call(ME, "Queing message");
      if (msgUnit == null) {
         log.error(ME+".Internal", "Can't queue null message");
         throw new XmlBlasterException(ME+".Internal", "Can't queue null message");
      }

      MsgQueueUpdateEntry entry = new MsgQueueUpdateEntry(glob, msgUnit, this.sessionQueue, getSessionName());

      queueMessage(entry);
   }
    */

   /**
    * Put the given message entry into the queue
    */
   public final void queueMessage(MsgQueueEntry entry) throws XmlBlasterException {
      I_Queue sessionQueue = this.sessionQueue;
      if (!hasCallback() || sessionQueue == null) {
         if (log.isLoggable(Level.FINE)) log.fine(ME+": Queing PtP message without having configured a callback to the client, the client needs to reconnect with a valid callback address later");
         //if (!connectQos.getSessionName().isPubSessionIdUser()) { // client has specified its own publicSessionId (> 0)
         //   throw new XmlBlasterException(glob, ErrorCode.USER_CONFIGURATION, ME, "No callback server is configured, can't callback client to send message " + entry.getKeyOid());
         //}
      }
      if (getPublicSessionId() < 0 && entry.isPersistent()) {
         entry.setPersistent(false);
         if (!this.transientWarn) {
            log.warning(ME+": Handling persistent messages in callback queue as transient as we have a login session with a negative public session id (we can't reconnect to same queue after restart)");
            this.transientWarn = true;
         }
      }
      sessionQueue.put(entry, I_Queue.USE_PUT_INTERCEPTOR);

      I_Checkpoint cp = glob.getCheckpointPlugin();
      if (cp != null) {
         cp.passingBy(I_Checkpoint.CP_UPDATE_QUEUE_ADD, entry.getMsgUnit(),
                  this.getSessionName(), null);
      }
   }

   public final ConnectQosServer getConnectQos() {
      return this.connectQos;
   }

   public final void updateConnectQos(ConnectQosServer newConnectQos) throws XmlBlasterException {
      boolean wantsCallbacks = (newConnectQos.getSessionCbQueueProperty().getCallbackAddresses().length > 0);

      CbQueueProperty cbQueueProperty = newConnectQos.getSessionCbQueueProperty();
      I_Queue sessionQueue = this.sessionQueue;
      if (sessionQueue != null) sessionQueue.setProperties(cbQueueProperty);
      if (wantsCallbacks && hasCallback()) {
         DispatchManager dispatchManager = this.dispatchManager;
         if (dispatchManager != null) {
            dispatchManager.updateProperty(cbQueueProperty.getCallbackAddresses());
            log.info(ME+": Successfully reconfigured callback address with new settings, other reconfigurations are not yet implemented");
            dispatchManager.notifyAboutNewEntry();
         }
      }
      else if (wantsCallbacks && !hasCallback()) {
         log.info(ME+": Successfully reconfigured and created dispatch manager with given callback address");
         DispatchManager tmpDispatchManager = new DispatchManager(glob, this.msgErrorHandler,
                              this.securityCtx, this.sessionQueue, (I_ConnectionStatusListener)null,
                              newConnectQos.getSessionCbQueueProperty().getCallbackAddresses(), this.sessionName);
         DispatchManager dispatchManager = this.dispatchManager;
         if (dispatchManager != null)
            tmpDispatchManager.setDispatcherActive(dispatchManager.isDispatcherActive());
         this.dispatchManager = tmpDispatchManager;
      }
      else if (!wantsCallbacks && hasCallback()) {
         DispatchManager dispatchManager = this.dispatchManager;
         if (dispatchManager != null) {
            dispatchManager.shutdown();
            log.info(ME+": Successfully shutdown dispatch manager as no callback address is configured");
         }
         this.dispatchManager = null;
      }
      else if (!wantsCallbacks && !hasCallback()) {
         if (log.isLoggable(Level.FINE)) log.fine(ME+": No callback exists and no callback is desired");
         // nothing to do
      }

      // Remember persistent values:
      newConnectQos.isFromPersistenceRecovery(this.connectQos.isFromPersistenceRecovery());
      newConnectQos.setPersistenceUniqueId(this.connectQos.getPersistenceUniqueId());
      if (this.connectQos.getData().isPersistent()) // otherwise persistent sessions could be made transient
         newConnectQos.getData().setPersistent(true); // and would never be deleted from persistence.
      this.connectQos = newConnectQos; // Replaces ConnectQosServer settings like bypassCredentialCheck

      // "__remoteProperties"
      if (newConnectQos.getData().getClientProperty(Constants.CLIENTPROPERTY_REMOTEPROPERTIES, false)) {
          mergeRemoteProperties(newConnectQos.getData().getClientProperties());
      }
   }

   /**
    * Access the unique login name of a client.
    * <br />
    * @return loginName
    */
   public final String getLoginName() {
      SubjectInfo subjectInfo = this.subjectInfo;
      return (subjectInfo==null)?"--":subjectInfo.getLoginName();
   }

   /**
    * Accessing the SubjectInfo object
    * <p />
    * @return SubjectInfo
    */
   public final SubjectInfo getSubjectInfo() {
      return this.subjectInfo;
   }

   /**
    * @return The secret sessionId of this login session
    */
   public String getSecretSessionId() {
      return this.securityCtx.getSecretSessionId();
   }

   public I_Session getSecuritySession() {
      return this.securityCtx;
   }

   public void setSecuritySession(I_Session ctx) {
      this.securityCtx = ctx;
   }

   /**
    * This queue holds all messages which where addressed to this session
    * @return null if no callback was configured
    */
   public I_Queue getSessionQueue() {
      return this.sessionQueue;
   }

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

   public final SessionName getSessionName() {
      return this.sessionName;
   }

   /**
    * Check cluster wide if the sessions are identical
    */
   public boolean isSameSession(SessionInfo sessionInfo) {
      return getId().equals(sessionInfo.getId());
   }

   /**
    * We register for queue size changes and notify the subject queue if
    * we are willing to accept messages again.
    * Enforced by I_StorageSizeListener
    */
   public void changed(I_Storage storage, long numEntries, long numBytes, boolean isShutdown) {
      if (isShutdown) return;
      SubjectInfo subjectInfo = getSubjectInfo();
      boolean hasSubjectEntries = (subjectInfo == null) ? false : subjectInfo.getSubjectQueue().getNumOfEntries() > 0;
      if (lastNumEntries != numEntries) {
         I_Queue sessionQueue = this.sessionQueue;
         long max = (sessionQueue == null) ? 0 : sessionQueue.getMaxNumOfEntries();
         if (hasSubjectEntries && numEntries < max && lastNumEntries > numEntries) {
            if (log.isLoggable(Level.FINE)) log.fine(ME+": SessionQueue has emptied from " + lastNumEntries +
                           " to " + numEntries + " entries, calling SubjectInfoShuffler.shuffle()");
            this.glob.getSubjectInfoShuffler().shuffle(subjectInfo);
         }
         lastNumEntries = numEntries; // to avoid recursion
      }
   }

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

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

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

      sb.append(offset).append("<SessionInfo id='").append(getId());

      Timeout expiryTimer = this.expiryTimer;
      long timeToLife = (expiryTimer != null) ? expiryTimer.spanToTimeout(timerKey) : 0;
      sb.append("' timeout='").append(timeToLife).append("'>");

      // Avoid dump of password
      if (props == null) props = new Properties();
      props.put(Constants.TOXML_NOSECURITY, ""+true);
      sb.append(this.connectQos.toXml(extraOffset+Constants.INDENT, props));

      DispatchManager dispatchManager = this.dispatchManager;
      if (dispatchManager != null) {
         sb.append(dispatchManager.toXml(extraOffset+Constants.INDENT));
      }
      else {
         sb.append(offset).append(Constants.INDENT).append("<DispatchManager id='NULL'/>");
      }

      I_Queue sessionQueue = this.sessionQueue;
      if (sessionQueue != null) {
         sb.append(sessionQueue.toXml(extraOffset+Constants.INDENT));
      }
      sb.append(offset).append("</SessionInfo>");

      return sb.toString();
   }

   //=========== Enforced by I_AdminSession ================
   public String getQos() {
      return (this.connectQos == null) ? "" : this.connectQos.toXml();
   }

   public final boolean isCallbackConfigured() {
      return hasCallback();
   }

   public final long getUptime() {
      return (System.currentTimeMillis() - this.startupTime)/1000L;
   }

   public final String getConnectionState() {
      if (this.dispatchManager != null) {
         return this.dispatchManager.getDispatchConnectionsHandler().getState().toString();
      }
      else {
         return "UNDEF";
      }

   }

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

   public final String getSessionTimeoutExpireDate() {
      long timeToLife = this.expiryTimer.spanToTimeout(timerKey);
      if (timeToLife == -1) {
         return "unlimited";
      }
      long ll = System.currentTimeMillis() + timeToLife;
      java.sql.Timestamp tt = new java.sql.Timestamp(ll);
      return tt.toString();
   }

   // JMX
   public final String getAliveSinceDate() {
      if (this.dispatchManager == null) return "";
      long ll = this.dispatchManager.getAliveSinceTime();
      if (ll == 0) return "";
      java.sql.Timestamp tt = new java.sql.Timestamp(ll);
      return tt.toString();
   }

   // JMX
   public final String getPollingSinceDate() {
      if (this.dispatchManager == null) return "";
      long ll = this.dispatchManager.getPollingSinceTime();
      if (ll == 0) return "";
      java.sql.Timestamp tt = new java.sql.Timestamp(ll);
      return tt.toString();
   }

   public final String getLastCallbackException() {
      return getDispatchStatistic().getLastDeliveryException();
   }

   public final void clearLastCallbackException() {
      getDispatchStatistic().setLastDeliveryException("");
   }

   public final int getNumCallbackExceptions() {
      return getDispatchStatistic().getNumDeliveryExceptions();
   }

   public final long getNumPublish() {
      return getDispatchStatistic().getNumPublish();
   }

   public final long getNumSubscribe() {
      return getDispatchStatistic().getNumSubscribe();
   }

   public final long getNumUnSubscribe() {
      return getDispatchStatistic().getNumUnSubscribe();
   }

   public final long getNumGet() {
      return getDispatchStatistic().getNumGet();
   }

   public final long getNumErase() {
      return getDispatchStatistic().getNumErase();
   }

   public final long getNumUpdateOneway() {
      return getDispatchStatistic().getNumUpdateOneway();
   }

   public final long getNumUpdate() {