InfoRepo Class Reference

#include <DCPSInfoRepoServ.h>

Inheritance diagram for InfoRepo:

[legend]Collaboration diagram for InfoRepo:

[legend]List of all members.

Public Member Functions
	InfoRepo (int argc, ACE_TCHAR *argv[])
	~InfoRepo ()
void	run ()
virtual void	shutdown ()
	ShutdownInterface used to schedule a shutdown.
void	sync_shutdown ()
virtual int	handle_exception (ACE_HANDLE fd=ACE_INVALID_HANDLE)
	Handler for the reactor to dispatch finalization activity to.
Private Member Functions
void	init ()
void	usage (const ACE_TCHAR *cmd)
void	parse_args (int argc, ACE_TCHAR *argv[])
void	finalize ()
	Actual finalization of service resources.
Private Attributes
CORBA::ORB_var	orb_
ACE_TString	ior_file_
std::string	listen_address_str_
int	listen_address_given_
bool	use_bits_
bool	resurrect_
ACE_Time_Value	reassociate_delay_
bool	finalized_
	Flag to indicate that finalization has already occurred.
bool	servant_finalized_
OpenDDS::Federator::ManagerImpl	federator_
	Repository Federation behaviors.
OpenDDS::Federator::Config	federatorConfig_
PortableServer::Servant_var< TAO_DDS_DCPSInfo_i >	info_servant_
ACE_Thread_Mutex	lock_
ACE_Condition_Thread_Mutex	cond_
bool	shutdown_complete_
ACE_Time_Value	dispatch_cleanup_delay_
Classes
struct	InitError

---

Detailed Description

Definition at line 26 of file DCPSInfoRepoServ.h.

---

Constructor & Destructor Documentation

InfoRepo::InfoRepo	(	int	argc,
		ACE_TCHAR *	argv[]
	)

Definition at line 42 of file DCPSInfoRepoServ.cpp.

References finalize(), and init().

: ior_file_(ACE_TEXT("repo.ior"))
, listen_address_given_(0)
#ifdef DDS_HAS_MINIMUM_BIT
, use_bits_(false)
#else
, use_bits_(true)
#endif
, resurrect_(true)
, finalized_(false)
, servant_finalized_(false)
, federator_(this->federatorConfig_)
, federatorConfig_(argc, argv)
, lock_()
, cond_(lock_)
, shutdown_complete_(false)
, dispatch_cleanup_delay_(30,0)
{
  try {
    this->init();
  } catch (...) {
    this->finalize();
    throw;
  }
}

InfoRepo::~InfoRepo

(

)

Definition at line 68 of file DCPSInfoRepoServ.cpp.

References finalize().

{
  this->finalize();
}

---

Member Function Documentation

void InfoRepo::finalize

(

)

[private]

Actual finalization of service resources.

Definition at line 85 of file DCPSInfoRepoServ.cpp.

References federator_, OpenDDS::Federator::ManagerImpl::finalize(), finalized_, info_servant_, orb_, servant_finalized_, and TheServiceParticipant.

Referenced by InfoRepo(), run(), and ~InfoRepo().

{
  if (this->finalized_) {
    return;
  }

  if (!this->servant_finalized_) {
    // reached if the ImR caused the ORB to shut down,
    // which bypasses InfoRepo::handle_exception()
    this->info_servant_->finalize();
    this->federator_.finalize();
    this->servant_finalized_ = true;
  }

  TheServiceParticipant->shutdown();

  if (!CORBA::is_nil(this->orb_)) {
    this->orb_->destroy();
  }

  this->finalized_ = true;
}

int InfoRepo::handle_exception

(

ACE_HANDLE

fd = ACE_INVALID_HANDLE

)

[virtual]

Handler for the reactor to dispatch finalization activity to.

Definition at line 109 of file DCPSInfoRepoServ.cpp.

References federator_, OpenDDS::Federator::ManagerImpl::finalize(), info_servant_, orb_, and servant_finalized_.

{
  // these should occur before ORB::shutdown() since they use the ORB/reactor
  this->info_servant_->finalize();
  this->federator_.finalize();
  this->servant_finalized_ = true;

  this->orb_->shutdown(true);
  return 0;
}

void InfoRepo::init

(

)

[private]

Definition at line 224 of file DCPSInfoRepoServ.cpp.

