queuepropertybase.cpp

java开源的企业总线.xmlBlaster

/*------------------------------------------------------------------------------
Name:      QueuePropertyBase.cpp
Project:   xmlBlaster.org
Copyright: xmlBlaster.org, see xmlBlaster-LICENSE file
Comment:   Holding callback queue properties
Version:   $Id: QueuePropertyBase.cpp 13445 2005-07-15 01:54:35Z ruff $
------------------------------------------------------------------------------*/


/**
 * Helper class holding callback queue properties.
 * <p />
 * See ConnectQos for XML syntax.
 * @see org.xmlBlaster.client.qos.ConnectQos
 */

#include <util/qos/storage/QueuePropertyBase.h>
#include <util/lexical_cast.h>
#include <util/Global.h>

namespace org { namespace xmlBlaster { namespace util { namespace qos { namespace storage {

using namespace std;
using namespace org::xmlBlaster::util;
using namespace org::xmlBlaster::util::qos::address;

const long DEFAULT_maxEntriesDefault = 1000L;
const long DEFAULT_maxEntriesCacheDefault = 1000L;
const long DEFAULT_bytesDefault = 10485760L; // 10 MB
const long DEFAULT_bytesCacheDefault = 2097152L; // 2 MB
/** The default settings (as a ratio relative to the maxBytesCache) for the storeSwapLevel */
const double DEFAULT_storeSwapLevelRatio = 0.70;
/** The default settings (as a ratio relative to the maxBytesCache) for the storeSwapBytes */
const double DEFAULT_storeSwapBytesRatio = 0.25;
/** The default settings (as a ratio relative to the maxBytesCache) for the storeSwapLevel */
const double DEFAULT_reloadSwapLevelRatio = 0.30;
/** The default settings (as a ratio relative to the maxBytesCache) for the storeSwapBytes */
const double DEFAULT_reloadSwapBytesRatio = 0.25;
const Timestamp DEFAULT_minExpires = 1000;
const Timestamp DEFAULT_maxExpires = 0;
const string DEFAULT_onOverflow = Constants::ONOVERFLOW_DEADMESSAGE;
const string DEFAULT_onFailure = Constants::ONOVERFLOW_DEADMESSAGE;

// static variables
string DEFAULT_type = "CACHE";
string DEFAULT_version = "1.0";
/** If not otherwise noted a queue dies after the max value, changeable with property e.g. "queue.expires=3600000" milliseconds */
long DEFAULT_expires;


/**
 * Configure property settings, add your own defaults in the derived class
 * @param propertyPrefix e.g. "history" or "callback" or "connection" or ""
 */
void QueuePropertyBase::initialize(const string& propertyPrefix)
{
   //if (log_.call()) log_.call(ME, string("::initialize with property prefix '") + propertyPrefix + "'");
   propertyPrefix_ = propertyPrefix;
   string prefix = getPrefix();
   //if (log_.trace()) log_.trace(ME, string("::initialize: got the prefix '") + prefix + "'");

   // Do we need this range settings?
   setMinExpires(global_.getProperty().getTimestampProperty("queue/expires.min", DEFAULT_minExpires));
   setMaxExpires(global_.getProperty().getTimestampProperty("queue/expires.max", DEFAULT_maxExpires)); // Long.MAX_VALUE);
   //if (log_.trace()) log_.trace(ME, "::initialize: expires set");
   if (nodeId_ != "") {
      setMinExpires(global_.getProperty().getTimestampProperty("queue/expires.min["+nodeId_+"]", getMinExpires()));
      setMaxExpires(global_.getProperty().getTimestampProperty("queue/expires.max["+nodeId_+"]", getMaxExpires())); // Long.MAX_VALUE);
   }
   //if (log_.trace()) log_.trace(ME, "::initialize: expires for the specific node set");

   // prefix is e.g. "queue/history/" or "persistence/topicStore/"
   setMaxEntries(global_.getProperty().getLongProperty(prefix+"maxEntries", DEFAULT_maxEntriesDefault));
   //if (log_.trace()) log_.trace(ME, "::initialize: setMaxEntries -> " + lexical_cast<string>(getMaxEntries()));
   setMaxEntriesCache(global_.getProperty().getLongProperty(prefix+"maxEntriesCache", DEFAULT_maxEntriesCacheDefault));
   //if (log_.trace()) log_.trace(ME, "::initialize: setMaxEntriesCache -> " + lexical_cast<string>(getMaxEntriesCache()));
   setMaxBytes(global_.getProperty().getLongProperty(prefix+"maxBytes", DEFAULT_bytesDefault));
   //if (log_.trace()) log_.trace(ME, "::initialize: setMaxBytes -> " + lexical_cast<string>(getMaxBytes()));
   setMaxBytesCache(global_.getProperty().getLongProperty(prefix+"maxBytesCache", DEFAULT_bytesCacheDefault));
   //if (log_.trace()) log_.trace(ME, "::initialize: setMaxBytesCache -> " + lexical_cast<string>(getMaxBytesCache()));

   setStoreSwapLevel(global_.getProperty().getLongProperty(prefix+"storeSwapLevel", (long)(DEFAULT_storeSwapLevelRatio*maxBytesCache_)));
   setStoreSwapBytes(global_.getProperty().getLongProperty(prefix+"storeSwapBytes", (long)(DEFAULT_storeSwapBytesRatio*maxBytesCache_)));
   setReloadSwapLevel(global_.getProperty().getLongProperty(prefix+"reloadSwapLevel", (long)(DEFAULT_reloadSwapLevelRatio*maxBytesCache_)));
   setReloadSwapBytes(global_.getProperty().getLongProperty(prefix+"reloadSwapBytes", (long)(DEFAULT_reloadSwapBytesRatio*maxBytesCache_)));

   //if (log_.trace()) log_.trace(ME, "::initialize: values for the swap control set");

   setExpires(global_.getProperty().getTimestampProperty(prefix+"expires", DEFAULT_maxExpires));
   setOnOverflow(global_.getProperty().getStringProperty(prefix+"onOverflow", DEFAULT_onOverflow));
   setOnFailure(global_.getProperty().getStringProperty(prefix+"onFailure", DEFAULT_onFailure));
   setType(global_.getProperty().getStringProperty(prefix+"type", DEFAULT_type));
   setVersion(global_.getProperty().getStringProperty(prefix+"version", DEFAULT_version));

   //if (log_.trace()) log_.trace(ME, "::initialize: going to set specific node properties");

   if (nodeId_ != "") {
      setMaxEntries(global_.getProperty().getLongProperty(prefix+"maxEntries["+nodeId_+"]", getMaxEntries()));
      setMaxEntriesCache(global_.getProperty().getLongProperty(prefix+"maxEntriesCache["+nodeId_+"]", getMaxEntriesCache()));
      setMaxBytes(global_.getProperty().getLongProperty(prefix+"maxBytes["+nodeId_+"]", getMaxBytes()));
      setMaxBytesCache(global_.getProperty().getLongProperty(prefix+"maxBytesCache["+nodeId_+"]", getMaxBytesCache()));
      setStoreSwapLevel(global_.getProperty().getLongProperty(prefix+"storeSwapLevel["+nodeId_+"]", getStoreSwapLevel()));
      setStoreSwapBytes(global_.getProperty().getLongProperty(prefix+"storeSwapBytes["+nodeId_+"]", getStoreSwapBytes()));
      setReloadSwapLevel(global_.getProperty().getLongProperty(prefix+"reloadSwapLevel["+nodeId_+"]", getReloadSwapLevel()));
      setReloadSwapBytes(global_.getProperty().getLongProperty(prefix+"reloadSwapBytes["+nodeId_+"]", getReloadSwapBytes()));
      setExpires(global_.getProperty().getTimestampProperty(prefix+"expires["+nodeId_+"]", getExpires()));
      setOnOverflow(global_.getProperty().getStringProperty(prefix+"onOverflow["+nodeId_+"]", getOnOverflow()));
      setOnFailure(global_.getProperty().getStringProperty(prefix+"onFailure["+nodeId_+"]", getOnFailure()));
      setType(global_.getProperty().getStringProperty(prefix+"type["+nodeId_+"]", getType()));
      setVersion(global_.getProperty().getStringProperty(prefix+"version["+nodeId_+"]", getVersion()));
   }
   if (log_.trace()) log_.trace(ME, string("::initialized to: ") + toXml());
}

/*
   void QueuePropertyBase::initialize()
   {
      // Do we need this range settings?
      setMinExpires(global_.getProperty().getTimestampProperty("queue/expires.min", DEFAULT_minExpires));
      setMaxExpires(global_.getProperty().getTimestampProperty("queue/expires.max", DEFAULT_maxExpires)); // Long.MAX_VALUE);
      if (nodeId_ != "") {
         setMinExpires(global_.getProperty().getTimestampProperty(string("queue/expires.min[")+nodeId_+string("]"), getMinExpires()));
         setMaxExpires(global_.getProperty().getTimestampProperty(string("queue/expires.max[")+nodeId_+string("]"), getMaxExpires())); // Long.MAX_VALUE);
      }

//         PluginInfo pluginInfo = new PluginInfo(glob, null, global_.getProperty().get("queue/defaultPlugin", DEFAULT_type));
//         DEFAULT_type = pluginInfo.getType();
//         DEFAULT_version = pluginInfo.getVersion();
   }
*/


