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

   1 #include "Supplier.h"
   2
   3
   4 #include "tao/ORB_Core.h"
   5
   6 TAO_Notify_Lanes_Supplier::TAO_Notify_Lanes_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 {
  13 }
  14
  15 TAO_Notify_Lanes_Supplier::~TAO_Notify_Lanes_Supplier ()
  16 {
  17 }
  18
  19 void
  20 TAO_Notify_Lanes_Supplier::init (CosNotifyChannelAdmin::SupplierAdmin_var& admin, int expected_consumer_count)
  21 {
  22   // First initialize the class members.
  23   this->admin_ = admin;
  24   this->expected_consumer_count_ = expected_consumer_count;
  25
  26   this->connect ();
  27 }
  28
  29 void
  30 TAO_Notify_Lanes_Supplier::run ()
  31 {
  32   // The Priority at which we send the first event to the first consumer.
  33   RTCORBA::Priority priority = 1;
  34
  35   {
  36     ACE_GUARD (TAO_SYNCH_MUTEX, mon, this->lock_);
  37
  38     ACE_DEBUG ((LM_DEBUG, "(%P, %t) Waiting for %d consumers to connect...\n", this->expected_consumer_count_));
  39
  40     // Wait till the consumers are ready to go.
  41     while (this->consumer_count_ != this->expected_consumer_count_)
  42       this->consumers_connected_.wait ();
  43   }
  44
  45   // Send an event each to each consumer.
  46   // Each Consumer expects a different priority.
  47   for (int i = 0; i < this->expected_consumer_count_; ++i, ++priority)
  48     {
  49       // Set this threads priority.
  50       this->orb_objects_.current_->the_priority (priority);
  51
  52       // Make sure the priority was set, get the priority of the current thread.
  53       RTCORBA::Priority thread_priority =
  54         this->orb_objects_.current_->the_priority ();
  55
  56       // We will send this event.
  57       CosNotification::StructuredEvent event;
  58
  59       // Populate the Priority field so that the consumer can deduce the suppliers priority
  60       // to do a sanity check when it receives the event.
  61       CosNotification::PropertySeq& opt =  event.header.variable_header;
  62       opt.length (1);
  63
  64       CORBA::Any buffer;
  65       buffer <<= (CORBA::Short) thread_priority;
  66
  67       opt[0].name = CORBA::string_dup (CosNotification::Priority);
  68       opt[0].value = buffer;
  69
  70       // Set the domain and type nams in the event's fixed header.
  71       char type[BUFSIZ];
  72       ACE_OS::sprintf (type, "TEST_TYPE_%d", thread_priority);
  73
  74       event.header.fixed_header.event_type.domain_name = CORBA::string_dup("TEST_DOMAIN");
  75       event.header.fixed_header.event_type.type_name = CORBA::string_dup(type);
  76
  77       ACE_DEBUG ((LM_DEBUG,
  78                   "(%P, %t) Supplier is sending an event of type %s at priority %d\n", type, thread_priority));
  79
  80       // send the event
  81       this->send_event (event);
  82     } // repeat for the next consumer at the next priority.
  83
  84   // Disconnect from the EC
  85   this->disconnect ();
  86
  87   // Deactivate this object.
  88   this->deactivate ();
  89
  90   // we're done. shutdown the ORB to exit the process.
  91   this->orb_objects_.orb_->shutdown (true);
  92 }
  93
  94 void
  95 TAO_Notify_Lanes_Supplier::connect ()
  96 {
  97   // Activate the supplier object.
  98   CosNotifyComm::StructuredPushSupplier_var objref = this->_this ();
  99
 100   // Obtain the proxy.
 101   CosNotifyChannelAdmin::ProxyConsumer_var proxyconsumer =
 102     this->admin_->obtain_notification_push_consumer (CosNotifyChannelAdmin::STRUCTURED_EVENT
 103                                                      , proxy_consumer_id_);
 104
 105   ACE_ASSERT (!CORBA::is_nil (proxyconsumer.in ()));
 106
 107   // narrow
 108   this->proxy_consumer_ =
 109     CosNotifyChannelAdmin::StructuredProxyPushConsumer::_narrow (proxyconsumer.in ());
 110
 111   ACE_ASSERT (!CORBA::is_nil (proxy_consumer_.in ()));
 112
 113   // connect to the proxyconsumer.
 114   proxy_consumer_->connect_structured_push_supplier (objref.in ());
 115 }
 116
 117 void
 118 TAO_Notify_Lanes_Supplier::disconnect ()
 119 {
 120   ACE_ASSERT (!CORBA::is_nil (this->proxy_consumer_.in ()));
 121
 122   this->proxy_consumer_->disconnect_structured_push_consumer();
 123 }
 124
 125 void
 126 TAO_Notify_Lanes_Supplier::deactivate ()
 127 {
 128   PortableServer::POA_var poa (this->_default_POA ());
 129
 130   PortableServer::ObjectId_var id (poa->servant_to_id (this));
 131
 132   poa->deactivate_object (id.in());
 133 }
 134
 135 void
 136 TAO_Notify_Lanes_Supplier::subscription_change (const CosNotification::EventTypeSeq & added,
 137                                       const CosNotification::EventTypeSeq & /*removed */)
 138 {
 139   ACE_GUARD (TAO_SYNCH_MUTEX, mon, this->lock_);
 140
 141   // Count the number of consumers connect and signal the supplier thread when the expected count have connected.
 142   if (added.length () > 0)
 143     {
 144       if (++this->consumer_count_ == this->expected_consumer_count_)
 145         this->consumers_connected_.signal ();
 146     }
 147 }
 148
 149 void
 150 TAO_Notify_Lanes_Supplier::send_event (const CosNotification::StructuredEvent& event)
 151 {
 152   ACE_ASSERT (!CORBA::is_nil (this->proxy_consumer_.in ()));
 153
 154   proxy_consumer_->push_structured_event (event);
 155 }
 156
 157 void
 158 TAO_Notify_Lanes_Supplier::disconnect_structured_push_supplier ()
 159 {
 160   this->deactivate ();
 161 }