References OpenDDS::Federator::Config::argv(), OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::Discovery::DEFAULT_REPO, OpenDDS::Federator::Config::federateIor(), OpenDDS::Federator::Config::federationDomain(), federator_, OpenDDS::Federator::FEDERATOR_IORTABLE_KEY, federatorConfig_, OpenDDS::DCPS::Service_Participant::get_discovery(), OpenDDS::Federator::ManagerImpl::id(), OpenDDS::Federator::ManagerImpl::info(), info_servant_, ior_file_, OpenDDS::Federator::ManagerImpl::localRepo(), OpenDDS::Federator::ManagerImpl::orb(), orb_, TAO_DDS_DCPSFederationId::overridden(), parse_args(), OpenDDS::Federator::REPOSITORY_IORTABLE_KEY, OpenDDS::DCPS::Service_Participant::set_repo_ior(), TheParticipantFactoryWithArgs, and TheServiceParticipant.

Referenced by InfoRepo().

{
  ACE_Argv_Type_Converter cvt(this->federatorConfig_.argc(),
                              this->federatorConfig_.argv());
  this->orb_ = CORBA::ORB_init(cvt.get_argc(), cvt.get_ASCII_argv(), "");

  this->info_servant_ =
    new TAO_DDS_DCPSInfo_i(this->orb_, this->resurrect_, this,
                           this->federatorConfig_.federationId());

  // Install the DCPSInfo_i into the Federator::Manager.
  this->federator_.info() = this->info_servant_.in();

  CORBA::Object_var obj =
    this->orb_->resolve_initial_references("RootPOA");
  PortableServer::POA_var root_poa = PortableServer::POA::_narrow(obj);

  PortableServer::POAManager_var poa_manager = root_poa->the_POAManager();

  // Use persistent and user id POA policies so the Info Repo's
  // object references are consistent.
  CORBA::PolicyList policies(2);
  policies.length(2);
  policies[0] = root_poa->create_id_assignment_policy(PortableServer::USER_ID);
  policies[1] = root_poa->create_lifespan_policy(PortableServer::PERSISTENT);
  PortableServer::POA_var info_poa = root_poa->create_POA("InfoRepo",
                                                          poa_manager,
                                                          policies);

  // Creation of the new POAs over, so destroy the Policy_ptr's.
  for (CORBA::ULong i = 0; i < policies.length(); ++i) {
    policies[i]->destroy();
  }

  PortableServer::ObjectId_var oid =
    PortableServer::string_to_ObjectId("InfoRepo");
  info_poa->activate_object_with_id(oid, this->info_servant_.in());
  obj = info_poa->id_to_reference(oid);
  // the object is created locally, so it is safe to do an
  // _unchecked_narrow, this was needed to prevent an exception
  // when dealing with ImR-ified objects
  OpenDDS::DCPS::DCPSInfo_var info_repo =
    OpenDDS::DCPS::DCPSInfo::_unchecked_narrow(obj);

  CORBA::String_var objref_str =
    orb_->object_to_string(info_repo);

  // Initialize the DomainParticipantFactory
  DDS::DomainParticipantFactory_var dpf =
    TheParticipantFactoryWithArgs(cvt.get_argc(),
                                  cvt.get_TCHAR_argv());

  // We need parse the command line options for DCPSInfoRepo after parsing DCPS specific
  // command line options.

  // Check the non-ORB arguments.
  this->parse_args(cvt.get_argc(), cvt.get_TCHAR_argv());

  // Activate the POA manager before initialize built-in-topics
  // so invocations can be processed.
  poa_manager->activate();

  if (this->use_bits_) {
    if (this->info_servant_->init_transport(this->listen_address_given_,
        this->listen_address_str_.c_str())
        != 0 /* init_transport returns 0 for success */) {
      ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ERROR: DCPSInfoRepo::init: ")
                 ACE_TEXT("Unable to initialize transport.\n")));
      throw InitError("Unable to initialize transport.");
    }

  } else {
    TheServiceParticipant->set_BIT(false);
  }

  // This needs to be done after initialization since we create the reference
  // to ourselves in the service here.
  OpenDDS::DCPS::Service_Participant* serv_part = TheServiceParticipant;
  serv_part->set_repo_ior(objref_str, OpenDDS::DCPS::Discovery::DEFAULT_REPO);

  OpenDDS::DCPS::Discovery_rch disc = serv_part->get_discovery(0 /*domainId*/);
  OpenDDS::DCPS::InfoRepoDiscovery_rch ird =
    OpenDDS::DCPS::static_rchandle_cast<OpenDDS::DCPS::InfoRepoDiscovery>(disc);
  if (!ird->set_ORB(orb_)) {
    ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ERROR: DCPSInfoRepo::init: ")
               ACE_TEXT("Unable to set the ORB in InfoRepoDiscovery.\n")));
    throw InitError("Unable to set the ORB in InfoRepoDiscovery.");
  }

  // Initialize persistence _after_ initializing the participant factory
  // and intializing the transport.
  if (!this->info_servant_->init_persistence()) {
    ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ERROR: DCPSInfoRepo::init: ")
               ACE_TEXT("Unable to initialize persistence.\n")));
    throw InitError("Unable to initialize persistence.");
  }

  // Initialize reassociation.
  if (this->reassociate_delay_ != ACE_Time_Value::zero &&
     !this->info_servant_->init_reassociation(this->reassociate_delay_)) {
    ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ERROR: DCPSInfoRepo::init: ")
               ACE_TEXT("Unable to initialize reassociation.\n")));
    throw InitError("Unable to initialize reassociation.");
  }

  // Initialize dispatch checking
  if (this->dispatch_cleanup_delay_ != ACE_Time_Value::zero &&
     !this->info_servant_->init_dispatchChecking(this->dispatch_cleanup_delay_)) {
    ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ERROR: DCPSInfoRepo::init: ")
               ACE_TEXT("Unable to initialize Dispatch checking.\n")));
    throw InitError("Unable to initialize dispatch checking.");
  }

  // Fire up the federator.
  OpenDDS::Federator::Manager_var federator;
  CORBA::String_var               federator_ior;

  if (federator_.id().overridden()) {
    oid = PortableServer::string_to_ObjectId("Federator");
    info_poa->activate_object_with_id(oid, &federator_);
    obj = info_poa->id_to_reference(oid);
    federator = OpenDDS::Federator::Manager::_narrow(obj);

    federator_ior = orb_->object_to_string(federator);

    // Add a local repository reference that can be returned via a
    // remote call to a peer.
    this->federator_.localRepo(info_repo);

    // It should be safe to initialize the federation mechanism at this
    // point.  What we really needed to wait for is the initialization of
    // the service components - like the DomainParticipantFactory and the
    // repository bindings.
    // N.B. This is done *before* being added to the IOR table to avoid any
    //      races with an eager client.
    this->federator_.orb(this->orb_);

    //
    // Add the federator to the info_servant update manager as an
    // additional updater interface to be called.
    // N.B. This needs to be done *after* the call to load_domains()
    //      since that is where the update manager is initialized in the
    //      info startup sequencing.
    this->info_servant_->add(&this->federator_);
  }

  // Grab the IOR table.
  CORBA::Object_var table_object =
    this->orb_->resolve_initial_references("IORTable");

  IORTable::Table_var adapter = IORTable::Table::_narrow(table_object);

  if (CORBA::is_nil(adapter)) {
    ACE_ERROR((LM_ERROR, ACE_TEXT("Nil IORTable\n")));

  } else {
    adapter->bind(OpenDDS::Federator::REPOSITORY_IORTABLE_KEY, objref_str);

    if (this->federator_.id().overridden()) {
      // Bind to '/Federator'
      adapter->bind(OpenDDS::Federator::FEDERATOR_IORTABLE_KEY, federator_ior);

      // Bind to '/Federator/1382379631'
      std::stringstream buffer(OpenDDS::Federator::FEDERATOR_IORTABLE_KEY);
      buffer << "/" << std::dec << this->federatorConfig_.federationDomain();
      adapter->bind(buffer.str().c_str(), federator_ior);
    }
  }

  FILE* output_file = ACE_OS::fopen(this->ior_file_.c_str(), ACE_TEXT("w"));

  if (output_file == 0) {
    ACE_ERROR((LM_ERROR, ACE_TEXT("ERROR: Unable to open IOR file: %s\n"),
               ior_file_.c_str()));
    throw InitError("Unable to open IOR file.");
  }

  ACE_OS::fprintf(output_file, "%s", objref_str.in());
  ACE_OS::fclose(output_file);

  // Initial federation join if specified on command line.
  if (this->federator_.id().overridden()
       && !this->federatorConfig_.federateIor().empty()) {
    if (OpenDDS::DCPS::DCPS_debug_level > 0) {
      ACE_DEBUG((LM_DEBUG,
                 ACE_TEXT("(%P|%t) INFO: DCPSInfoRepo::init() - ")
                 ACE_TEXT("joining federation with repository %s\n"),
                 this->federatorConfig_.federateIor().c_str()));
    }

    obj = this->orb_->string_to_object(
          this->federatorConfig_.federateIor().c_str());

    if (CORBA::is_nil(obj)) {
      ACE_ERROR((LM_ERROR,
                 ACE_TEXT("(%P|%t) ERROR: could not resolve %s for initial federation.\n"),
                 this->federatorConfig_.federateIor().c_str()));
      throw InitError("Unable to resolve IOR for initial federation.");
    }

    OpenDDS::Federator::Manager_var peer =
      OpenDDS::Federator::Manager::_narrow(obj);

    if (CORBA::is_nil(peer)) {
      ACE_ERROR((LM_ERROR,
                 ACE_TEXT("(%P|%t) ERROR: could not narrow %s.\n"),
                 this->federatorConfig_.federateIor().c_str()));
      throw InitError("Unable to narrow peer for initial federation.");
    }

    // Actually join.
    peer->join_federation(federator,
                          this->federatorConfig_.federationDomain());
  }
}

