persistencecacheplugin.java

java开源的企业总线.xmlBlaster

      }
   }

   /**
    * @see I_Map#remove(I_MapEntry)
    */
   private int removeNoNotify(final I_MapEntry mapEntry) throws XmlBlasterException {
      if (log.isLoggable(Level.FINER)) log.finer("remove(" + mapEntry.getLogId() + ")");
      synchronized (this) {
         // search in RAM storage
         int num = this.transientStore.remove(mapEntry);
         int num2 = 0;
         if (mapEntry.isPersistent() || (num == 0 && numSwapped() > 0)) {
            if (this.persistentStore != null)
               num2 = this.persistentStore.remove(mapEntry);
         }
         return Math.max(num, num2);
      }
   }

   /**
    * @see I_Map#remove(I_MapEntry)
    */
   public int remove(final I_MapEntry mapEntry) throws XmlBlasterException {
      int ret = removeNoNotify(mapEntry);
      this.storageSizeListenerHelper.invokeStorageSizeListener();
      return ret;
   }

   /**
    * @see I_Map#remove(long)
    */
   public int remove(final long uniqueId) throws XmlBlasterException {
      if (log.isLoggable(Level.FINER)) log.finer("remove(" + uniqueId + ")");
      int ret = 0;
      synchronized (this) {
         I_MapEntry mapEntry = get(uniqueId);
         if (mapEntry != null)
            ret = removeNoNotify(mapEntry);
      }
      this.storageSizeListenerHelper.invokeStorageSizeListener();
      return ret;
   }

   /**
    * @see I_Map#removeOldest()
    */
   public I_MapEntry removeOldest() throws XmlBlasterException {
      throw new XmlBlasterException(glob, ErrorCode.INTERNAL_NOTIMPLEMENTED, ME, "removeOldest is not implemented");
   }

   /**
    * @see I_Map#removeTransient()
    */
   public int removeTransient() throws XmlBlasterException {
      throw new XmlBlasterException(glob, ErrorCode.INTERNAL_NOTIMPLEMENTED, ME, "removeTransient is not implemented");
   }

   private long numSwapped() {
      if (this.persistentStore == null) {
         return 0L;
      }
      return this.persistentStore.getNumOfEntries() - this.persistentStore.getNumOfPersistentEntries();
   }

   /**
    * It returns the size of the queue. Note that this call will return the size
    * stored in cache, i.e. it will NOT make a call to the underlying DB.
    *
    * @see I_Map#getNumOfEntries()
    */
   public long getNumOfEntries() {
      long ret = 0L;
      if (this.persistentStore != null && this.isConnected) {
         ret = this.persistentStore.getNumOfEntries();
         if (ret < 0L) return this.transientStore.getNumOfEntries();
         ret += this.transientStore.getNumOfEntries() - this.transientStore.getNumOfPersistentEntries();
         return ret;
      }
      return this.transientStore.getNumOfEntries();
   }

   /** @see I_AdminMap#getNumOfCachedEntries() */
   public long getNumOfCachedEntries() {
      return this.transientStore.getNumOfEntries();
   }

   /**
    * It returns the size of persistent entries in the queue. Note that this call will return the size
    * stored in cache, i.e. it will NOT make a call to the underlying DB.
    *
    * @see I_Map#getNumOfPersistentEntries()
    */
   public long getNumOfPersistentEntries() {
      if (this.persistentStore != null && this.isConnected) {
         final long ret = this.persistentStore.getNumOfPersistentEntries();
         if (ret < 0L) return this.transientStore.getNumOfEntries();
         return ret;
      }
      return this.transientStore.getNumOfPersistentEntries();
   }

   /**
    * @see I_Map#getMaxNumOfEntries()
    */
   public long getMaxNumOfEntries() {
      if (this.persistentStore != null && this.isConnected)
         return this.persistentStore.getMaxNumOfEntries();
      return this.transientStore.getMaxNumOfEntries();
   }

   /**
    * @see I_Map#getMaxNumOfEntriesCache()
    */
   public long getMaxNumOfEntriesCache() {
      return this.transientStore.getMaxNumOfEntries();
   }

   /**
    * @see I_Map#getNumOfBytes()
    */
   public long getNumOfBytes() {
      if (this.persistentStore != null && this.isConnected) {
         long ret = this.persistentStore.getNumOfBytes();
         if (ret < 0L) return this.transientStore.getNumOfBytes();
         ret += this.transientStore.getNumOfBytes() - this.transientStore.getNumOfPersistentBytes();
         return ret;
      }
      return this.transientStore.getNumOfBytes();
   }

   /** @see I_AdminMap#getNumOfCachedBytes */
   public long getNumOfCachedBytes() {
      return this.transientStore.getNumOfBytes();
   }

   /**
    * @see I_Map#getNumOfPersistentBytes()
    */
   public long getNumOfPersistentBytes() {
      if (this.persistentStore != null && this.isConnected) {
         final long ret = this.persistentStore.getNumOfPersistentBytes();
         if (ret < 0L) return this.transientStore.getNumOfPersistentBytes();
         return ret;
      }
      return this.transientStore.getNumOfPersistentBytes();
   }

   /**
    * @see I_AdminMap#setMaxNumOfBytes(long)
    */
   public String setMaxNumOfBytes(long max) {
      return setMaxNum(true, false, max);
   }
   
   /**
    * @see I_AdminMap#setMaxNumOfBytesCache(long)
    */
   public String setMaxNumOfBytesCache(long max) {
      return setMaxNum(true, true, max);
   }
   
   /**
    * @see I_AdminMap#setMaxNumOfEntries(long)
    */
   public String setMaxNumOfEntries(long max) {
      return setMaxNum(false, false, max);
   }
   
   /**
    * @see I_AdminMap#setMaxNumOfEntriesCache(long)
    */
   public String setMaxNumOfEntriesCache(long max) {
      return setMaxNum(false, true, max);
   }
   
   private String setMaxNum(boolean setBytes, boolean setCache, long max) {
      String loc;
      if (setBytes) {
         loc = (setCache) ? "NumOfBytesCache" : "NumOfBytes";
      }
      else {
         loc = (setCache) ? "NumOfEntriesCache" : "NumOfEntries";
      }
      I_Map pm = this.persistentStore;
      if (pm == null)
         return getStorageId() + ": No persistence store found, setMax"+loc+"(" + max + ") ignored";

      Object obj = pm.getProperties();
      if (!(obj instanceof QueuePropertyBase))
         return getStorageId() + ": Configuration does not support setMax"+loc+"(" + max + "), ignored request";
      QueuePropertyBase property = (QueuePropertyBase)obj;
      long oldMax = 0;
      if (setBytes) {
         if (setCache)
            oldMax = property.getMaxBytesCache();
         else
            oldMax = property.getMaxBytes();
      }
      else { // maxEntries
         if (setCache)
            oldMax = property.getMaxEntriesCache();
         else
            oldMax = property.getMaxEntries();
      }

      log.info(getStorageId() + ": Change request of max"+loc+"=" + oldMax + ", to " + max + " ...");

      if (max < oldMax)
         // Allow decreasing, not yet intense tested!!!
         // And TopicHandler.allowedToReconfigureTopicAndFixWrongLimits() limits it currently! 
         ; //return getStorageId() + ": Currently max"+loc+"=" + oldMax + ", decreasing setMax"+loc+"(" + max + ") is not supported";
      else if (max == oldMax)
         return getStorageId() + ": Currently max"+loc+"=" + oldMax + ", changing to setMax"+loc+"(" + max + ") are identical";

      property = (QueuePropertyBase)property.clone();
      if (setBytes) {
         if (setCache)
            property.setMaxBytesCache(max);
         else
            property.setMaxBytes(max);
      }
      else { // maxEntries
         if (setCache)
            property.setMaxEntriesCache(max);
         else
            property.setMaxEntries(max);
      }
      String word = (max < oldMax) ? "decreased" : "increased";

      String tmpRet = getStorageId() + ": Successfully " + word + " max"+loc+"=" + oldMax + " to " + max + ". This is a NOT persistent change and is lost on restart when the configuration of existing msgUnits are recovered from harddisk";
      
      // persistent change
      try {
         // Find out my topic name:
         String tmp = this.storageId.getPostfix();
         ContextNode ctx = ContextNode.valueOf(tmp);
         String oid = ctx.getInstanceName(); 
         if (oid != null) {
            org.xmlBlaster.client.key.PublishKey pk = new org.xmlBlaster.client.key.PublishKey(glob, oid);
            org.xmlBlaster.client.qos.PublishQos pq = new org.xmlBlaster.client.qos.PublishQos(glob);
            pq.getData().setAdministrative(true);
            pq.setPriority(PriorityEnum.MAX_PRIORITY);
            pq.getData().getTopicProperty().setMsgUnitStoreProperty((MsgUnitStoreProperty)property);
            org.xmlBlaster.util.MsgUnit msgUnit = new org.xmlBlaster.util.MsgUnit(pk, "", pq);
            this.glob.getRequestBroker().publish(this.glob.getInternalSessionInfo(), msgUnit);
            String resultStr = getStorageId() + ": Persistenty " + word + " max"+loc+"=" + oldMax + " to " + max + ".";
            log.info(resultStr);
            return resultStr;
         }
         else {
            String resultStr = tmpRet + "\nPersistent change failed as we couldn't determine the topic oid of '" + this.storageId + "'";
            log.severe(resultStr);
            return resultStr;
         }
      }
      catch (XmlBlasterException e) {
         String resultStr = tmpRet + "\n" + e.toString();
         log.warning(resultStr);
         return resultStr;
      }
   }
   
   /**
    * @see I_Map#getMaxNumOfBytes()
    */
   public long getMaxNumOfBytes() {
      if (this.persistentStore != null && this.isConnected)
         return this.persistentStore.getMaxNumOfBytes();
      return this.transientStore.getMaxNumOfBytes();
   }

   /**
    * @see I_Map#getMaxNumOfBytesCache()
    */
   public long getMaxNumOfBytesCache() {
      return this.transientStore.getMaxNumOfBytes();
   }

   /**
    * Clears everything and removes the queue (i.e. frees the associated table)
    */
   public long clear() {
      long ret = 0L;
      try {
         ret = this.transientStore.clear();
      }
      catch (Exception ex) {
         log.severe("could not clear transient storage. Cause: " + ex.getMessage());
      }

      try {
         if (this.persistentStore != null && this.isConnected)
            ret += this.persistentStore.clear();
      }
      catch (Exception ex) {