util/qos/address/AddressBase.cpp

Go to the documentation of this file.

/*------------------------------------------------------------------------------
Name:      AddressBase.cpp
Project:   xmlBlaster.org
Copyright: xmlBlaster.org, see xmlBlaster-LICENSE file
Comment:   Holding connect address and callback address string including protocol
Version:   $Id: AddressBase.cpp 14417 2005-12-22 12:13:15Z ruff $
------------------------------------------------------------------------------*/

#include <util/qos/address/AddressBase.h>
#include <util/lexical_cast.h>
#include <util/Global.h>

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

namespace org { namespace xmlBlaster { namespace util { namespace qos { namespace address {

const int    DEFAULT_port               = 3412;
const string DEFAULT_type               = Global::getDefaultProtocol(); //"SOCKET";
const string DEFAULT_version            = "1.0";
const long   DEFAULT_collectTime        = 0;
const int    DEFAULT_burstModeMaxEntries= 1;
const long   DEFAULT_burstModeMaxBytes  = -1L;
const bool   DEFAULT_oneway             = false;
const bool   DEFAULT_dispatcherActive   = true;
const string DEFAULT_compressType       = "";
const long   DEFAULT_minSize            = 0L;
const bool   DEFAULT_ptpAllowed         = true;
const string DEFAULT_sessionId          = "unknown";
const bool   DEFAULT_useForSubjectQueue = true;
string DEFAULT_dispatchPlugin     = "";
string ATTRIBUTE_TAG = "attribute";


AddressBase::AddressBase(Global& global, const string& rootTag)
   : ReferenceCounterBase(), global_(global), log_(global.getLog("org.xmlBlaster.util.qos")),
      attributes_()
{

   defaultPingInterval_ = 0L;
   defaultRetries_      = 0;
   defaultDelay_        = 0L;

   // set the defaults here ...
   ME                   = "AddressBase";
   nodeId_              = "";
   maxEntries_          = 0;
   address_             = "";
   hostname_            = "";
   isHardcodedHostname_ = false;
   type_                = DEFAULT_type;
   port_                = DEFAULT_port;
   version_             = DEFAULT_version;
   collectTime_         = DEFAULT_collectTime;
   burstModeMaxEntries_ = DEFAULT_burstModeMaxEntries;
   burstModeMaxBytes_   = DEFAULT_burstModeMaxBytes;
   pingInterval_        = defaultPingInterval_;
   retries_             = defaultRetries_;
   delay_               = defaultDelay_;
   oneway_              = DEFAULT_oneway;
   dispatcherActive_    = DEFAULT_dispatcherActive;
   compressType_        = DEFAULT_compressType;
   minSize_             = DEFAULT_minSize;
   ptpAllowed_          = DEFAULT_ptpAllowed;
   sessionId_           = DEFAULT_sessionId;
   useForSubjectQueue_  = DEFAULT_useForSubjectQueue;
   dispatchPlugin_      = DEFAULT_dispatchPlugin;
   setRootTag(rootTag);
}

AddressBase::AddressBase(const AddressBase& addr)
   : global_(addr.global_), log_(addr.log_),
      attributes_()
{
   copy(addr);
}

AddressBase& AddressBase::operator =(const AddressBase& addr)
{
   copy(addr);
   return *this;
}


AddressBase::~AddressBase()
{
}

void AddressBase::copy(const AddressBase& addr)
{
   port_                = addr.port_;
   ME                   = addr.ME;
   rootTag_             = addr.rootTag_;
   address_             = addr.address_;
   hostname_            = addr.hostname_;
   isHardcodedHostname_ = addr.isHardcodedHostname_;
   type_                = addr.type_;
   version_             = addr.version_;
   collectTime_         = addr.collectTime_;
   pingInterval_        = addr.pingInterval_;
   retries_             = addr.retries_;
   delay_               = addr.delay_;
   oneway_              = addr.oneway_;
   dispatcherActive_    = addr.dispatcherActive_;
   compressType_        = addr.compressType_;
   minSize_             = addr.minSize_;
   ptpAllowed_          = addr.ptpAllowed_;
   sessionId_           = addr.sessionId_;
   useForSubjectQueue_  = addr.useForSubjectQueue_;
   dispatchPlugin_      = addr.dispatchPlugin_;
   nodeId_              = addr.nodeId_;
   maxEntries_          = addr.maxEntries_;
   defaultPingInterval_ = addr.defaultPingInterval_;
   defaultRetries_      = addr.defaultRetries_;
   defaultDelay_        = addr.defaultDelay_;
   attributes_          = addr.attributes_;
}

string AddressBase::getName()
{
   return getHostname() + string(":") + lexical_cast<std::string>(getPort());
}

bool AddressBase::isSameAddress(AddressBase& other)
{
   string oa = other.getRawAddress();
   if ( (oa!="") && (oa == getRawAddress()) ) return true;
   string oh = other.getHostname();
   int op = other.getPort();
   if ( (op>0) && (op==getPort()) && (oh!="") && (oh==getHostname()))
      return true;
   return false;
}

string AddressBase::getSettings() const
{
   string ret = string("type=") + type_ + string(" oneway=") + lexical_cast<std::string>(oneway_) + string(" burstMode.collectTime=") + lexical_cast<std::string>(getCollectTime());
   return ret;
}

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

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

void AddressBase::setHostname(const string& host)
{
   initHostname(host);
   isHardcodedHostname_ = true;
}

bool AddressBase::isHardcodedHostname()
{
   return isHardcodedHostname_;
}

bool AddressBase::hasHostname() {
   return hostname_ != "";
}

string AddressBase::getHostname() const
{
   if (hostname_ == "") {
      hostname_ = global_.getBootstrapHostname();
      address_  = ""; // reset cache
   }
   return hostname_;
}

void AddressBase::setPort(int port)
{
   port_    = port;
   address_ = ""; // reset cache
}

int AddressBase::getPort() const
{
   return port_;
}

void AddressBase::setAddress(const string& address)
{
   address_ = address;
}

// TODO: Protocol abstraction from plugin
string AddressBase::getRawAddress() const
{
   if (address_ == "") {
      string schema = (getType() == Constants::SOCKET) ? "socket" : "http";
      address_ = schema + "://" + getHostname();  // socket://192.168.1.1:7607
      if (getPort() > 0)
         address_ += ":" + lexical_cast<std::string>(getPort());
   }
   return address_;
}

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

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

string AddressBase::getOnExhaust() const
{
   return Constants::ONEXHAUST_KILL_SESSION; // in future possibly Constants.ONEXHAUST_KILL_CALLBACK
}

bool AddressBase::getOnExhaustKillSession() const
{
   return getOnExhaust() == Constants::ONEXHAUST_KILL_SESSION;
}

long AddressBase::getCollectTime() const
{
   return collectTime_;
}

void AddressBase::setCollectTime(long collectTime)
{
   if (collectTime < 0) collectTime_ = 0;
   else collectTime_ = collectTime;
}

int AddressBase::getBurstModeMaxEntries() const
{
   return burstModeMaxEntries_;
}

void AddressBase::setBurstModeMaxEntries(int burstModeMaxEntries)
{
   if (burstModeMaxEntries < -1) {
      burstModeMaxEntries_ = 1;
   }
   else if (burstModeMaxEntries == 0) {
      log_.warn(ME, string("<burstMode maxEntries='") + lexical_cast<std::string>(burstModeMaxEntries) + string("'> is not supported and may cause strange behavior"));
      burstModeMaxEntries_ = burstModeMaxEntries;
   }
   else {
      burstModeMaxEntries_ = burstModeMaxEntries;
   }
}

long AddressBase::getBurstModeMaxBytes() const
{
   return burstModeMaxBytes_;
}

void AddressBase::setBurstModeMaxBytes(long burstModeMaxBytes)
{
   if (burstModeMaxBytes < -1) burstModeMaxBytes_ = -1;
   else burstModeMaxBytes_ = burstModeMaxBytes;
}

long AddressBase::getPingInterval() const
{
   return pingInterval_;
}

void AddressBase::setPingInterval(long pingInterval)
{
   if (pingInterval <= 0) pingInterval_ = 0;
   else if (pingInterval < 10) {
      log_.warn(ME, string("pingInterval=") + lexical_cast<std::string>(pingInterval) + string(" msec is too short, setting it to 10 millis"));
      pingInterval_ = 10;
   }
   else pingInterval_ = pingInterval;
}

int AddressBase::getRetries() const
{
   return retries_;
}

void AddressBase::setRetries(int retries)
{
   if (retries < -1) retries_ = -1;
   else retries_ = retries;
}

long AddressBase::getDelay() const
{
   return delay_;
}

void AddressBase::setDelay(long delay)
{
   if (delay < 0) delay_ = 0;
   else delay_ = delay;
}

bool AddressBase::oneway() const
{
   return oneway_;
}

void AddressBase::setOneway(bool oneway)
{
   oneway_ = oneway;
}

bool AddressBase::isDispatcherActive() const
{
   return dispatcherActive_;
}

void AddressBase::setDispatcherActive(bool dispatcherActive)
{
   dispatcherActive_ = dispatcherActive;
}

void AddressBase::setPtpAllowed(bool ptpAllowed)
{
   ptpAllowed_ = ptpAllowed;
}

bool AddressBase::isPtpAllowed()
{
   return ptpAllowed_;
}

void AddressBase::setCompressType(const string& compressType)
{
   compressType_ = compressType;

   // TODO !!!
   if (compressType != "")
      log_.warn(ME, "Compression of messages is not yet supported");
}

string AddressBase::getSecretSessionId() const
{
   return sessionId_;
}

void AddressBase::setSecretSessionId(const string& sessionId)
{
   sessionId_ = sessionId;
}

string AddressBase::getCompressType() const
{
   return compressType_;
}

long AddressBase::getMinSize() const
{
   return minSize_;
}

void AddressBase::setMinSize(long minSize)
{
   minSize_ = minSize;
}

void AddressBase::setDispatchPlugin(const string& dispatchPlugin)
{
   dispatchPlugin_ = dispatchPlugin;
}

string AddressBase::getDispatchPlugin() const
{
   return dispatchPlugin_;
}

void AddressBase::addAttribute(const ClientProperty& attribute)
{
   attributes_.insert(ClientPropertyMap::value_type(attribute.getName(), attribute));   
}

const AddressBase::ClientPropertyMap& AddressBase::getAttributes() const
{
   return attributes_;
}

string AddressBase::dumpAttributes(const string& extraOffset, bool clearText) const
{
   string ret = "";
   QosData::ClientPropertyMap::const_iterator iter = attributes_.begin();
   while (iter != attributes_.end()) {
      const ClientProperty& cp = (*iter).second;
      ret += cp.toXml(extraOffset, clearText, ATTRIBUTE_TAG);
      iter++;
   }
   return ret;
}


string AddressBase::toXml(const string& extraOffset) const
{
   //if (log_.call()) log_.call(ME, "::toXml");
   string ret;
   string offset = Constants::OFFSET + extraOffset;
   string offset2 = offset + Constants::INDENT;

   // <address type='SOCKET' bootstrapHostname='127.0.0.1' dispatchPlugin='undef'> ...
   ret += offset + string("<") + rootTag_  + string(" type='") + getType() + string("'");
   if ( (getVersion()!="") && (getVersion()!=DEFAULT_version))
      ret += string(" version='") + getVersion() + string("'");
   if (getHostname() != "")
      ret += string(" bootstrapHostname='") + getHostname() + string("'");
   if (DEFAULT_port != getPort())
       ret += string(" bootstrapPort='") + lexical_cast<std::string>(getPort()) + string("'");
   if (DEFAULT_sessionId != getSecretSessionId())
       ret += string(" sessionId='") + getSecretSessionId() + string("'");
   if (defaultPingInterval_ != getPingInterval())
       ret += string(" pingInterval='") + lexical_cast<std::string>(getPingInterval()) + string("'");
   if (defaultRetries_ != getRetries())
       ret += string(" retries='") + lexical_cast<std::string>(getRetries()) + string("'");
   if (defaultDelay_ != getDelay())
       ret += string(" delay='") + lexical_cast<std::string>(getDelay()) + string("'");
   if (DEFAULT_oneway != oneway()) {
       ret += string(" oneway='") + lexical_cast<std::string>(oneway()) + string("'");
   }
   if (DEFAULT_dispatcherActive != isDispatcherActive()) {
       ret += string(" dispatcherActive='") + lexical_cast<std::string>(isDispatcherActive()) + string("'");
   }
   if (DEFAULT_useForSubjectQueue != useForSubjectQueue_) {
       ret += string(" useForSubjectQueue='") + lexical_cast<std::string>(useForSubjectQueue_) + string("'");
   }
   if (DEFAULT_dispatchPlugin != dispatchPlugin_)
       ret += string(" dispatchPlugin='") + dispatchPlugin_ + string("'");
   ret += string(">");
   if (getRawAddress() != "")
      ret += offset + string("   ") + getRawAddress();
   if (getCollectTime() != DEFAULT_collectTime || getBurstModeMaxEntries() != DEFAULT_burstModeMaxEntries || getBurstModeMaxBytes() != DEFAULT_burstModeMaxBytes) {
      ret += offset2 + string("<burstMode");
      if (getCollectTime() != DEFAULT_collectTime)
         ret += string(" collectTime='") + lexical_cast<std::string>(getCollectTime()) + string("'");
      if (getBurstModeMaxEntries() != DEFAULT_burstModeMaxEntries)
         ret += string(" maxEntries='") + lexical_cast<std::string>(getBurstModeMaxEntries()) + string("'");
      if (getBurstModeMaxBytes() != DEFAULT_burstModeMaxBytes)
         ret += string(" maxBytes='") + lexical_cast<std::string>(getBurstModeMaxBytes()) + string("'");
      ret += string("/>");
   }
   if (getCompressType() != DEFAULT_compressType)
      ret += offset2 + string("<compress type='") + getCompressType() + string("' minSize='") + lexical_cast<std::string>(getMinSize()) + string("'/>");
   if (ptpAllowed_ != DEFAULT_ptpAllowed) {
      ret += offset2 + string("<ptp>") + lexical_cast<std::string>(ptpAllowed_) + string("</ptp>");
   }
   ret += dumpAttributes(offset2);
   ret += offset + string("</") + rootTag_ + string(">");
   return ret;
}

}}}}} // namespaces