TAO/orbsvcs/examples/Notify/ThreadPool/Supplier.cpp

   1 #include "Supplier.h"
   2
   3
   4 #include "tao/ORB_Core.h"
   5
   6 TAO_Notify_ThreadPool_Supplier::TAO_Notify_ThreadPool_Supplier  (TAO_Notify_ORB_Objects& orb_objects)
   7   : orb_objects_ (orb_objects)
   8     , proxy_consumer_id_ (0)
   9     , expected_consumer_count_ (2)
  10     , consumers_connected_ (lock_)
  11     , consumer_count_ (0)
  12     , max_events_ (10)
  13     , proxy_consumer_thread_count_ (0)
  14 {
  15 }
  16
  17 TAO_Notify_ThreadPool_Supplier::~TAO_Notify_ThreadPool_Supplier ()
  18 {
  19 }
  20
  21 void
  22 TAO_Notify_ThreadPool_Supplier::init (CosNotifyChannelAdmin::SupplierAdmin_var& admin, int expected_consumer_count ,int max_events,
  23                        int proxy_consumer_thread_count)
  24 {
  25   // First initialize the class members.
  26   this->admin_ = admin;
  27   this->expected_consumer_count_ = expected_consumer_count;
  28   this->max_events_ = max_events;
  29   this->proxy_consumer_thread_count_ = proxy_consumer_thread_count;
  30
  31   this->connect ();
  32 }
  33
  34 void
  35 TAO_Notify_ThreadPool_Supplier::run ()
  36 {
  37   {
  38     ACE_GUARD (TAO_SYNCH_MUTEX, mon, this->lock_);
  39
  40     ACE_DEBUG ((LM_DEBUG, "(%P, %t) Waiting for %d consumers to connect...\n", this->expected_consumer_count_-1));
  41
  42     // Wait till the consumers are ready to go.
  43     while (this->consumer_count_ != this->expected_consumer_count_)
  44       this->consumers_connected_.wait ();
  45   }
  46
  47   ACE_DEBUG ((LM_DEBUG,
  48               "(%P, %t) Supplier is sending an events...\n"));
  49
  50   // Send events to each consumer.
  51   for (int i = 0; i < this->max_events_; ++i)
  52     {
  53       for (int j = 0; j < this->expected_consumer_count_; ++j)
  54         {
  55           // send the event
  56           this->send_event (this->event_[j]);
  57         }
  58     }
  59
  60   // Disconnect from the EC
  61   this->disconnect ();
  62
  63   // Deactivate this object.
  64   this->deactivate ();
  65
  66   // we're done. shutdown the ORB to exit the process.
  67   this->orb_objects_.orb_->shutdown (true);
  68 }
  69
  70 void
  71 TAO_Notify_ThreadPool_Supplier::connect ()
  72 {
  73   // Activate the supplier object.
  74   CosNotifyComm::StructuredPushSupplier_var objref = this->_this ();
  75
  76   CosNotifyChannelAdmin::ProxyConsumer_var proxyconsumer;
  77
  78   if (this->proxy_consumer_thread_count_ != 0)
  79     {
  80       // Narrow to the extended interface.
  81       NotifyExt::SupplierAdmin_var admin_ext = NotifyExt::SupplierAdmin::_narrow (this->admin_.in ());
  82
  83       NotifyExt::ThreadPoolParams tp_params = { NotifyExt::CLIENT_PROPAGATED, 0,
  84                                                 0, static_cast<CORBA::ULong> (this->proxy_consumer_thread_count_),
  85                                                 0, 0, 0, 0, 0 };
  86
  87       CosNotification::QoSProperties qos (1);
  88       qos.length (1);
  89       qos[0].name = CORBA::string_dup (NotifyExt::ThreadPool);
  90       qos[0].value <<= tp_params;
  91
  92       // Obtain the proxy. The QoS is applied to the POA in which the Proxy is hosted.
  93       proxyconsumer = admin_ext->obtain_notification_push_consumer_with_qos (CosNotifyChannelAdmin::STRUCTURED_EVENT
  94                                                                                    , proxy_consumer_id_, qos);
  95     }
  96   else
  97     {
  98       // Obtain the proxy.
  99       proxyconsumer = this->admin_->obtain_notification_push_consumer (CosNotifyChannelAdmin::STRUCTURED_EVENT
 100                                                                        , proxy_consumer_id_);
 101     }
 102
 103   ACE_ASSERT (!CORBA::is_nil (proxyconsumer.in ()));
 104
 105   // narrow
 106   this->proxy_consumer_ =
 107     CosNotifyChannelAdmin::StructuredProxyPushConsumer::_narrow (proxyconsumer.in ());
 108
 109   ACE_ASSERT (!CORBA::is_nil (proxy_consumer_.in ()));
 110
 111   // connect to the proxyconsumer.
 112   proxy_consumer_->connect_structured_push_supplier (objref.in ());
 113
 114   ACE_DEBUG ((LM_DEBUG, "(%P,%t) Created Supplier %d with %d threads at the ProxyConsumer\n", proxy_consumer_id_,
 115               this->proxy_consumer_thread_count_));
 116 }
 117
 118 void
 119 TAO_Notify_ThreadPool_Supplier::disconnect ()
 120 {
 121   ACE_ASSERT (!CORBA::is_nil (this->proxy_consumer_.in ()));
 122
 123   this->proxy_consumer_->disconnect_structured_push_consumer();
 124 }
 125
 126 void
 127 TAO_Notify_ThreadPool_Supplier::deactivate ()
 128 {
 129   PortableServer::POA_var poa (this->_default_POA ());
 130
 131   PortableServer::ObjectId_var id (poa->servant_to_id (this));
 132
 133   poa->deactivate_object (id.in());
 134 }
 135
 136 void
 137 TAO_Notify_ThreadPool_Supplier::subscription_change (const CosNotification::EventTypeSeq & added,
 138                                       const CosNotification::EventTypeSeq & /*removed */)
 139 {
 140   ACE_GUARD (TAO_SYNCH_MUTEX, mon, this->lock_);
 141
 142   // Count the number of consumers connect and signal the supplier thread when the expected count have connected.
 143   // Only 1 consumer connects at a time.
 144   if (added.length () > 0)
 145     {
 146       // Set the domain and type nams in the event's fixed header.
 147       this->event_[consumer_count_].header.fixed_header.event_type.domain_name = CORBA::string_dup(added[0].domain_name);
 148       this->event_[consumer_count_].header.fixed_header.event_type.type_name = CORBA::string_dup(added[0].type_name);
 149
 150       ++this->consumer_count_;
 151
 152       ACE_DEBUG ((LM_DEBUG, "(%P,%t) Received Type %d: (%s)\n", this->consumer_count_, added[0].type_name.in ()));
 153
 154       if (this->consumer_count_ == this->expected_consumer_count_)
 155         this->consumers_connected_.signal ();
 156     }
 157 }
 158
 159 void
 160 TAO_Notify_ThreadPool_Supplier::send_event (const CosNotification::StructuredEvent& event)
 161 {
 162   ACE_ASSERT (!CORBA::is_nil (this->proxy_consumer_.in ()));
 163
 164   proxy_consumer_->push_structured_event (event);
 165 }
 166
 167 void
 168 TAO_Notify_ThreadPool_Supplier::disconnect_structured_push_supplier ()
 169 {
 170   this->deactivate ();
 171 }