util/qos/storage/QueuePropertyBase.cpp

Go to the documentation of this file.

/*------------------------------------------------------------------------------
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 $
------------------------------------------------------------------------------*/


#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
const double DEFAULT_storeSwapLevelRatio = 0.70;
const double DEFAULT_storeSwapBytesRatio = 0.25;
const double DEFAULT_reloadSwapLevelRatio = 0.30;
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";
long DEFAULT_expires;


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();
   }

   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);
      }
   }

   string QueuePropertyBase::getRelating() const
   {
      return relating_;
   }

   Timestamp QueuePropertyBase::getExpires() const
   {
      return expires_;
   }

   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_;
   }


   long QueuePropertyBase::getMaxEntries() const
   {
      return maxEntries_;
   }

   void QueuePropertyBase::setMaxEntries(long maxEntries)
   {
      maxEntries_ = maxEntries;
   }


   string QueuePropertyBase::getType() const
   {
      return type_;
   }

   void QueuePropertyBase::setType(const string& type)
   {
      type_ = type;
   }

   string QueuePropertyBase::getVersion() const
   {
      return version_;
   }

   void QueuePropertyBase::setVersion(const string& version)
   {
      version_ = version;
   }

   long QueuePropertyBase::getMaxEntriesCache() const
   {
      return maxEntriesCache_;
   }

   void QueuePropertyBase::setMaxEntriesCache(long maxEntriesCache)
   {
      maxEntriesCache_ = maxEntriesCache;
   }


   long QueuePropertyBase::getMaxBytes() const
   {
      return maxBytes_;
   }

   void QueuePropertyBase::setMaxBytes(long maxBytes)
   {
      maxBytes_ = maxBytes;
   }


   long QueuePropertyBase::getMaxBytesCache() const
   {
      return maxBytesCache_;
   }


   long QueuePropertyBase::getStoreSwapLevel() const
   {
      return storeSwapLevel_;
   }

   void QueuePropertyBase::setStoreSwapLevel(long storeSwapLevel)
   {
      storeSwapLevel_ = storeSwapLevel;
   }

   long QueuePropertyBase::getStoreSwapBytes() const
   {
      return storeSwapBytes_;
   }

   void QueuePropertyBase::setStoreSwapBytes(long storeSwapBytes)
   {
      storeSwapBytes_ = storeSwapBytes;
   }

   long QueuePropertyBase::getReloadSwapLevel() const
   {
      return reloadSwapLevel_;
   }

   void QueuePropertyBase::setReloadSwapLevel(long reloadSwapLevel)
   {
      reloadSwapLevel_ = reloadSwapLevel;
   }

   long QueuePropertyBase::getReloadSwapBytes() const
   {
      return reloadSwapBytes_;
   }

   void QueuePropertyBase::setReloadSwapBytes(long reloadSwapBytes)
   {
      reloadSwapBytes_ = reloadSwapBytes;
   }

   void QueuePropertyBase::setMaxBytesCache(long maxBytesCache)
   {
      maxBytesCache_ = maxBytesCache;
   }


   void QueuePropertyBase::setOnOverflow(const string& onOverflow)
   {
      /*
      if (Constants.ONOVERFLOW_BLOCK.equalsIgnoreCase(onOverflow)) {
         this.onOverflow = Constants.ONOVERFLOW_BLOCK;
      }
      */
      if (Constants::ONOVERFLOW_DEADMESSAGE == onOverflow) {
         onOverflow_ = Constants::ONOVERFLOW_DEADMESSAGE;
      }
      else if (Constants::ONOVERFLOW_DISCARDOLDEST == onOverflow) {
         onOverflow_ = Constants::ONOVERFLOW_DISCARDOLDEST;

         onOverflow_ = Constants::ONOVERFLOW_DEADMESSAGE; // TODO !!!
         log_.error(ME, string("queue onOverflow='") + string(Constants::ONOVERFLOW_DISCARDOLDEST) + string("' is not implemented, switching to ") + onOverflow_ + string(" mode"));
      }
      else {
         onOverflow_ = Constants::ONOVERFLOW_DEADMESSAGE;
         log_.warn(ME, string("The queue onOverflow attribute is invalid '") + onOverflow + string("', setting to '") + onOverflow_ + string("'"));
      }
   }

   string QueuePropertyBase::getOnOverflow() const
   {
      return onOverflow_;
   }

   /*
    * The default mode, when queue is full the publisher blocks until
    * there is space again.
   public final boolean onOverflowBlock() {
      if (Constants.ONOVERFLOW_BLOCK.equalsIgnoreCase(getOnOverflow()))
         return true;
      return false;
   }
    */

   void QueuePropertyBase::setOnFailure(const string& onFailure)
   {
      if (Constants::ONOVERFLOW_DEADMESSAGE == onFailure)
         onFailure_ = Constants::ONOVERFLOW_DEADMESSAGE;
      else {
         log_.warn(ME, string("The queue onFailure attribute is invalid '") + onFailure + string("', setting to 'deadMessage'"));
         onFailure_ = Constants::ONOVERFLOW_DEADMESSAGE;
      }
   }

   string QueuePropertyBase::getOnFailure() const
   {
      return onFailure_;
   }

   bool QueuePropertyBase::onFailureDeadMessage()
   {
      if (Constants::ONOVERFLOW_DEADMESSAGE == getOnFailure())
         return true;
      return false;
   }

   AddressVector QueuePropertyBase::getAddresses() const
   {
      return addressArr_;
   }

   string QueuePropertyBase::toXml(const string& extraOffset) const
   {
      string offset = Constants::OFFSET + extraOffset;
      string ret;   
      ret += offset + string("<!-- QueuePropertyBase -->");

      ret += offset + string("<queue relating='") + getRelating();
      if (DEFAULT_type != getType())
         ret += string("' type='") + getType();
      if (DEFAULT_version != getVersion())
         ret += string("' version='") + getVersion();
      if (DEFAULT_maxEntriesDefault != getMaxEntries())
         ret += string("' maxEntries='") + lexical_cast<std::string>(getMaxEntries());
      if (DEFAULT_maxEntriesCacheDefault != getMaxEntriesCache())
         ret += string("' maxEntriesCache='") + lexical_cast<std::string>(getMaxEntriesCache());
      if (DEFAULT_bytesDefault != getMaxBytes())
         ret += string("' maxBytes='") + lexical_cast<std::string>(getMaxBytes());
      if (DEFAULT_bytesCacheDefault != getMaxBytesCache())
         ret += string("' maxBytesCache='") + lexical_cast<std::string>(getMaxBytesCache());
      ret += string("' storeSwapLevel='") + lexical_cast<std::string>(getStoreSwapLevel());
      ret += string("' storeSwapBytes='") + lexical_cast<std::string>(getStoreSwapBytes());
      ret += string("' reloadSwapLevel='") + lexical_cast<std::string>(getReloadSwapLevel());
      ret += string("' reloadSwapBytes='") + lexical_cast<std::string>(getReloadSwapBytes());
      if (DEFAULT_expires != getExpires())
         ret += string("' expires='") + lexical_cast<std::string>(getExpires());
      if (DEFAULT_onOverflow != getOnOverflow())
         ret += string("' onOverflow='") + getOnOverflow();
      if (DEFAULT_onFailure != getOnFailure())
         ret += string("' onFailure='") + getOnFailure();

      if (!addressArr_.empty()) {
         ret += string("'>");
         AddressVector::const_iterator iter = addressArr_.begin();
         while (iter != addressArr_.end()) {
            ret += (*iter)->toXml(extraOffset + Constants::INDENT);
            iter++;
         }
         ret += offset + string("</queue>");

      }
      else
         ret += string("'/>");
      return ret;
   }

   Global& QueuePropertyBase::getGlobal()
   {
      return global_;
   }

/*
   void QueuePropertyBase::cleanupAddresses()
   {
      AddressVector::iterator iter = addressArr_.begin();
      while (iter != addressArr_.end()) {
        AddressBase* el = *iter;
        addressArr_.erase(iter);
        delete el;
        iter = addressArr_.begin();
      }
   }
*/

   string QueuePropertyBase::getPropertyPrefix() const
   {
      return propertyPrefix_;
   }

   void QueuePropertyBase::setpropertyPrefix(const string& prefix)
   {
      propertyPrefix_ = prefix;
   }

   string QueuePropertyBase::getPrefix()
   {
      return (propertyPrefix_.length() > 0) ?
                   getRootTagName()+"/"+propertyPrefix_+"/" :
                   getRootTagName()+"/";
   }

   string QueuePropertyBase::getPropName(const string& token)
   {
      return "-" + getPrefix() + token;
   }

   string QueuePropertyBase::getRootTagName() const
   {
      return rootTagName_;
   }


}}}}} // namespaces