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

   1 #include "Consumer.h"
   2
   3 TAO_Notify_Lanes_Consumer::TAO_Notify_Lanes_Consumer (TAO_Notify_ORB_Objects& orb_objects)
   4   : orb_objects_ (orb_objects)
   5 {
   6 }
   7
   8 TAO_Notify_Lanes_Consumer::~TAO_Notify_Lanes_Consumer (void)
   9 {
  10 }
  11
  12 void
  13 TAO_Notify_Lanes_Consumer::init (PortableServer::POA_var& poa, CosNotifyChannelAdmin::ConsumerAdmin_var& admin, ACE_CString& event_type)
  14 {
  15   this->default_POA_ = poa;
  16   this->admin_ = admin;
  17   this->event_type_ = event_type;
  18
  19   this->connect ();
  20 }
  21
  22 PortableServer::POA_ptr
  23 TAO_Notify_Lanes_Consumer::_default_POA (void)
  24 {
  25   return PortableServer::POA::_duplicate (this->default_POA_.in ());
  26 }
  27
  28 void
  29 TAO_Notify_Lanes_Consumer::run (void)
  30 {
  31   // Nothing to do.
  32 }
  33
  34 void
  35 TAO_Notify_Lanes_Consumer::connect (void)
  36 {
  37   // Activate the consumer with the default_POA_
  38   CosNotifyComm::StructuredPushConsumer_var objref = this->_this ();
  39
  40   CosNotifyChannelAdmin::ProxySupplier_var proxysupplier =
  41     this->admin_->obtain_notification_push_supplier (CosNotifyChannelAdmin::STRUCTURED_EVENT
  42                                                      , proxy_supplier_id_);
  43
  44   ACE_ASSERT (!CORBA::is_nil (proxysupplier.in ()));
  45
  46   // narrow
  47   this->proxy_supplier_ =
  48     CosNotifyChannelAdmin::StructuredProxyPushSupplier::_narrow (proxysupplier.in ());
  49
  50   ACE_ASSERT (!CORBA::is_nil (proxy_supplier_.in ()));
  51
  52   this->proxy_supplier_->connect_structured_push_consumer (objref.in ());
  53
  54   // Call subscription_change to inform the supplier that this consumer is available.
  55   CosNotification::EventTypeSeq added (1);
  56   CosNotification::EventTypeSeq removed;
  57
  58   added.length (1);
  59   added[0].domain_name = CORBA::string_dup ("TEST_DOMAIN");
  60   added[0].type_name = CORBA::string_dup (this->event_type_.c_str ());
  61
  62   this->proxy_supplier_->subscription_change (added, removed);
  63 }
  64
  65 void
  66 TAO_Notify_Lanes_Consumer::disconnect (void)
  67 {
  68   this->proxy_supplier_->disconnect_structured_push_supplier();
  69 }
  70
  71 void
  72 TAO_Notify_Lanes_Consumer::offer_change (const CosNotification::EventTypeSeq & /*added*/,
  73                                const CosNotification::EventTypeSeq & /*removed*/)
  74 {
  75   // No-Op.
  76 }
  77
  78 void
  79 TAO_Notify_Lanes_Consumer::push_structured_event (const CosNotification::StructuredEvent & notification)
  80 {
  81   try
  82     {
  83       // Check the current threads priority.
  84       RTCORBA::Priority thread_priority =
  85         this->orb_objects_.current_->the_priority ();
  86
  87       const CosNotification::PropertySeq& prop_seq = notification.header.variable_header;
  88
  89       // Extract the priority at which the supplier send it.
  90       RTCORBA::Priority event_priority = 0;
  91
  92       for (CORBA::ULong i = 0; i < prop_seq.length (); ++i)
  93         {
  94           if (ACE_OS::strcmp (prop_seq[i].name.in (), CosNotification::Priority) == 0)
  95             prop_seq[i].value >>= event_priority;
  96         }
  97
  98       ACE_DEBUG ((LM_DEBUG, "(%P, %t) Consumer received event with priority = %d and thread priority = %d\n",
  99                   event_priority, thread_priority));
 100
 101       // The current thread priority and the event priority must match.
 102       if (event_priority != thread_priority)
 103         ACE_DEBUG ((LM_DEBUG, "(%P, %t) Error: Event priority and thread priority are different.\n"));
 104
 105       // Disconnect from the EC
 106       this->disconnect ();
 107
 108       // Deactivate this object.
 109       this->deactivate ();
 110
 111       // We received the event, shutdown the ORB.
 112       this->orb_objects_.orb_->shutdown (1);
 113     }
 114   catch (const CORBA::Exception& ex)
 115     {
 116       ex._tao_print_exception (ACE_TEXT ("Consumer error "));
 117
 118       return;
 119     }
 120 }
 121
 122 void
 123 TAO_Notify_Lanes_Consumer::deactivate (void)
 124 {
 125   PortableServer::POA_var poa (this->_default_POA ());
 126
 127   PortableServer::ObjectId_var id (poa->servant_to_id (this));
 128
 129   poa->deactivate_object (id.in());
 130 }
 131
 132 void
 133 TAO_Notify_Lanes_Consumer::disconnect_structured_push_consumer (void)
 134 {
 135   this->deactivate ();
 136 }