TAO/orbsvcs/examples/Notify/Federation/Gate/Gate.cpp

   1 // author    : Boris Kolpackov <boris@dre.vanderbilt.edu>
   2 #include "Gate.h"
   3 #include "ace/Truncate.h"
   4
   5 using namespace CORBA;
   6 using namespace CosNotifyComm;
   7 using namespace CosNotification;
   8 using namespace CosNotifyChannelAdmin;
   9
  10 Gate::~Gate ()
  11 {
  12   // Stop tracker thread.
  13   {
  14     ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->mutex_);
  15     stop_ = true;
  16   }
  17
  18   thread_mgr_.wait ();
  19 }
  20
  21
  22 Gate::
  23 Gate (ACE_INET_Addr const& group, EventChannel_ptr ch)
  24     : socket_ (group, false),
  25       stop_ (false)
  26 {
  27   init (ch->default_consumer_admin (),
  28         ch->default_supplier_admin ());
  29 }
  30
  31 Gate::
  32 Gate (ACE_INET_Addr const& group,
  33       ConsumerAdmin_ptr consumer_admin,
  34       SupplierAdmin_ptr supplier_admin)
  35     : socket_ (group, false),
  36       stop_ (false)
  37 {
  38   init (consumer_admin, supplier_admin);
  39 }
  40
  41 void Gate::
  42 init (ConsumerAdmin_ptr consumer_admin,
  43       SupplierAdmin_ptr supplier_admin)
  44 {
  45   // Generate unique id. It is used to prevent event looping.
  46   //
  47   ACE_Utils::UUID uuid;
  48   ACE_Utils::UUID_GENERATOR::instance ()->init ();
  49   ACE_Utils::UUID_GENERATOR::instance ()->generate_UUID (uuid);
  50
  51   id_ = string_alloc (ACE_Utils::truncate_cast<CORBA::ULong> (
  52                         uuid.to_string ()->length () + 2));
  53   ACE_OS::strcpy (id_.inout (), "_");
  54   ACE_OS::strcpy (id_.inout () + 1, uuid.to_string ()->rep ());
  55
  56   // ACE_DEBUG ((LM_DEBUG, "ID: %s\n", id_.in ()));
  57
  58
  59   // Obtain proxy consumer.
  60   //
  61   ProxyConsumer_var pc (
  62     supplier_admin->obtain_notification_push_consumer (
  63       STRUCTURED_EVENT, consumer_id_));
  64
  65   consumer_ = StructuredProxyPushConsumer::_narrow (pc.in ());
  66
  67   consumer_->connect_structured_push_supplier (
  68     StructuredPushSupplier::_nil ());
  69
  70
  71   // Register as consumer.
  72   //
  73   StructuredPushConsumer_var ref (_this ()); // Activate on default POA.
  74
  75   ProxySupplier_var ps (
  76     consumer_admin->obtain_notification_push_supplier (
  77       STRUCTURED_EVENT, supplier_id_));
  78
  79   supplier_ = StructuredProxyPushSupplier::_narrow (ps.in ());
  80
  81   supplier_->connect_structured_push_consumer (ref.in ());
  82
  83
  84   // Create tracker thread.
  85   //
  86   thread_mgr_.spawn (tracker_thunk, this);
  87 }
  88
  89 ACE_THR_FUNC_RETURN Gate::
  90 tracker_thunk (void* arg)
  91 {
  92   Gate* a = reinterpret_cast<Gate*> (arg);
  93   a->tracker ();
  94   return 0;
  95 }
  96
  97 void Gate::
  98 tracker ()
  99 {
 100   // Time period after which a manual cancellation request is
 101   // checked for.
 102   //
 103   ACE_Time_Value const timeout (0, 500);
 104
 105   while (true)
 106   {
 107     ssize_t n;
 108
 109     while (true)
 110     {
 111       n = socket_.size (timeout);
 112
 113       // Check for cancellation request.
 114       //
 115       {
 116         ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->mutex_);
 117
 118         if (stop_)
 119           return;
 120       }
 121
 122       if (n == -1)
 123       {
 124         if (errno != ETIME)
 125           ACE_OS::abort ();
 126       }
 127       else
 128         break;
 129     }
 130
 131     CORBA::ULong seqn = ACE_Utils::truncate_cast<CORBA::ULong> (n);
 132     OctetSeq seq (seqn);
 133     seq.length (seqn);
 134
 135     char* buffer = reinterpret_cast<char*> (seq.get_buffer ());
 136
 137     if (socket_.recv (buffer, n) != n)
 138     {
 139       ACE_ERROR ((LM_ERROR,
 140                   "recv() reported different size than size()\n"));
 141       continue;
 142     }
 143
 144     TAO_InputCDR cdr (buffer, n);
 145
 146     StructuredEvent e;
 147
 148     cdr >> e;
 149
 150     // Add TTL header to prevent infinite message looping.
 151     //
 152     ULong i (0);
 153
 154     for (; i < e.header.variable_header.length (); ++i)
 155     {
 156       if (ACE_OS::strcmp (e.header.variable_header[i].name.in (), id_.in ()) == 0)
 157         break;
 158     }
 159
 160     if (i == e.header.variable_header.length ())
 161     {
 162       e.header.variable_header.length (i + 1);
 163
 164       e.header.variable_header[i].name = string_dup (id_.in ());
 165     }
 166
 167     //ACE_DEBUG ((LM_DEBUG,
 168     //            "adding %s as header #%d\n",
 169     //            e.header.variable_header[i].name.in (), i));
 170
 171     e.header.variable_header[i].value <<= ULong (1);
 172
 173     /*
 174     cerr << "IN: "
 175          << e.header.fixed_header.event_type.domain_name << "::"
 176          << e.header.fixed_header.event_type.type_name << " "
 177          << e.header.fixed_header.event_name << endl;
 178     */
 179
 180     consumer_->push_structured_event (e);
 181   }
 182 }
 183
 184 void Gate::
 185 push_structured_event (StructuredEvent const& e)
 186 {
 187   for (ULong i (0); i < e.header.variable_header.length (); ++i)
 188   {
 189     if (ACE_OS::strcmp (e.header.variable_header[i].name.in (), id_.in ()) == 0)
 190     {
 191       ULong ttl;
 192
 193       e.header.variable_header[i].value >>= ttl;
 194
 195       if (ttl <= 1)
 196       {
 197         //ACE_DEBUG ((LM_DEBUG,
 198         //            "DROPPED\n"));
 199         return;
 200       }
 201
 202       break;
 203     }
 204   }
 205
 206   /*
 207   cerr << "OUT: "
 208        << e.header.fixed_header.event_type.domain_name << "::"
 209        << e.header.fixed_header.event_type.type_name << " "
 210        << e.header.fixed_header.event_name << endl;
 211   */
 212
 213   TAO_OutputCDR cdr;
 214
 215   cdr << e;
 216
 217   CORBA::ULong size (ACE_Utils::truncate_cast<CORBA::ULong> (cdr.total_length ()));
 218
 219   OctetSeq seq (size);
 220   seq.length (size);
 221
 222   char* buffer = reinterpret_cast<char*> (seq.get_buffer ());
 223
 224   {
 225     char* buf = buffer;
 226
 227     for (ACE_Message_Block const* mb = cdr.begin ();
 228          mb != 0;
 229          mb = mb->cont ())
 230     {
 231       ACE_OS::memcpy (buf, mb->rd_ptr (), mb->length ());
 232       buf += mb->length ();
 233     }
 234   }
 235
 236   socket_.send (buffer, size);
 237 }
 238
 239
 240 void
 241 Gate::disconnect_structured_push_consumer (void)
 242 {
 243   // We don't care.
 244 }
 245
 246 void
 247 Gate::offer_change (EventTypeSeq const&, EventTypeSeq const&)
 248 {
 249   // We don't care.
 250 }