TAO/orbsvcs/performance-tests/RTEvent/lib/Loopback_Consumer.cpp

   1 /**
   2  * @file Loopback_Consumer.cpp
   3  *
   4  * @author Carlos O'Ryan <coryan@uci.edu>
   5  */
   6
   7 #include "Loopback_Consumer.h"
   8 #include "Implicit_Deactivator.h"
   9 #include "orbsvcs/Event_Service_Constants.h"
  10
  11 Loopback_Consumer::
  12 Loopback_Consumer (CORBA::Long experiment_id,
  13                    CORBA::Long event_type,
  14                    Loopback_Supplier *supplier,
  15                    PortableServer::POA_ptr poa)
  16   : experiment_id_ (experiment_id)
  17   , event_type_ (event_type)
  18   , supplier_ (PortableServer::Servant_var<Loopback_Supplier>::_duplicate (supplier))
  19   , default_POA_ (PortableServer::POA::_duplicate (poa))
  20 {
  21 }
  22
  23 void
  24 Loopback_Consumer::connect (RtecEventChannelAdmin::EventChannel_ptr ec)
  25 {
  26   RtecEventChannelAdmin::ConsumerAdmin_var consumer_admin =
  27     ec->for_consumers ();
  28
  29   {
  30     ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->mutex_);
  31     if (!CORBA::is_nil (this->proxy_supplier_.in ()))
  32       return;
  33
  34     this->proxy_supplier_ =
  35       consumer_admin->obtain_push_supplier ();
  36   }
  37
  38   RtecEventComm::PushConsumer_var consumer =
  39     this->_this ();
  40
  41   RtecEventChannelAdmin::ConsumerQOS consumer_qos;
  42   consumer_qos.is_gateway = 0;
  43   consumer_qos.dependencies.length (2);
  44   RtecEventComm::EventHeader& h0 =
  45     consumer_qos.dependencies[0].event.header;
  46   h0.type   = ACE_ES_DISJUNCTION_DESIGNATOR;
  47   h0.source = 1;
  48
  49   RtecEventComm::EventHeader& h1 =
  50     consumer_qos.dependencies[1].event.header;
  51   h1.source = this->experiment_id_;
  52   h1.type   = this->event_type_;
  53
  54   this->proxy_supplier_->connect_push_consumer (consumer.in (),
  55                                                 consumer_qos);
  56 }
  57
  58 void
  59 Loopback_Consumer::disconnect (void)
  60 {
  61   RtecEventChannelAdmin::ProxyPushSupplier_var proxy;
  62   {
  63     ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->mutex_);
  64     if (CORBA::is_nil (this->proxy_supplier_.in ()))
  65       return;
  66     proxy = this->proxy_supplier_._retn ();
  67   }
  68
  69   Implicit_Deactivator deactivator (this);
  70
  71   try
  72     {
  73       proxy->disconnect_push_supplier ();
  74     }
  75   catch (const CORBA::Exception&){}
  76 }
  77
  78 void
  79 Loopback_Consumer::push (const RtecEventComm::EventSet &events)
  80 {
  81   // ACE_DEBUG ((LM_DEBUG, "Loopback_Consumer::push (%P|%t)\n"));
  82   this->supplier_->push (events);
  83 }
  84
  85 void
  86 Loopback_Consumer::disconnect_push_consumer (void)
  87 {
  88   ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->mutex_);
  89   this->proxy_supplier_ =
  90     RtecEventChannelAdmin::ProxyPushSupplier::_nil ();
  91 }
  92
  93 PortableServer::POA_ptr
  94 Loopback_Consumer::_default_POA (void)
  95 {
  96   return PortableServer::POA::_duplicate (this->default_POA_.in ());
  97 }