TAO_RTEvent: TAO_EC_Basic_ObserverStrategy class Reference

00047   :  event_channel_ (ec),
00048      lock_ (lock),
00049      handle_generator_ (1)
00050 {
00051 }

00063 {
00064   delete this->lock_;
00065   this->lock_ = 0;
00066 }

00071 {
00072   {
00073     ACE_GUARD_THROW_EX (
00074         ACE_Lock, ace_mon, *this->lock_,
00075         RtecEventChannelAdmin::EventChannel::SYNCHRONIZATION_ERROR());
00076 
00077     this->handle_generator_++;
00078     Observer_Entry entry (this->handle_generator_,
00079                           RtecEventChannelAdmin::Observer::_duplicate (obs));
00080 
00081     if (this->observers_.bind (entry.handle, entry) == -1)
00082       throw RtecEventChannelAdmin::EventChannel::CANT_APPEND_OBSERVER();
00083   }
00084 
00085   RtecEventChannelAdmin::ConsumerQOS c_qos;
00086   this->fill_qos (c_qos);
00087   obs->update_consumer (c_qos);
00088 
00089   RtecEventChannelAdmin::SupplierQOS s_qos;
00090   this->fill_qos (s_qos);
00091   obs->update_supplier (s_qos);
00092 
00093   return this->handle_generator_;
00094 }

00182 {
00183   this->consumer_qos_update (supplier);
00184 }

00140 {
00141   this->supplier_qos_update (consumer);
00142 }

00196 {
00197   if (supplier->subscriptions ().is_gateway)
00198     return;
00199 
00200   RtecEventChannelAdmin::ConsumerQOS c_qos;
00201   this->fill_qos (c_qos);
00202 
00203   RtecEventChannelAdmin::Observer_var *tmp = 0;
00204   int size = this->create_observer_list (tmp);
00205   ACE_Auto_Basic_Array_Ptr<RtecEventChannelAdmin::Observer_var> copy (tmp);
00206 
00207   for (int i = 0; i != size; ++i)
00208     {
00209       try
00210         {
00211           copy[i]->update_consumer (c_qos);
00212         }
00213       catch (const CORBA::Exception&)
00214         {
00215           // Ignore exceptions, we must isolate other observers from
00216           // failures on this one.
00217         }
00218     }
00219 }

00111 {
00112   ACE_GUARD_THROW_EX (ACE_Lock, ace_mon, *this->lock_,
00113                  RtecEventChannelAdmin::EventChannel::SYNCHRONIZATION_ERROR());
00114 
00115   int size = static_cast<int> (this->observers_.current_size ());
00116   RtecEventChannelAdmin::Observer_var *tmp = 0;
00117   ACE_NEW_RETURN (tmp,
00118                   RtecEventChannelAdmin::Observer_var[size],
00119                   0);
00120   ACE_Auto_Basic_Array_Ptr<RtecEventChannelAdmin::Observer_var> copy (tmp);
00121 
00122   Observer_Map_Iterator end = this->observers_.end ();
00123   int j = 0;
00124   for (Observer_Map_Iterator i  = this->observers_.begin ();
00125        i != end;
00126        ++i)
00127     {
00128       Observer_Entry& entry = (*i).int_id_;
00129       copy[j++] =
00130         RtecEventChannelAdmin::Observer::_duplicate (entry.observer.in ());
00131     }
00132 
00133   lst = copy.release ();
00134   return size;
00135 }

00189 {
00190   this->consumer_qos_update (supplier);
00191 }

00147 {
00148   this->supplier_qos_update (consumer);
00149 }

00249 {
00250   Headers headers;
00251 
00252   TAO_EC_Accumulate_Consumer_Headers worker (headers);
00253   this->event_channel_->for_each_supplier (&worker);
00254 
00255   qos.publications.length (static_cast<CORBA::ULong> (headers.current_size ()));
00256 
00257   CORBA::ULong count = 0;
00258   for (HeadersIterator i = headers.begin (); i != headers.end (); ++i)
00259     {
00260       qos.publications[count++].event.header = (*i).key ();
00261     }
00262 }