   QueuePropertyBase::QueuePropertyBase(Global& global, const string& nodeId)
      : ME("QueuePropertyBase"),
        global_(global),
        log_(global.getLog("org.xmlBlaster.util.qos")),
        type_(DEFAULT_type),
        version_(DEFAULT_version),
        minExpires_(DEFAULT_minExpires),
        maxExpires_(DEFAULT_minExpires),
        relating_(Constants::RELATING_CALLBACK),
        expires_(DEFAULT_expires),
        maxEntries_(DEFAULT_maxEntriesDefault),
        maxBytes_(DEFAULT_bytesDefault),
        maxEntriesCache_(DEFAULT_maxEntriesCacheDefault),
        storeSwapLevel_(0),
        storeSwapBytes_(0),
        reloadSwapLevel_(0),
        reloadSwapBytes_(0),
        maxBytesCache_(DEFAULT_bytesCacheDefault),
        onOverflow_(Constants::ONOVERFLOW_DEADMESSAGE),
        onFailure_(Constants::ONOVERFLOW_DEADMESSAGE),
        addressArr_(),
        nodeId_(nodeId),
        propertyPrefix_(""),
        rootTagName_("queue")
   {
   }

   QueuePropertyBase::QueuePropertyBase(const QueuePropertyBase& prop)
      : ME("QueuePropertyBase"), global_(prop.global_), log_(prop.log_)
   {
      copy(prop);
   }

