msgunitwrapper.java

来自「java开源的企业总线.xmlBlaster」· Java 代码 · 共 716 行 · 第 1/2 页

      return this.uniqueIdStr;
   }

   public final String getLogId() {
      return getKeyOid() + "/" + getMsgQosData().getRcvTimestamp();
   }

   public final boolean isInternal() {
      return getMsgKeyData().isInternal();
   }

   /**
    * @return ServerEntryFactory.ENTRY_TYPE_MSG_XML or ServerEntryFactory.ENTRY_TYPE_MSG_SERIAL
    */
   public String getEmbeddedType() {
      return this.embeddedType;
   }

   /**
    * The embedded object. 
    * Object[] = { this.msgUnit, new Integer(this.referenceCounter) }  or<br />
    * qos.toXml, key.toXml, contentBytes
    * <p>
    * IMPORTANT NOTE:
    * If you change the data here you need to change MsgQueueUpdateEntry#getEmbeddedObject() as well!
    * Check ServerEntryFactory as well.
    * </p>
    */
   public Object getEmbeddedObject() {
      if (this.embeddedType.equals(ServerEntryFactory.ENTRY_TYPE_MSG_SERIAL)) {
         Object[] obj = { this.msgUnit, new Integer(this.referenceCounter),
                          new Integer(this.historyReferenceCounter) };
         return obj;
      }
      else {
         Object[] obj = { this.msgUnit.getQosData().toXml(), this.msgUnit.getKeyData().toXml(),
                          this.msgUnit.getContent(), new Integer(this.referenceCounter),
                          new Integer(this.historyReferenceCounter) };
         return obj;
      }
   }

   /**
    * Returns a shallow clone
    */
   public Object clone() {
      try {
         return super.clone();
      }
      catch (CloneNotSupportedException e) {
         return null;
      }
   }

   public final String toXml() {
      return toXml((String)null, false);
   }
   
   public final void embeddedObjectToXml(java.io.OutputStream out, java.util.Properties props) throws java.io.IOException {
      /*
      boolean forceReadable = (props!=null) && props.contains("forceReadable"); Constants.TOXML_FORCEREADABLE
      int maxContentLen = -1;
      if (props!=null && props.contains("maxContentLen")) { Constants.TOXML_MAXCONTENTLEN
         try { maxContentLen = new Integer((String)props.get("maxContentLen")).intValue(); } catch(NumberFormatException e) {}
      }
      */
      MsgUnit msgUnit = getMsgUnit();
      if (msgUnit != null)
         msgUnit.toXml(out, props);
   }

   /**
    * Dumps the message. 
    * NOTE: max 80 bytes of the content are displayed
    */
   public String toXml(String extraOffset, boolean forceReadable) {
      StringBuffer sb = new StringBuffer(1024);
      if (extraOffset == null) extraOffset = "";
      String offset = Constants.OFFSET + extraOffset;

      int maxContentDumpSize = 80;
      sb.append(offset).append("<MsgUnitWrapper id='").append(getLogId());
      sb.append("' referenceCount='").append(getReferenceCounter());
      sb.append("' state='").append(getStateStr()).append("'>");
      sb.append(this.msgUnit.toXml(Constants.INDENT + extraOffset, maxContentDumpSize, forceReadable));
      sb.append(offset).append("</MsgUnitWrapper>");
      return sb.toString();
   }

   /**
    * Notification if this entry is added to storage
    * @see org.xmlBlaster.util.queue.I_Entry#added(StorageId)
    */
   public void added(StorageId storageId) {
      log.severe("added("+storageId.getId()+") invocation not expected");
   }

   /**
    * Notification if this entry is removed from storage
    * @see org.xmlBlaster.util.queue.I_Entry#removed(StorageId)
    */
   public void removed(StorageId storageId) {
      log.severe("removed("+storageId.getId()+") invocation not expected");
   }

   /**
    */
   private boolean isAlive() {
      return this.state == ALIVE;
   }
   
   /**
    * The state may still be alive. 
    * @return true is the configured life span is elapsed
    */
   public boolean hasRemainingLife() {
      long lifeTime = getMsgQosData().getLifeTime();
      if (lifeTime > -1) {
         long timeout = getMsgQosData().getRemainingLife();
         if (timeout <= 0L) {
            return false;
         }
      }
      return true;
   }

   public void startExpiryTimer() {
      synchronized (this) {
         if (this.state != ALIVE) {
            log.severe("Unexpected startExpiryTimer in state " + getStateStr());
            return;
         }
         if (this.timerKey != null) {
            log.severe("Unexpected expiry timer in state " + getStateStr());
            return;
            //this.destroyTimer.removeTimeoutListener(this.timerKey);
            //this.timerKey = null;
            //log.error(ME + getLogId(), "Unexpected expiry timer in state " + getStateStr());
         }

         long lifeTime = getMsgQosData().getLifeTime();
         if (lifeTime > -1) {
            long timeout = getMsgQosData().getRemainingLife();
            if (timeout <= 0L) {
               this.state = PRE_EXPIRED;
               timeout = 0L;
               //timeout(null); // Will deadlock if called by constructor // switch to EXPIRED or DESTROYED
               // We span the timer to fire later and destroy us from another thread 
            }
            this.timerKey = this.destroyTimer.addTimeoutListener(this, timeout, null);
         }
      }
   }

   /**
    */
   public boolean isExpired() {
      return this.state == EXPIRED || this.state == PRE_EXPIRED;
   }

   private void toExpired() throws XmlBlasterException {
      synchronized (this) {
         if (this.timerKey != null) {
            this.destroyTimer.removeTimeoutListener(this.timerKey);
            this.timerKey = null;
         }
         if (this.state == EXPIRED) {
            return;
         }
         this.state = EXPIRED;
      }
      
      if (this.referenceCounter <= 0L) {
         toDestroyed();
         return;
      }
      
      if (this.historyReferenceCounter > 0) {
         StorageId st = new StorageId(Constants.RELATING_HISTORY, "dummy");
         incrementReferenceCounter((-1)*this.historyReferenceCounter, st);
      }
   }

   /**
    */
   public boolean isDestroyed() {
      return this.state == DESTROYED;
   }

   /**
    * Called by TopicHandler.java or ReferenceEntry.java
    */
   public void toDestroyed() {
      synchronized (this) {
         if (this.timerKey != null) {
            this.destroyTimer.removeTimeoutListener(this.timerKey);
            this.timerKey = null;
         }
         if (this.state == DESTROYED) {
            return;
         }
         this.state = DESTROYED;
      }
      
      if (log.isLoggable(Level.FINEST)) {
         log.finest("toDestroyed: " + toXml());
         Thread.dumpStack();
      }

      boolean async = false;
      
      if (async)
         this.glob.getTopicAccessor().entryDestroyed_scheduleForExecution(this);
      else {
         TopicHandler topicHandler = this.glob.getTopicAccessor().access(getKeyOid());
         if (topicHandler != null) { // Topic could be erased in the mean time with forceDestroy=true
            try {
               topicHandler.entryDestroyed(this);
            }
            finally {
               this.glob.getTopicAccessor().release(topicHandler);
            }
         }
      }
   }

   /**
    * This timeout occurs after a configured expiration delay
    */
   public final void timeout(Object userData) {
      if (getMsgQosData().isForceDestroy()) {
         toDestroyed();
      }
      else {
         try {
            toExpired();
         }
         catch (XmlBlasterException e) {
            log.severe("Unexpected exception from toExpired() which we can't handle: " + e.getMessage());
         }
      }
   }

   public String getStateStr() {
      if (isAlive()) {
         return "ALIVE";
      }
      else if (isExpired()) {
         return "EXPIRED";
      }
      else if (isDestroyed()) {
         return "DESTROYED";
      }
      return "UNDEF";
   }

   /**
    * @see org.xmlBlaster.util.queue.I_Entry#setStored(boolean)
    */
   public final void setStored(boolean stored) {
      this.stored = stored;
   }

   /**
    * @see org.xmlBlaster.util.queue.I_Entry#isStored()
    */
   public final boolean isStored() {
      return this.stored;
   }
   
   /**
    * Sets this flag to true/false. This flag can be passed to the MsgQueueEntry
    * @param wantReturnObj
    */
   public void setWantReturnObj(boolean wantReturnObj) {
      this.wantReturnObj = wantReturnObj;
   }
   
   /**
    * 
    * @return
    */   
   public boolean getWantReturnObj() {
      return this.wantReturnObj;
   }
   
   /**
    * @return returnObj The carried object used as return QoS in sync or async I_Queue.put() mode, can be null.
    */
   public Object getReturnObj() {
      return this.returnObj;
   }

   /**
    * Set the object to be carried as return value. 
    * NOTE: This can be used only once as the first call to this method
    * destroys the reference to the clone original instance.
    */
   public void setReturnObj(Object returnObj) {
      this.returnObj = returnObj;
   }

   /**
    * Can be used by cache implementation to implement LRU
    * @return null if not previously set by setSortTimestamp()
    */
   public final Timestamp getSortTimestamp() {
      return this.sortTimestamp;
   }

   /**
    * Can be used by cache implementation to implement LRU
    * @return timestamp This is chosen by the cache algorithm
    */
   public final void setSortTimestamp(Timestamp timestamp) {
      this.sortTimestamp = timestamp;
   }

   /*
    * Measure size for XML-ASCII versus java.io.Serializable persistence. 
    * <pre> 
    * java org.xmlBlaster.engine.MsgUnitWrapper
    * </pre> 
    * Result:
    * <p>
    * java.io.Serialized file 'MsgUnitWrapper.ser' size=1407 bytes versus XML dump=123 bytes
    * </p>
   public static void main(String[] args) {
      Global glob = new Global(args);
      String fileName = "MsgUnitWrapper.ser";
      try {
         PublishKey publishKey = new PublishKey(glob, "HA");
         PublishQosServer publishQosServer = new PublishQosServer(glob, "<qos><persistent/></qos>");
         publishQosServer.getData().setPriority(PriorityEnum.HIGH_PRIORITY);
         MsgUnit msgUnit = new MsgUnit(publishKey.getData(), "HO".getBytes(), publishQosServer.getData());
         StorageId storageId = new StorageId("mystore", "someid");
         MsgUnitWrapper msgUnitWrapper = new MsgUnitWrapper(glob, msgUnit, storageId);
         try {
            java.io.FileOutputStream f = new java.io.FileOutputStream(fileName);
            java.io.ObjectOutputStream objStream = new java.io.ObjectOutputStream(f);
            objStream.writeObject(msgUnitWrapper);
            objStream.flush();
            java.io.File file = new java.io.File(fileName);
            System.out.println("SUCCESS written java.io.Serialized file '" + fileName + "' size=" + file.length() +
                               " versus XML dump=" + msgUnitWrapper.getSizeInBytes());
         }
         catch (Exception e) {
            System.err.println("ERROR: " + e.getMessage());
         }
      }
      catch (XmlBlasterException e) {
         System.err.println("ERROR: " + e.getMessage());
      }
   }
   */
}