TAO/orbsvcs/tests/FT_App/ReplicationManagerFaultConsumerAdapter.cpp

   1 // -*- C++ -*-
   2 #include "ReplicationManagerFaultConsumerAdapter.h"
   3 #include "ace/Get_Opt.h"
   4 #include "orbsvcs/PortableGroup/PG_Properties_Encoder.h"
   5 #include "orbsvcs/FT_ReplicationManager/FT_DefaultFaultAnalyzer.h"
   6 // FUZZ: disable check_for_streams_include
   7 #include "ace/streams.h"
   8 #include "ace/OS_NS_stdio.h"
   9
  10 ReplicationManagerFaultConsumerAdapter::ReplicationManagerFaultConsumerAdapter()
  11   : orb_(CORBA::ORB::_nil())
  12   , quit_(0)
  13   , readyFile_(0)
  14   , detector_ior_(0)
  15   , factory_(FT::FaultDetectorFactory::_nil())
  16   , notifier_ior_(0)
  17   , notifier_(FT::FaultNotifier::_nil())
  18   , p_fault_consumer_(0)
  19   , consumer_servant_(0)
  20 {
  21 }
  22
  23
  24 ReplicationManagerFaultConsumerAdapter::~ReplicationManagerFaultConsumerAdapter()
  25 {
  26 }
  27
  28 size_t ReplicationManagerFaultConsumerAdapter::notifications () const
  29 {
  30   // Delegate to the FT_FaultConsumer.
  31   return this->p_fault_consumer_->notifications ();
  32 }
  33
  34
  35 int ReplicationManagerFaultConsumerAdapter::parse_args (int argc, ACE_TCHAR * argv[])
  36 {
  37   int optionError = 0;
  38   ACE_Get_Opt get_opts (argc, argv, ACE_TEXT("o:r:d:n:"));
  39   int c;
  40   while ((c = get_opts ()) != -1)
  41   {
  42     switch (c)
  43     {
  44       case 'r':
  45       {
  46         this->replica_iors_.push_back (ACE_TEXT_ALWAYS_CHAR(get_opts.opt_arg ()));
  47         break;
  48       }
  49       case 'd':
  50       {
  51         this->detector_ior_ = get_opts.opt_arg ();
  52         break;
  53       }
  54       case 'n':
  55       {
  56         this->notifier_ior_ = get_opts.opt_arg ();
  57         break;
  58       }
  59       case 'o':
  60       {
  61         this->readyFile_ = get_opts.opt_arg ();
  62         break;
  63       }
  64
  65       default:
  66         // fall thru
  67       case '?':
  68       {
  69         break;
  70       }
  71     }
  72   }
  73
  74   if(! optionError)
  75   {
  76     if (0 == this->replica_iors_.size())
  77     {
  78       ACE_ERROR ((LM_ERROR,
  79         "-r option is required.\n"
  80         ));
  81       optionError = -1;
  82     }
  83     if (0 == this->detector_ior_)
  84     {
  85       ACE_ERROR ((LM_ERROR,
  86         "-d option is required.\n"
  87         ));
  88       optionError = -1;
  89     }
  90   }
  91
  92   if(optionError)
  93   {
  94     ACE_ERROR ((LM_ERROR,
  95       "usage:  %s"
  96       " -r <replica.ior[ -r replica.ior]>"
  97       " -d <detector.ior>"
  98       " -o <this.ior>"
  99       " -n <nameService name>"
 100       "\n",
 101       argv [0]
 102       ));
 103   }
 104   return optionError;
 105 }
 106
 107 /**
 108  * Register this object.
 109  */
 110 int ReplicationManagerFaultConsumerAdapter::init (
 111   CORBA::ORB_ptr orb)
 112 {
 113   ACE_DEBUG ((
 114     LM_DEBUG,
 115     ACE_TEXT ("Entered ReplicationManagerFaultConsumerAdapter::init.\n")
 116   ));
 117
 118   int result = 0;
 119   this->orb_ = CORBA::ORB::_duplicate (orb);
 120
 121   //////////////////////////////////////////
 122   // resolve reference to detector factory
 123   ACE_DEBUG ((
 124     LM_DEBUG,
 125     ACE_TEXT ("ReplicationManagerFaultConsumerAdapter::init: ")
 126     ACE_TEXT ("Getting ready to read iorDetectorFile.\n")
 127   ));
 128
 129   CORBA::Object_var detector_obj = this->orb_->string_to_object (
 130     this->detector_ior_);
 131   this->factory_ = ::FT::FaultDetectorFactory::_narrow (
 132     detector_obj.in());
 133   if (CORBA::is_nil (this->factory_.in()))
 134   {
 135     ACE_ERROR_RETURN ((
 136       LM_ERROR,
 137       ACE_TEXT ("ReplicationManagerFaultConsumerAdapter::init: ")
 138       ACE_TEXT ("FaultDetectorFactory IOR is nil: %s\n"),
 139       this->detector_ior_),
 140       -1);
 141   }
 142
 143   //////////////////////////////////////////
 144   // resolve references to notifier
 145   ACE_DEBUG ((
 146     LM_DEBUG,
 147     ACE_TEXT ("ReplicationManagerFaultConsumerAdapter::init: ")
 148     ACE_TEXT ("Getting ready to read Notifier IOR file.\n")
 149   ));
 150
 151   CORBA::Object_var notifier_ior = this->orb_->string_to_object (
 152     this->notifier_ior_);
 153   this->notifier_ = ::FT::FaultNotifier::_narrow (
 154     notifier_ior.in());
 155   if (CORBA::is_nil (this->notifier_.in()))
 156   {
 157     ACE_ERROR_RETURN ((
 158       LM_ERROR,
 159       ACE_TEXT ("ReplicationManagerFaultConsumerAdapter::init: ")
 160       ACE_TEXT ("FaultNotifier IOR is nil: %s\n"),
 161       this->notifier_ior_),
 162       -1);
 163   }
 164
 165   // Create the real FaultConsumer.
 166   //
 167   // Note: We have to hang onto the servant class pointer so we can
 168   // invoke member functions on it, but we also give ownership of it
 169   // to a PortableServer::ServantBase_var.
 170   ACE_DEBUG ((
 171     LM_DEBUG,
 172     ACE_TEXT ("ReplicationManagerFaultConsumerAdapter::init: ")
 173     ACE_TEXT ("Getting ready to create the real FaultConsumer.\n")
 174   ));
 175
 176   ACE_NEW_RETURN (this->p_fault_consumer_, TAO::FT_FaultConsumer (), -1);
 177   if (this->p_fault_consumer_ != 0)
 178   {
 179     this->consumer_servant_ = this->p_fault_consumer_;
 180   }
 181
 182   //////////////////////////
 183   // Get ready to initialize the consumer.  We need to provide it
 184   // with the following:
 185   // - The POA in which it is to be activated.
 186   // - FT::FaultNotifier IOR.
 187   // - FT::ReplicationManager IOR (fake it for now).
 188
 189   // Get the RootPOA from the ORB.
 190   CORBA::Object_var poa_obj = this->orb_->resolve_initial_references (
 191     "RootPOA");
 192   PortableServer::POA_var poa = PortableServer::POA::_narrow (
 193     poa_obj.in());
 194
 195   // Create a fault analyzer.
 196   TAO::FT_FaultAnalyzer * analyzer = 0;
 197   ACE_NEW_RETURN (analyzer, TAO::FT_DefaultFaultAnalyzer (), -1);
 198
 199   // Initialize the FaultConsumer.
 200   // It will activate itself in the POA we pass it and connect to the
 201   // Fault Notifier we pass it.
 202   ACE_DEBUG ((
 203     LM_DEBUG,
 204     ACE_TEXT ("ReplicationManagerFaultConsumerAdapter::init: ")
 205     ACE_TEXT ("Getting ready to initialize the real FaultConsumer.\n")
 206   ));
 207
 208   result = this->p_fault_consumer_->init (
 209     poa.in(),
 210     this->notifier_.in(),
 211     analyzer);
 212   if (result != 0)
 213   {
 214     ACE_ERROR_RETURN ((
 215       LM_ERROR,
 216       ACE_TEXT ("ReplicationManagerFaultConsumerAdapter::init: ")
 217       ACE_TEXT ("Unable to initialize the real FaultConsumer.\n")),
 218       result);
 219     }
 220
 221   this->identity_ = "ReplicationManagerFaultConsumerAdapter";
 222
 223   // Activate the RootPOA.
 224   PortableServer::POAManager_var poa_manager =
 225     poa->the_POAManager ();
 226   poa_manager->activate ();
 227
 228   /////////////////////////
 229   // Set up fault detectors
 230   if (result == 0)
 231   {
 232     ////////////////////////////////////
 233     // resolve references to replicas
 234     // create a fault detector for each replica
 235     size_t replicaCount = this->replica_iors_.size();
 236     ACE_DEBUG ((LM_DEBUG,
 237       ACE_TEXT ("Number of replicas being monitored: (%u)\n"),
 238       static_cast<unsigned int> (replicaCount)
 239     ));
 240     for (size_t nRep = 0; result == 0 && nRep < replicaCount; ++nRep)
 241     {
 242       const char * iorName = this->replica_iors_[nRep];
 243       CORBA::Object_var replica_obj = this->orb_->string_to_object (
 244         iorName);
 245       FT::PullMonitorable_var replica = FT::PullMonitorable::_narrow (
 246         replica_obj.in());
 247       if (CORBA::is_nil(replica.in()))
 248       {
 249         ACE_ERROR_RETURN ((
 250           LM_ERROR,
 251           ACE_TEXT ("ReplicationManagerFaultConsumerAdapter::init: ")
 252           ACE_TEXT ("Can't resolve Replica IOR: %s\n"),
 253           iorName),
 254           -1);
 255       }
 256       else
 257       {
 258         this->replicas_.push_back(replica);
 259
 260         CORBA::String_var type_id = CORBA::string_dup("FaultDetector");
 261
 262         TAO_PG::Properties_Encoder encoder;
 263
 264         PortableGroup::Value value;
 265         value <<= notifier_.in ();
 266         encoder.add(::FT::FT_NOTIFIER, value);
 267
 268         value <<= replica.in ();
 269         encoder.add(::FT::FT_MONITORABLE, value);
 270
 271         FT::FTDomainId domain_id = 0;
 272         value <<= domain_id;
 273         encoder.add(::FT::FT_DOMAIN_ID, value);
 274
 275         PortableGroup::Location object_location;
 276         object_location.length(2);
 277         object_location[0].id = CORBA::string_dup("test");
 278         object_location[1].id = CORBA::string_dup("Location_A");
 279         value <<= object_location;
 280         encoder.add(::FT::FT_LOCATION, value);
 281
 282         PortableGroup::TypeId_var object_type = CORBA::string_dup (
 283           "IDL:org.omg/CosNaming/NamingContextExt:1.0");
 284         value <<= object_type.in ();
 285         encoder.add(::FT::FT_TYPE_ID, value);
 286
 287         PortableGroup::ObjectGroupId group_id =
 288           static_cast<PortableGroup::ObjectGroupId> (6191982);
 289         value <<= group_id;
 290         encoder.add(::FT::FT_GROUP_ID, value);
 291
 292         // allocate and populate the criteria
 293         PortableGroup::Criteria_var criteria;
 294         ACE_NEW_NORETURN (criteria,
 295           PortableGroup::Criteria);
 296         if (criteria.ptr() == 0)
 297         {
 298           ACE_ERROR_RETURN ((
 299             LM_ERROR,
 300             ACE_TEXT ("ReplicationManagerFaultConsumerAdapter::init: ")
 301             ACE_TEXT ("Error cannot allocate criteria.\n")),
 302             -1);
 303         }
 304         else
 305         {
 306           encoder.encode(criteria);
 307           PortableGroup::GenericFactory::FactoryCreationId_var factory_creation_id;
 308
 309           this->factory_->create_object (
 310             type_id.in(),
 311             criteria.in(),
 312             factory_creation_id);
 313         }
 314       }
 315     }
 316
 317     // Signal that we are ready to go.
 318     if (result == 0 && this->readyFile_ != 0)
 319     {
 320       FILE* ready = ACE_OS::fopen (this->readyFile_, "w");
 321       if (ready)
 322       {
 323         ACE_OS::fprintf (ready, "ready\n");
 324         ACE_OS::fclose (ready);
 325       }
 326     }
 327   }
 328
 329   return result;
 330 }
 331
 332 /**
 333  * Return a string to identify this object for logging/console message purposes.
 334  */
 335 const char * ReplicationManagerFaultConsumerAdapter::identity () const
 336 {
 337   return this->identity_.c_str();
 338 }
 339
 340 /**
 341  * Clean house for process shut down.
 342  */
 343 int ReplicationManagerFaultConsumerAdapter::fini ()
 344 {
 345   // Delegate to the FT_FaultConsumer.
 346   return this->p_fault_consumer_->fini ();
 347 }
 348
 349
 350 int ReplicationManagerFaultConsumerAdapter::idle(int & result)
 351 {
 352   ACE_UNUSED_ARG(result);
 353   int quit = 0;
 354
 355   if (this->replicas_.size() == this->p_fault_consumer_->notifications())
 356   {
 357     quit = 1;
 358   }
 359   return quit;
 360 }