addressbase.cpp

java开源的企业总线.xmlBlaster

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

/**
 * Abstract helper class holding connect address and callback address string
 * and protocol string.
 * <p />
 * See examples in the implementing classes
 * @see Address
 * @see CallbackAddress
 */

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

/**
 * A nice human readable name for this address (used for logging)
 */
string AddressBase::getName()
{
   return getHostname() + string(":") + lexical_cast<std::string>(getPort());
}

/**
 * Check if supplied address would connect to the address of this instance
 */
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;
}

/**
 * Show some important settings for logging
 */
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;
}

/**
 * @param type    The protocol type, e.g. "IOR", "SOCKET", "XMLRPC"
 */
void AddressBase::setType(const string& type)
{
   type_ = type;
}

/**
 * @param version   The protocol version, e.g. "1.0"
 */
void AddressBase::setVersion(const string& version)
{
   version_ = version;
}

/**
 * Updates the internal address as well. 
 * @param host An IP or DNS
 */
void AddressBase::setHostname(const string& host)
{
   initHostname(host);
   isHardcodedHostname_ = true;
}

/**
 * @return true if the bootstrapHostname is explicitly set by user with setHostname()
 * false if it is determined automatically
 */
bool AddressBase::isHardcodedHostname()
{
   return isHardcodedHostname_;
}

/**             
 * Check if a bootstrapHostname is set already
 */
bool AddressBase::hasHostname() {
   return hostname_ != "";
}

/**
 * @return The Hostname, IP or "" if not known
 */
string AddressBase::getHostname() const
{
   if (hostname_ == "") {
      hostname_ = global_.getBootstrapHostname();
      address_  = ""; // reset cache
   }
   return hostname_;
}

/**
 * Set the bootstrapping port. 
 * Updates the internal address as well. 
 */
void AddressBase::setPort(int port)
{
   port_    = port;
   address_ = ""; // reset cache
}

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

/**
 * Set the callback address, it should fit to the protocol-type.
 *
 * @param address The callback address, e.g. "socket://192.168.1.1:7607"
 */
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_;
}

/**
 * Returns the protocol type.
 * @return e.g. "SOCKET" or "IOR" (never null).
 */
string AddressBase::getType() const
{
   return type_;
}

/**
 * Returns the protocol version.
 * @return e.g. "1.0" or null
 */
string AddressBase::getVersion() const
{
   return version_;
}

/**
 * What to do if max retries is exhausted. 
 * <p />
 * This mode is currently not configurable, we always destroy the login session. 
 * This is interpreted only server side if callback fails.
 * @return Constants.ONEXHAUST_KILL_SESSION="killSession"
 */
string AddressBase::getOnExhaust() const
{
   return Constants::ONEXHAUST_KILL_SESSION; // in future possibly Constants.ONEXHAUST_KILL_CALLBACK
}

/**
 * Kill login session if max callback retries is exhausted?
 */
bool AddressBase::getOnExhaustKillSession() const
{
   return getOnExhaust() == Constants::ONEXHAUST_KILL_SESSION;
}

/**
 * BurstMode: The time span to collect messages before sending. 
 * @return The time to collect in milliseconds
 */
long AddressBase::getCollectTime() const
{
   return collectTime_;
}

/**
 * BurstMode: The time to collect messages for sending in a bulk. 
 * @param The time to collect in milliseconds
 */
void AddressBase::setCollectTime(long collectTime)
{
   if (collectTime < 0) collectTime_ = 0;
   else collectTime_ = collectTime;
}

int AddressBase::getBurstModeMaxEntries() const
{