void InfoRepo::parse_args	(	int	argc,
		ACE_TCHAR *	argv[]
	)			[private]

Definition at line 162 of file DCPSInfoRepoServ.cpp.

References dispatch_cleanup_delay_, ior_file_, listen_address_given_, listen_address_str_, reassociate_delay_, resurrect_, and TURN_ON_VERBOSE_DEBUG.

Referenced by init().

{
  ACE_Arg_Shifter arg_shifter(argc, argv);

  const ACE_TCHAR* current_arg = 0;

  while (arg_shifter.is_anything_left()) {
    if ((current_arg = arg_shifter.get_the_parameter(ACE_TEXT("-a"))) != 0) {
      this->listen_address_str_ = ACE_TEXT_ALWAYS_CHAR(current_arg);
      this->listen_address_given_ = 1;
      arg_shifter.consume_arg();
    // Must check for -ReassociateDelay before -r
    } else if ((current_arg = arg_shifter.get_the_parameter(ACE_TEXT("-ReassociateDelay"))) != 0) {
      long msec = ACE_OS::atoi(current_arg);
      this->reassociate_delay_.msec(msec);

      arg_shifter.consume_arg();
    } else if ((current_arg = arg_shifter.get_the_parameter
                              (ACE_TEXT("-r"))) != 0) {
      int p = ACE_OS::atoi(current_arg);
      this->resurrect_ = true;

      if (p == 0) {
        this->resurrect_ = false;
      }

      arg_shifter.consume_arg();

    } else if ((current_arg = arg_shifter.get_the_parameter(ACE_TEXT("-o"))) != 0) {
      this->ior_file_ = current_arg;
      arg_shifter.consume_arg();

    } else if (arg_shifter.cur_arg_strncasecmp(ACE_TEXT("-NOBITS")) == 0) {
      this->use_bits_ = false;
      arg_shifter.consume_arg();

    } else if (arg_shifter.cur_arg_strncasecmp(ACE_TEXT("-z")) == 0) {
      TURN_ON_VERBOSE_DEBUG;
      arg_shifter.consume_arg();

    } else if ((current_arg = arg_shifter.get_the_parameter(ACE_TEXT("-DispatchingCheckDelay"))) != 0) {
      long sec = ACE_OS::atoi(current_arg);
      this->dispatch_cleanup_delay_.sec(sec);
      arg_shifter.consume_arg();

    }

    // The '-?' option
    else if (arg_shifter.cur_arg_strncasecmp(ACE_TEXT("-?")) == 0) {
      this->usage(argv[0]);
      throw InitError("Usage");
    }

    // Anything else we just skip

    else {
      arg_shifter.ignore_arg();
    }
  }
}

