jdbcqueuecommontableplugin.java

java开源的企业总线.xmlBlaster

            return -1L;
         }
      }
   }


   /**
    * It returns the number of persistent entries in the queue.
    * In case this value is -1L (which means a previous attempt to read from the
    * DB failed) it will synchronize against the DB by making a call to the DB.
    * If that fails it will return -1L.
    *
    * @see I_Queue#getNumOfPersistentBytes()
    */
   public long getNumOfPersistentBytes() {
      try {
         return getNumOfPersistentBytes_(true);
      }
      catch (XmlBlasterException ex) {
         log.severe("getNumOfPersistentBytes, exception: " + ex.getMessage());
         return this.numOfPersistentBytes;
      }
   }


   /**
    * @see I_Queue#getMaxNumOfBytes()
    */
   public long getMaxNumOfBytes() {
      return this.property.getMaxBytes();
   }


   /**
    * Updates the given message queue entry with a new value. Note that this
    * can be used if an entry with the unique id already exists.
    * ?? Does this really make sense here since we need to store history ??
    * ?? Should we define a switch which can deactivate storage of history ??
    */
   public int update(I_QueueEntry queueEntry) throws XmlBlasterException
   {
      throw new XmlBlasterException(glob, ErrorCode.INTERNAL_NOTIMPLEMENTED, ME, "update not implemented");
   }

   public JdbcManagerCommonTable getManager() {
      return this.manager;
   }

   /**
    * Clears everything and removes the queue (i.e. frees the associated table)
    */
   public long clear() {
      try {
         ReturnDataHolder ret = this.manager.deleteFirstEntries(getStorageId().getStrippedId(), -1, -1L);
         this.numOfEntries = 0L;
         this.numOfBytes = 0L;
         this.numOfPersistentEntries = 0L;
         this.numOfPersistentBytes = 0L;
         this.storageSizeListenerHelper.invokeStorageSizeListener();
         return ret.countEntries;
      }
      catch (XmlBlasterException ex) {
         log.severe("exception: " + ex.getMessage());
         return 0;
      }
   }

   /**
    * @see I_Queue#removeHead(I_QueueEntry)
    */
   public long removeHead(I_QueueEntry toEntry) throws XmlBlasterException {
      throw new XmlBlasterException(glob, ErrorCode.INTERNAL_NOTIMPLEMENTED, ME, "removeHead not implemented yet");
   }

   /**
    * Shutdown the implementation, sync with data store
    */
   public void shutdown() {
      synchronized (this) {
         if (log.isLoggable(Level.FINER)) log.finer("shutdown '" + this.storageId + "' (currently the value of 'isDown' is '" + this.isDown + "'");
         if (this.isDown) return;
         this.isDown = true;
      }
      
      this.storageSizeListenerHelper.invokeStorageSizeListener();
      this.removeStorageSizeListener(null);
      
      synchronized (this) {
         this.manager.unregisterQueue(this);
      }
      
      glob.getQueuePluginManager().cleanup(this);
      
      if (glob instanceof ServerScope) {
         ((ServerScope)glob).getStoragePluginManager().cleanup(this);
      }
   }

   public boolean isShutdown() {
      return this.isDown;
   }

   /**
    * @return a human readable usage help string
    */
   public String usage() {
      return "no usage";
   }

   /**
    * @return Internal state as an XML ASCII string
    */
   public final String toXml(String extraOffset) {
      StringBuffer sb = new StringBuffer(1000);
      if (extraOffset == null) extraOffset = "";
      String offset = Constants.OFFSET + extraOffset;

      // NOTE: Recursion problems when using getNumOfEntries() instead of this.numOfEntries
      // if an exception is thrown in getNumOfEntries() which uses toXml to dump the problem ...

      sb.append(offset).append("<JdbcQueueCommonTablePlugin id='").append(getStorageId().getId());
      sb.append("' type='").append(getType());
      sb.append("' version='").append(getVersion());
      sb.append("' numOfEntries='").append(this.numOfEntries);
      sb.append("' numOfBytes='").append(this.numOfBytes);
      sb.append("'>");
      if (this.property != null) {
         sb.append(this.property.toXml(extraOffset+Constants.INDENT));
      }
      else {
         sb.append(offset).append("<isDown>").append(this.isDown).append("</isDown>");
      
      }
      try {
         sb.append(offset).append(" <numOfPersistentsCached>").append(this.numOfPersistentEntries).append("</numOfPersistentsCached>");
         sb.append(offset).append(" <sizeOfPersistentsCached>").append(getNumOfPersistentBytes()).append("</sizeOfPersistentsCached>");

         sb.append(offset).append(" <numOfEntriesCached>").append(this.numOfEntries).append("</numOfEntriesCached>");
         sb.append(offset).append(" <numOfBytesCached>").append(getNumOfBytes()).append("</numOfBytesCached>");

         sb.append(offset).append(" <numOfEntries>").append(getNumOfEntries_()).append("</numOfEntries>");
         sb.append(offset).append(" <numOfBytes>").append(getNumOfBytes_()).append("</numOfBytes>");


         sb.append(offset).append(" <numOfPersistents>").append(getNumOfPersistentEntries_(false)).append("</numOfPersistents>");
         sb.append(offset).append(" <sizeOfPersistents>").append(getNumOfPersistentBytes_(false)).append("</sizeOfPersistents>");
      }
      catch (XmlBlasterException e) {
      }

      sb.append(offset).append("</JdbcQueueCommonTablePlugin>");
      return sb.toString();
   }

   /**
    * Enforced by I_Plugin
    * @see org.xmlBlaster.util.plugin.I_Plugin#init(org.xmlBlaster.util.Global, PluginInfo)
    */
   public void init(org.xmlBlaster.util.Global glob, PluginInfo pluginInfo) {
      this.glob = glob;
      this.pluginInfo = pluginInfo;
   }

   /**
    * Enforced by I_Plugin
    * @return "JDBC"
    */
   public String getType() { return "JDBC"; }

   /**
    * Enforced by I_Plugin
    * @return "1.0"
    */
   public String getVersion() { return "1.0"; }

   /**
    * Enforced by I_StoragePlugin
    * @return the pluginInfo object.
    */
   public PluginInfo getInfo() { return this.pluginInfo; }

   /**
    * Cleans up the current queue (it deletes all the entries and frees the
    * table used and cleans up all tables in the database). This method should
    * never be called, only for testing purposes or to clean up the entire
    * database.
    */
   public void destroy() throws XmlBlasterException {
      this.clear();
      this.shutdown();
      this.property = null;
      this.manager.cleanUp(this.storageId.getStrippedId());
   }


   /////////////////////////// I_Map implementation ///////////////////////
   public I_MapEntry get(final long uniqueId) throws XmlBlasterException {
      long[] idArr = new long[] { uniqueId };
      ArrayList list = this.manager.getEntries(getStorageId(), idArr);
      if (list.size() < 1) {
         return null;
      }
      return (I_MapEntry)list.get(0);
   }

   /**
    * @see I_Map#getAll()
    */
   public I_MapEntry[] getAll(I_EntryFilter entryFilter) throws XmlBlasterException {
      ArrayList list = this.manager.getEntries(getStorageId(), -1, -1L, entryFilter);
      return (I_MapEntry[])list.toArray(new I_MapEntry[list.size()]);
   }

   /**
    * @see I_Map#put(I_MapEntry)
    */
   public int put(I_MapEntry mapEntry) throws XmlBlasterException {
      if (mapEntry == null) {
         throw new XmlBlasterException(glob, ErrorCode.INTERNAL_ILLEGALARGUMENT, ME, "put(I_MapEntry="+mapEntry+")");
      }
      synchronized (this.modificationMonitor) {
         if (put((I_Entry)mapEntry))
            return 1;

         return 0;
      }
   }

   public int remove(final I_MapEntry mapEntry) throws XmlBlasterException {
      int num = removeRandom(mapEntry);
      synchronized (this.modificationMonitor) {
      }
      return num;
   }

   public int remove(final long uniqueId) throws XmlBlasterException {
      int num = removeRandom(uniqueId);
      synchronized (this.modificationMonitor) {
      }
      return num;
   }

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

   /**
    * @see org.xmlBlaster.util.queue.I_StorageProblemNotifier#registerStorageProblemListener(I_StorageProblemListener)
    */
   public boolean registerStorageProblemListener(I_StorageProblemListener listener) {
      if (this.manager == null) return false;
      return this.manager.registerStorageProblemListener(listener);
   }


   /**
    * @see org.xmlBlaster.util.queue.I_StorageProblemNotifier#unRegisterStorageProblemListener(I_StorageProblemListener)
    */
   public boolean unRegisterStorageProblemListener(I_StorageProblemListener listener) {
      if (this.manager == null || listener == null) return false;
      return this.manager.unRegisterStorageProblemListener(listener);
   }


   /**
    * @see I_Map#change(I_MapEntry, I_ChangeCallback)
    */
   public I_MapEntry change(I_MapEntry entry, I_ChangeCallback callback) throws XmlBlasterException {
      synchronized(this) { // is this the correct synchronization ??
         long oldSizeInBytes = entry.getSizeInBytes(); // must be here since newEntry could reference same obj.
         I_MapEntry newEntry = entry;
         if (callback != null) newEntry = callback.changeEntry(entry);
         if (oldSizeInBytes != newEntry.getSizeInBytes()) {
            throw new XmlBlasterException(this.glob, ErrorCode.INTERNAL_UNKNOWN, ME + ".change", "the size of the entry '" + entry.getUniqueId() + "' has changed from '" + oldSizeInBytes + "' to '" + newEntry.getSizeInBytes() +"'. This is not allowed");
         } 
         this.manager.modifyEntry(getStorageId().getStrippedId(), newEntry);
         return newEntry;
      }
   }


   /**
    * @see I_Map#change(long, I_ChangeCallback)
    */
   public I_MapEntry change(long uniqueId, I_ChangeCallback callback) throws XmlBlasterException {
      synchronized (this) {
         I_MapEntry oldEntry = get(uniqueId);
         return change(oldEntry, callback);
      }
   }


   /**
    * @see I_Queue#addStorageSizeListener(I_StorageSizeListener)
    */
   public void addStorageSizeListener(I_StorageSizeListener listener) {
      this.storageSizeListenerHelper.addStorageSizeListener(listener);
   }
   
   /**
    * @see I_Queue#removeStorageSizeListener(I_StorageSizeListener)
    */
   public void removeStorageSizeListener(I_StorageSizeListener listener) {
      this.storageSizeListenerHelper.removeStorageSizeListener(listener);
   }
   
   /**
    * @see I_Queue#hasStorageSizeListener(I_StorageSizeListener)
    */
   public boolean hasStorageSizeListener(I_StorageSizeListener listener) {
      return this.storageSizeListenerHelper.hasStorageSizeListener(listener);
   }

   /**
    * @see I_Storage#getStorageSizeListeners()
    */
   public I_StorageSizeListener[] getStorageSizeListeners() {
      return storageSizeListenerHelper.getStorageSizeListeners();
   }

   /**
    * @see I_Map#embeddedObjectsToXml(OutputStream, Properties)
    */
   public long embeddedObjectsToXml(final OutputStream out, final Properties props) throws Exception {
      if (out == null) return 0;
      entryCounter = 0;
      /*I_Entry[] results = */getAll(new I_EntryFilter() {
         public I_Entry intercept(I_Entry entry, I_Storage storage) {
            entryCounter++;
            try {
               entry.embeddedObjectToXml(out, props);
            }
            catch (IOException e) {
               log.warning("Ignoring dumpToFile() problem: "+e.toString());
            }
            return null;
         }
      });
      
      return entryCounter;
   }
}