   QueuePropertyBase&
   QueuePropertyBase::operator =(const QueuePropertyBase& prop)
   {
      copy(prop);
      return *this;
   }


   QueuePropertyBase::~QueuePropertyBase()
   {
      addressArr_.clear();
      // delete all entries of the address vector since they are pointers
      // owned by this object.
//      cleanupAddresses();
   }

   /**
    * @param relating    To what is this queue related: Constants.RELATING_CALLBACK | Constants.RELATING_SUBJECT | Constants.RELATING_CLIENT
    */
   void QueuePropertyBase::setRelating(const string& relating)
   {
      if (Constants::RELATING_CALLBACK == relating)
         relating_ = Constants::RELATING_CALLBACK;
      else if (Constants::RELATING_SUBJECT == relating)
         relating_ = Constants::RELATING_SUBJECT;
      else if (Constants::RELATING_CLIENT == relating)
         relating_ = Constants::RELATING_CLIENT;
      else if (Constants::RELATING_HISTORY == relating)
         relating_ = Constants::RELATING_HISTORY;
      else if (Constants::RELATING_MSGUNITSTORE == relating)
         relating_ = Constants::RELATING_MSGUNITSTORE;
      else if (Constants::RELATING_TOPICSTORE == relating)
         relating_ = Constants::RELATING_TOPICSTORE;
      else {
         log_.warn(ME, string("Ignoring relating=") + relating);
      }
   }

   /**
    * Returns the queue type.
    * @return relating    To what is this queue related: Constants.RELATING_CALLBACK | Constants.RELATING_SUBJECT
    */
   string QueuePropertyBase::getRelating() const
   {
      return relating_;
   }

   /**
    * Span of life of this queue.
    * @return Expiry time in milliseconds or 0L if forever
    */
   Timestamp QueuePropertyBase::getExpires() const
   {
      return expires_;
   }

   /**
    * Span of life of this queue.
    * @param Expiry time in milliseconds
    */
   void QueuePropertyBase::setExpires(Timestamp expires)
   {
      if (maxExpires_ <= 0) expires_ = expires;
      else if ( (expires>0) && (maxExpires_>0) && (expires>maxExpires_) )
         expires_ = maxExpires_;
      else if ( (expires<=0) && (maxExpires_>0) )
         expires_ = maxExpires_;

      if ( (expires>0) && (expires<minExpires_) )
         expires_ = minExpires_;
   }


   /**
    * Max number of messages for this queue.
    * <br />
    * @return number of messages
    */
   long QueuePropertyBase::getMaxEntries() const
   {
      return maxEntries_;
   }

   /**
    * Max number of messages for this queue.
    * <br />
    * @param maxEntries
    */
   void QueuePropertyBase::setMaxEntries(long maxEntries)
   {
      maxEntries_ = maxEntries;
   }


   /**
    * The plugin type. 
    * <br />
    * @return e.g. "CACHE"
    */
   string QueuePropertyBase::getType() const
   {
      return type_;
   }

   /**
    * The plugin type
    * <br />
    * @param type
    */
   void QueuePropertyBase::setType(const string& type)
   {
      type_ = type;
   }

   /**
    * The plugin version. 
    * <br />
    * @return e.g. "1.0"
    */
   string QueuePropertyBase::getVersion() const
   {
      return version_;
   }

   /**
    * The plugin version
    * <br />
    * @param version
    */
   void QueuePropertyBase::setVersion(const string& version)
   {
      version_ = version;
   }

   /**
    * Max number of messages for the cache of this queue.
    * <br />
    * @return number of messages
    */
   long QueuePropertyBase::getMaxEntriesCache() const
   {
      return maxEntriesCache_;
   }

   /**
    * Max number of messages for the cache of this queue.
    * <br />
    * @param maxEntries
    */
   void QueuePropertyBase::setMaxEntriesCache(long maxEntriesCache)
   {
      maxEntriesCache_ = maxEntriesCache;
   }


   /**
    * Max message queue size.
    * <br />