void InfoRepo::run

(

)

Definition at line 74 of file DCPSInfoRepoServ.cpp.

References cond_, finalize(), orb_, and shutdown_complete_.

Referenced by ACE_TMAIN().

{
  this->shutdown_complete_ = false;
  this->orb_->run();
  this->finalize();
  ACE_GUARD(ACE_Thread_Mutex, g, this->lock_);
  this->shutdown_complete_ = true;
  this->cond_.signal();
}

void InfoRepo::shutdown

(

)

[virtual]

ShutdownInterface used to schedule a shutdown.

Implements ShutdownInterface.

Definition at line 121 of file DCPSInfoRepoServ.cpp.

References orb_.

Referenced by InfoRepo_Shutdown::operator()(), and sync_shutdown().

{
  this->orb_->orb_core()->reactor()->notify(this);
  // reactor will invoke our InfoRepo::handle_exception()
}

void InfoRepo::sync_shutdown

(

)

shutdown() and wait for it to complete: cannot be called from the reactor thread.

Definition at line 128 of file DCPSInfoRepoServ.cpp.

References cond_, and shutdown().

{
  this->shutdown();
  ACE_GUARD(ACE_Thread_Mutex, g, this->lock_);

  while (!this->shutdown_complete_) {
    this->cond_.wait();
  }
}

