TAO/orbsvcs/tests/FtRtEvent/PushSupplier.cpp

   1 #include "orbsvcs/CosNamingC.h"
   2 #include "orbsvcs/Event_Utilities.h"
   3 #include "PushSupplier.h"
   4 #include "ace/Reactor.h"
   5 #include "ace/Select_Reactor.h"
   6 #include "tao/MProfile.h"
   7 #include "tao/Stub.h"
   8 #include "orbsvcs/FtRtEvent/Utils/resolve_init.h"
   9
  10 int
  11 PushSupplier_impl::ReactorTask::svc ()
  12 {
  13   ACE_DEBUG((LM_DEBUG, "Reactor Thread started\n"));
  14   ACE_Reactor reactor (new ACE_Select_Reactor) ;
  15   reactor_ = &reactor;
  16
  17   extern ACE_Time_Value timer_interval;
  18
  19   if (reactor_->schedule_timer(handler_, 0, ACE_Time_Value::zero, timer_interval)== -1)
  20     ACE_ERROR_RETURN((LM_ERROR,"Cannot schedule timer\n"),-1);
  21
  22   reactor_->run_reactor_event_loop();
  23   ACE_DEBUG((LM_DEBUG, "Reactor Thread ended\n"));
  24
  25   return 0;
  26 }
  27
  28
  29 PushSupplier_impl::PushSupplier_impl(CORBA::ORB_ptr orb)
  30 : orb_(orb), seq_no_(0), reactor_task_(this)
  31 {
  32 }
  33
  34 PushSupplier_impl::~PushSupplier_impl()
  35 {
  36   reactor_task_.wait();
  37 }
  38
  39 int PushSupplier_impl::init(RtecEventChannelAdmin::EventChannel_ptr channel)
  40 {
  41   ACE_DEBUG((LM_DEBUG, "for_suppliers\n"));
  42   RtecEventChannelAdmin::SupplierAdmin_var supplier_admin =
  43     channel->for_suppliers();
  44
  45   ACE_DEBUG((LM_DEBUG, "obtain_push_consumer\n"));
  46   consumer_ =
  47     supplier_admin->obtain_push_consumer();
  48
  49
  50   ACE_DEBUG((LM_DEBUG, "got push_consumer with %d profiles\n",
  51                         consumer_->_stubobj ()->base_profiles ().profile_count ()));
  52
  53     RtecEventChannelAdmin::SupplierQOS qos;
  54     qos.publications.length (1);
  55     RtecEventComm::EventHeader& h0 =
  56         qos.publications[0].event.header;
  57     h0.type   = ACE_ES_EVENT_UNDEFINED; // first free event type
  58     h0.source = 1;                      // first free event source
  59
  60   RtecEventComm::PushSupplier_var supplier = _this();
  61
  62     ACE_DEBUG((LM_DEBUG, "connect_push_supplier\n"));
  63   consumer_->connect_push_supplier(supplier.in(),
  64     qos);
  65
  66   ACE_DEBUG((LM_DEBUG, "push_consumer connected\n"));
  67
  68
  69   if (!reactor_task_.thr_count() &&
  70     reactor_task_.activate (THR_NEW_LWP | THR_JOINABLE, 1) != 0)
  71         ACE_ERROR_RETURN ((LM_ERROR,
  72     "Cannot activate reactor thread\n"),
  73     -1);
  74
  75   return 0;
  76 }
  77
  78
  79 void  PushSupplier_impl::disconnect_push_supplier (
  80       )
  81 {
  82   reactor_task_.reactor_->end_reactor_event_loop();
  83
  84   PortableServer::Current_var current =
  85     resolve_init<PortableServer::Current>(orb_.in(), "POACurrent");
  86
  87   PortableServer::POA_var poa = current->get_POA();
  88
  89   PortableServer::ObjectId_var oid = current->get_object_id();
  90
  91   poa->deactivate_object (oid.in ());
  92 }
  93
  94 int PushSupplier_impl::handle_timeout (const ACE_Time_Value &current_time,
  95                      const void *act)
  96 {
  97   ACE_UNUSED_ARG(act);
  98   ACE_UNUSED_ARG(current_time);
  99
 100   try{
 101     RtecEventComm::EventSet event (1);
 102     event.length (1);
 103     event[0].header.type   = ACE_ES_EVENT_UNDEFINED;
 104     event[0].header.source = 1;
 105     event[0].header.ttl    = 1;
 106
 107     ACE_Time_Value time_val = ACE_OS::gettimeofday ();
 108
 109     event[0].header.ec_send_time = time_val.sec () * 10000000 + time_val.usec ()* 10;
 110     event[0].data.any_value <<= seq_no_;
 111
 112     consumer_->push(event);
 113     ACE_DEBUG((LM_DEBUG, "sending data %d\n", seq_no_));
 114     ++seq_no_;
 115   }
 116   catch (const CORBA::Exception& ex)
 117   {
 118     ex._tao_print_exception ("A CORBA Exception occurred.");
 119   }
 120   return 0;
 121 }