00224 {
00225   Headers headers;
00226 
00227   TAO_EC_Accumulate_Supplier_Headers worker (headers);
00228   this->event_channel_->for_each_consumer (&worker);
00229 
00230   RtecEventChannelAdmin::DependencySet& dep = qos.dependencies;
00231 
00232   dep.length (static_cast<CORBA::ULong> (headers.current_size () + 1));
00233 
00234   dep[0].event.header.type = ACE_ES_DISJUNCTION_DESIGNATOR;
00235   dep[0].event.header.source = static_cast<CORBA::ULong> (headers.current_size ());
00236   dep[0].event.header.creation_time = ORBSVCS_Time::zero ();
00237   dep[0].rt_info = 0;
00238 
00239   CORBA::ULong count = 1;
00240   for (HeadersIterator i = headers.begin (); i != headers.end (); ++i)
00241     {
00242       dep[count++].event.header = (*i).key ();
00243     }
00244 }

00100 {
00101   ACE_GUARD_THROW_EX (ACE_Lock, ace_mon, *this->lock_,
00102       RtecEventChannelAdmin::EventChannel::SYNCHRONIZATION_ERROR());
00103 
00104   if (this->observers_.unbind (handle) == -1)
00105     throw RtecEventChannelAdmin::EventChannel::CANT_REMOVE_OBSERVER();
00106 }

00154 {
00155   if (consumer->publications ().is_gateway)
00156     return;
00157 
00158   RtecEventChannelAdmin::SupplierQOS s_qos;
00159   this->fill_qos (s_qos);
00160 
00161   RtecEventChannelAdmin::Observer_var *tmp = 0;
00162   int size = this->create_observer_list (tmp);
00163   ACE_Auto_Basic_Array_Ptr<RtecEventChannelAdmin::Observer_var> copy (tmp);
00164 
00165   for (int i = 0; i != size; ++i)
00166     {
00167       try
00168         {
00169           copy[i]->update_supplier (s_qos);
00170         }
00171       catch (const CORBA::Exception&)
00172         {
00173           // Ignore exceptions, we must isolate other observers from
00174           // failures on this one.
00175         }
00176     }
00177 }


Public Types
typedef ACE_Map_Manager< RtecEventChannelAdmin::Observer_Handle, Observer_Entry, ACE_Null_Mutex >	Observer_Map
typedef ACE_Map_Iterator< RtecEventChannelAdmin::Observer_Handle, Observer_Entry, ACE_Null_Mutex >	Observer_Map_Iterator
typedef ACE_RB_Tree< RtecEventComm::EventHeader, int, Header_Compare, ACE_Null_Mutex >	Headers
typedef ACE_RB_Tree_Iterator< RtecEventComm::EventHeader, int, Header_Compare, ACE_Null_Mutex >	HeadersIterator
Public Member Functions
	TAO_EC_Basic_ObserverStrategy (TAO_EC_Event_Channel_Base ec, ACE_Lock lock)
	Constructor.
virtual	~TAO_EC_Basic_ObserverStrategy (void)
	Destructor.
virtual RtecEventChannelAdmin::Observer_Handle	append_observer (RtecEventChannelAdmin::Observer_ptr)
	The basic methods to support the EC strategies.
virtual void	remove_observer (RtecEventChannelAdmin::Observer_Handle)
virtual void	connected (TAO_EC_ProxyPushConsumer *)
virtual void	disconnected (TAO_EC_ProxyPushConsumer *)
virtual void	connected (TAO_EC_ProxyPushSupplier *)
virtual void	disconnected (TAO_EC_ProxyPushSupplier *)
Protected Member Functions

virtual void	consumer_qos_update (TAO_EC_ProxyPushSupplier *supplier)
	Recompute EC consumer subscriptions and send them out to all observers.
virtual void	supplier_qos_update (TAO_EC_ProxyPushConsumer *consumer)
	Recompute EC supplier publications and send them out to all observers.
void	fill_qos (RtecEventChannelAdmin::ConsumerQOS &qos)
	Compute consumer QOS.
void	fill_qos (RtecEventChannelAdmin::SupplierQOS &qos)
	Compute supplier QOS.
int	create_observer_list (RtecEventChannelAdmin::Observer_var *&lst)
Protected Attributes
TAO_EC_Event_Channel_Base *	event_channel_
	The event channel.
ACE_Lock *	lock_
	The lock.
RtecEventChannelAdmin::Observer_Handle	handle_generator_
Observer_Map	observers_
	Keep the set of Observers.

TAO_EC_Basic_ObserverStrategy Class Reference

Public Types

Public Member Functions

Protected Member Functions

Protected Attributes

Detailed Description

Memory Management

Member Typedef Documentation

Constructor & Destructor Documentation

Member Function Documentation

Member Data Documentation