void InfoRepo::usage

(

const ACE_TCHAR *

cmd

)

[private]

Definition at line 139 of file DCPSInfoRepoServ.cpp.

{
  // NOTE: The federation arguments are parsed early by the
  //       FederationConfig object.
  ACE_DEBUG((LM_INFO,
             ACE_TEXT("Usage:\n")
             ACE_TEXT("  %s\n")
             ACE_TEXT("    -a <address> listening address for Built-In Topics\n")
             ACE_TEXT("    -o <file> write ior to file\n")
             ACE_TEXT("    -NOBITS disable the Built-In Topics\n")
             ACE_TEXT("    -z turn on verbose Transport logging\n")
             ACE_TEXT("    -r Resurrect from persistent file\n")
             ACE_TEXT("    -FederatorConfig <file> configure federation from <file>\n")
             ACE_TEXT("    -FederationId <number> value for this repository\n")
             ACE_TEXT("    -FederateWith <ior> federate initially with object at <ior>\n")
             ACE_TEXT("    -ReassociateDelay <msec> delay between reassociations\n")
             ACE_TEXT("    -DispatchingCheckDelay <sec> delay between checks for cleaning up dispatching connections.\n")
             ACE_TEXT("    -?\n")
             ACE_TEXT("\n"),
             cmd));
}

---

Member Data Documentation

ACE_Condition_Thread_Mutex InfoRepo::cond_ [private]

Definition at line 77 of file DCPSInfoRepoServ.h.

Referenced by run(), and sync_shutdown().

ACE_Time_Value InfoRepo::dispatch_cleanup_delay_ [private]

Definition at line 80 of file DCPSInfoRepoServ.h.

Referenced by parse_args().

OpenDDS::Federator::ManagerImpl InfoRepo::federator_ [private]

Repository Federation behaviors.

Definition at line 71 of file DCPSInfoRepoServ.h.

Referenced by finalize(), handle_exception(), and init().

OpenDDS::Federator::Config InfoRepo::federatorConfig_ [private]

Definition at line 72 of file DCPSInfoRepoServ.h.

Referenced by init().

bool InfoRepo::finalized_ [private]

Flag to indicate that finalization has already occurred.

Definition at line 67 of file DCPSInfoRepoServ.h.

Referenced by finalize().

PortableServer::Servant_var<TAO_DDS_DCPSInfo_i> InfoRepo::info_servant_ [private]

Definition at line 74 of file DCPSInfoRepoServ.h.

Referenced by finalize(), handle_exception(), and init().

ACE_TString InfoRepo::ior_file_ [private]

Definition at line 59 of file DCPSInfoRepoServ.h.

Referenced by init(), and parse_args().

int InfoRepo::listen_address_given_ [private]

Definition at line 61 of file DCPSInfoRepoServ.h.

Referenced by parse_args().

std::string InfoRepo::listen_address_str_ [private]

Definition at line 60 of file DCPSInfoRepoServ.h.

Referenced by parse_args().

ACE_Thread_Mutex InfoRepo::lock_ [private]

Definition at line 76 of file DCPSInfoRepoServ.h.

CORBA::ORB_var InfoRepo::orb_ [private]

Definition at line 57 of file DCPSInfoRepoServ.h.

Referenced by finalize(), handle_exception(), init(), run(), and shutdown().

ACE_Time_Value InfoRepo::reassociate_delay_ [private]

Definition at line 64 of file DCPSInfoRepoServ.h.

Referenced by parse_args().

bool InfoRepo::resurrect_ [private]

Definition at line 63 of file DCPSInfoRepoServ.h.

Referenced by parse_args().

bool InfoRepo::servant_finalized_ [private]

Definition at line 68 of file DCPSInfoRepoServ.h.

Referenced by finalize(), and handle_exception().

bool InfoRepo::shutdown_complete_ [private]

Definition at line 78 of file DCPSInfoRepoServ.h.

Referenced by run().

bool InfoRepo::use_bits_ [private]

Definition at line 62 of file DCPSInfoRepoServ.h.

---

The documentation for this class was generated from the following files:

• DCPSInfoRepoServ.h
• DCPSInfoRepoServ.cpp

---