TAO_Notify_Consumer Class Reference

Astract Base class for wrapping consumer objects that connect to the EventChannel. More...

#include <Consumer.h>

Inheritance diagram for TAO_Notify_Consumer:

[legend]Collaboration diagram for TAO_Notify_Consumer:

[legend]List of all members.

Public Types
enum	DispatchStatus { DISPATCH_SUCCESS, DISPATCH_RETRY, DISPATCH_DISCARD, DISPATCH_FAIL }
	Status returned from dispatch attempts. More...
Public Member Functions
	TAO_Notify_Consumer (TAO_Notify_ProxySupplier *proxy)
	Constuctor.
virtual	~TAO_Notify_Consumer ()
	Destructor.
TAO_Notify_ProxySupplier *	proxy_supplier (void)
	Access Specific Proxy.
virtual TAO_Notify_Proxy *	proxy (void)
	Access Base Proxy.
void	deliver (TAO_Notify_Method_Request_Event *request)
	Dispatch Event to consumer.
virtual void	push (const CORBA::Any &event)=0
	Push to this consumer.
virtual void	push (const CosNotification::StructuredEvent &event)=0
	Push to this consumer.
virtual void	push (const CosNotification::EventBatch &event)=0
	Push a batch of events to this consumer.
DispatchStatus	dispatch_batch (const CosNotification::EventBatch &batch)
	Dispatch the batch of events to the attached consumer.
void	dispatch_pending ()
	Dispatch the pending events.
CORBA::Boolean	is_suspended (void)
	Is the connection suspended?
void	suspend ()
	Suspend Connection.
void	resume ()
	Resume Connection.
virtual void	shutdown ()
	Shutdown the consumer.
virtual void	reconnect_from_consumer (TAO_Notify_Consumer *old_consumer)=0
virtual void	qos_changed (const TAO_Notify_QoSProperties &qos_properties)
	Override, Peer::qos_changed.
Protected Types
typedef ACE_Unbounded_Queue< TAO_Notify_Method_Request_Event_Queueable * >	Request_Queue
Protected Member Functions
DispatchStatus	dispatch_request (TAO_Notify_Method_Request_Event *request)
virtual bool	dispatch_from_queue (Request_Queue &requests, ACE_Guard< TAO_SYNCH_MUTEX > &ace_mon)
	Attempt to dispatch event from a queue.
void	enqueue_request (TAO_Notify_Method_Request_Event *request)
virtual bool	enqueue_if_necessary (TAO_Notify_Method_Request_Event *request)
virtual void	dispatch_updates_i (const CosNotification::EventTypeSeq &added, const CosNotification::EventTypeSeq &removed)
	Implementation of Peer specific dispatch_updates.
TAO_SYNCH_MUTEX *	proxy_lock (void)
	Get the shared Proxy Lock.
virtual int	handle_timeout (const ACE_Time_Value &current_time, const void *act=0)
void	schedule_timer (bool is_error=false)
	Schedule timer.
void	cancel_timer (void)
	Cancel timer.
Request_Queue &	pending_events ()
	= Protected Data Members
Protected Attributes
TAO_Notify_ProxySupplier *	proxy_
	The Proxy that we associate with.
CORBA::Boolean	is_suspended_
	Suspended Flag.
CosNotifyComm::NotifyPublish_var	publish_
	Interface that accepts offer_changes.
const TAO_Notify_Property_Time &	pacing_
	The Pacing Interval.
TAO_Notify_Property_Long	max_batch_size_
	Max. batch size.
long	timer_id_
	Timer Id.
TAO_Notify_Timer::Ptr	timer_
	The Timer Manager that we use.
Private Attributes
ACE_Auto_Ptr< Request_Queue >	pending_events_
	Events pending to be delivered.

---

Detailed Description

Astract Base class for wrapping consumer objects that connect to the EventChannel.

Definition at line 44 of file Consumer.h.

---

Member Typedef Documentation

typedef ACE_Unbounded_Queue<TAO_Notify_Method_Request_Event_Queueable *> TAO_Notify_Consumer::Request_Queue [protected]

Definition at line 110 of file Consumer.h. Referenced by dispatch_from_queue(), and TAO_Notify_Consumer().

---

Member Enumeration Documentation

enum TAO_Notify_Consumer::DispatchStatus

Status returned from dispatch attempts. Enumeration values: DISPATCH_SUCCESS  DISPATCH_RETRY  DISPATCH_DISCARD  DISPATCH_FAIL  Definition at line 51 of file Consumer.h. Referenced by deliver(), dispatch_batch(), dispatch_from_queue(), and dispatch_request(). { DISPATCH_SUCCESS, DISPATCH_RETRY, // retry this message DISPATCH_DISCARD, // discard this message DISPATCH_FAIL}; // discard all messages and disconnect consumer

---

Constructor & Destructor Documentation

TAO_BEGIN_VERSIONED_NAMESPACE_DECL TAO_Notify_Consumer::TAO_Notify_Consumer (  TAO_Notify_ProxySupplier *  proxy  )

Constuctor. Definition at line 31 of file Consumer.cpp. References ACE_NEW, pending_events(), pending_events_, Request_Queue, and ACE_Auto_Basic_Ptr< X >::reset(). : proxy_ (proxy) , is_suspended_ (0) , pacing_ (proxy->qos_properties_.pacing_interval ()) , max_batch_size_ (CosNotification::MaximumBatchSize, 0) , timer_id_ (-1) , timer_ (0) { Request_Queue* pending_events = 0; ACE_NEW (pending_events, TAO_Notify_Consumer::Request_Queue ()); this->pending_events_.reset( pending_events ); this->timer_.reset( this->proxy ()->timer () ); }

TAO_Notify_Consumer::~TAO_Notify_Consumer (   )  [virtual]

Destructor. Definition at line 46 of file Consumer.cpp. References cancel_timer(). { if (this->timer_.isSet()) { this->cancel_timer (); this->timer_.reset (); } }

---

Member Function Documentation

void TAO_Notify_Consumer::cancel_timer (  void   )  [protected]

Cancel timer. Definition at line 638 of file Consumer.cpp. References ACE_DEBUG, ACE_TEXT(), DEBUG_LEVEL, and LM_DEBUG. Referenced by shutdown(), and ~TAO_Notify_Consumer(). { if (this->timer_.isSet() && this->timer_id_ != -1) { if (DEBUG_LEVEL > 5) ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("Consumer %d canceling dispatch timer.\n"), static_cast<int> (this->proxy ()->id ()) )); this->timer_->cancel_timer (timer_id_); } this->timer_id_ = -1; }

void TAO_Notify_Consumer::deliver (  TAO_Notify_Method_Request_Event *  request  )

Dispatch Event to consumer. Definition at line 150 of file Consumer.cpp. References ACE_CATCHANY, ACE_CHECK, ACE_DEBUG, ACE_DECLARE_NEW_ENV, ACE_ENDTRY, ACE_ENV_ARG_PARAMETER, ACE_ENV_SINGLE_ARG_PARAMETER, ACE_TEXT(), ACE_TRY, ACE_TRY_CHECK, DEBUG_LEVEL, TAO_Notify_ProxySupplier::destroy(), DISPATCH_DISCARD, DISPATCH_FAIL, dispatch_request(), DISPATCH_RETRY, DISPATCH_SUCCESS, DispatchStatus, enqueue_if_necessary(), enqueue_request(), LM_DEBUG, proxy_supplier(), and schedule_timer(). Referenced by TAO_Notify_Method_Request_Dispatch::execute_i(). { // Increment reference counts (safely) to prevent this object and its proxy // from being deleted while the push is in progress. TAO_Notify_Proxy::Ptr proxy_guard (this->proxy ()); bool queued = enqueue_if_necessary (request ACE_ENV_ARG_PARAMETER); ACE_CHECK; if (!queued) { DispatchStatus status = this->dispatch_request (request); switch (status) { case DISPATCH_SUCCESS: { request->complete (); break; } case DISPATCH_RETRY: { if (DEBUG_LEVEL > 1) ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("Consumer %d enqueing event %d due ") ACE_TEXT ("to failed dispatch.\n"), static_cast<int> (this->proxy ()->id ()), request->sequence ())); this->enqueue_request (request ACE_ENV_ARG_PARAMETER); ACE_CHECK; this->schedule_timer (true); break; } case DISPATCH_DISCARD: { if (DEBUG_LEVEL > 0) ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%P|%t) Consumer %d: Error during " "direct dispatch. Discarding event:%d.\n"), static_cast<int> (this->proxy ()->id ()), request->sequence () )); request->complete (); break; } case DISPATCH_FAIL: { if (DEBUG_LEVEL > 0) ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%P|%t) Consumer %d: Failed during " "direct dispatch :%d. Discarding event.\n"), static_cast<int> (this->proxy ()->id ()), request->sequence () )); request->complete (); ACE_DECLARE_NEW_ENV; ACE_TRY { this->proxy_supplier ()->destroy (ACE_ENV_SINGLE_ARG_PARAMETER); ACE_TRY_CHECK; } ACE_CATCHANY { // todo is there something meaningful we can do here? ; } ACE_ENDTRY; break; } } } }

TAO_Notify_Consumer::DispatchStatus TAO_Notify_Consumer::dispatch_batch (  const CosNotification::EventBatch &  batch  )

Dispatch the batch of events to the attached consumer. Definition at line 359 of file Consumer.cpp. References ACE_CATCH, ACE_CATCHANY, ACE_DEBUG, ACE_DECLARE_NEW_ENV, ACE_ENDTRY, ACE_ENV_ARG_PARAMETER, ACE_ERROR, ACE_TEXT(), ACE_TRY, ACE_TRY_CHECK, DEBUG_LEVEL, DISPATCH_DISCARD, DISPATCH_FAIL, DISPATCH_RETRY, DISPATCH_SUCCESS, DispatchStatus, CosNotification::EventBatch, LM_ERROR, push(), TAO_INVOCATION_SEND_REQUEST_MINOR_CODE, TAO_POA_DISCARDING, and TAO_POA_HOLDING. Referenced by TAO_Notify_SequencePushConsumer::dispatch_from_queue(). { DispatchStatus result = DISPATCH_SUCCESS; ACE_DECLARE_NEW_ENV; ACE_TRY { this->push (batch ACE_ENV_ARG_PARAMETER); ACE_TRY_CHECK; } ACE_CATCH (CORBA::OBJECT_NOT_EXIST, ex) { if (DEBUG_LEVEL > 0) ACE_DEBUG ((LM_ERROR, ACE_TEXT ("(%P|%t) TAO_Notify_Consumer " "%d::dispatch_batch() %s\n"), static_cast<int> (this->proxy ()->id ()), ex._info ().c_str () )); result = DISPATCH_FAIL; } ACE_CATCH (CORBA::TRANSIENT, ex) { if (DEBUG_LEVEL > 0) ACE_DEBUG ((LM_ERROR, ACE_TEXT ("(%P|%t) TAO_Notify_Consumer " "%d::dispatch_batch() Transient (minor=%d) %s\n"), static_cast<int> (this->proxy ()->id ()), ex.minor (), ex._info ().c_str () )); const CORBA::ULong BITS_5_THRU_12_MASK = 0x00000f80u; switch (ex.minor () & 0xfffff000u) { case CORBA::OMGVMCID: switch (ex.minor () & 0x00000fffu) { case 2: // No usable profile case 3: // Request cancelled case 4: // POA destroyed result = DISPATCH_FAIL; break; default: result = DISPATCH_DISCARD; } break; case TAO::VMCID: default: switch (ex.minor () & BITS_5_THRU_12_MASK) { case TAO_INVOCATION_SEND_REQUEST_MINOR_CODE: result = DISPATCH_FAIL; break; case TAO_POA_DISCARDING: case TAO_POA_HOLDING: default: result = DISPATCH_RETRY; } break; } } ACE_CATCH (CORBA::TIMEOUT, ex) { if (DEBUG_LEVEL > 0) ACE_DEBUG ((LM_ERROR, ACE_TEXT ("(%P|%t) TAO_Notify_Consumer " "%u::dispatch_batch() %s\n"), this->proxy ()->id (), ex._info().c_str () )); result = DISPATCH_FAIL; } ACE_CATCH (CORBA::COMM_FAILURE, ex) { if (DEBUG_LEVEL > 0) ACE_DEBUG ((LM_ERROR, ACE_TEXT ("(%P|%t) TAO_Notify_Consumer " "%u::dispatch_batch() %s\n"), this->proxy ()->id (), ex._info().c_str () )); result = DISPATCH_FAIL; } ACE_CATCH (CORBA::SystemException, ex) { if (DEBUG_LEVEL > 0) { ACE_DEBUG ((LM_ERROR, ACE_TEXT ("(%P|%t) TAO_Notify_Consumer " "%d::dispatch_batch() SystemException %s\n"), static_cast<int> (this->proxy ()->id ()), ex._info ().c_str () )); } result = DISPATCH_DISCARD; } ACE_CATCHANY { ACE_ERROR ((LM_ERROR, ACE_TEXT ("(%P|%t) TAO_Notify_Consumer " "%d::dispatch_batch() Caught unexpected " "exception pushing batch to consumer.\n"), static_cast<int> (this->proxy ()->id ()) )); result = DISPATCH_DISCARD; } ACE_ENDTRY; return result; }

bool TAO_Notify_Consumer::dispatch_from_queue (  Request_Queue &  requests, ACE_Guard< TAO_SYNCH_MUTEX > &  ace_mon )  [protected, virtual]

Attempt to dispatch event from a queue. Called by dispatch_pending. Deliver one or more events to the Consumer. If delivery fails, events are left in the queue (or discarded depending on QoS parameters.) Undelivered, undiscarded requests are left at the front of the queue. Overridden in sequence consumer to dispatch as an EventBatch. Returns: false if delivery failed and the request(s) cannot be discarded. Reimplemented in TAO_Notify_SequencePushConsumer. Definition at line 499 of file Consumer.cpp. References ACE_CATCHANY, ACE_DEBUG, ACE_DECLARE_NEW_ENV, ACE_ENDTRY, ACE_ENV_SINGLE_ARG_PARAMETER, ACE_TEXT(), ACE_TRY, ACE_TRY_CHECK, ACE_Guard< ACE_LOCK >::acquire(), TAO_Notify_Method_Request_Event::complete(), DEBUG_LEVEL, ACE_Unbounded_Queue< T >::dequeue_head(), TAO_Notify_ProxySupplier::destroy(), DISPATCH_DISCARD, DISPATCH_FAIL, dispatch_request(), DISPATCH_RETRY, DISPATCH_SUCCESS, DispatchStatus, ACE_Unbounded_Queue< T >::enqueue_head(), LM_DEBUG, proxy_supplier(), ACE_Message_Block::release(), ACE_Guard< ACE_LOCK >::release(), Request_Queue, and TAO_Notify_Method_Request_Event::sequence(). Referenced by dispatch_pending(). { bool result = true; TAO_Notify_Method_Request_Event_Queueable * request; if (requests.dequeue_head (request) == 0) { ace_mon.release (); DispatchStatus status = this->dispatch_request (request); switch (status) { case DISPATCH_SUCCESS: { request->complete (); request->release (); result = true; ace_mon.acquire (); break; } case DISPATCH_RETRY: { if (DEBUG_LEVEL > 0) ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%P|%t) Consumer %d: Will retry %d\n"), static_cast<int> (this->proxy ()->id ()), request->sequence () )); ace_mon.acquire (); requests.enqueue_head (request); // put the failed event back where it was result = false; break; } case DISPATCH_DISCARD: { if (DEBUG_LEVEL > 0) ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%P|%t) Consumer %d: Error during " "dispatch. Discarding event:%d.\n"), static_cast<int> (this->proxy ()->id ()), request->sequence () )); request->complete (); ace_mon.acquire (); result = true; break; } case DISPATCH_FAIL: { if (DEBUG_LEVEL > 0) ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%P|%t) Consumer %d: Failed. " "Discarding event %d.\n"), static_cast<int> (this->proxy ()->id ()), request->sequence () )); request->complete (); ace_mon.acquire (); while (requests.dequeue_head (request) == 0) { ace_mon.release (); request->complete (); ace_mon.acquire (); } ace_mon.release (); ACE_DECLARE_NEW_ENV; ACE_TRY { this->proxy_supplier ()->destroy (ACE_ENV_SINGLE_ARG_PARAMETER); ACE_TRY_CHECK; } ACE_CATCHANY { // todo is there something reasonable to do here? } ACE_ENDTRY; ace_mon.acquire (); result = true; break; } default: { ace_mon.acquire (); result = false; break; } } } return result; }

void TAO_Notify_Consumer::dispatch_pending (   )

Dispatch the pending events. Definition at line 469 of file Consumer.cpp. References ACE_DEBUG, ACE_GUARD, ACE_TEXT(), DEBUG_LEVEL, dispatch_from_queue(), TAO_Notify_Object::has_shutdown(), ACE_Unbounded_Queue< T >::is_empty(), LM_DEBUG, pending_events(), proxy_supplier(), schedule_timer(), ACE_Unbounded_Queue< T >::size(), and TAO_SYNCH_MUTEX. Referenced by TAO_Notify_SequencePushConsumer::enqueue_if_necessary(), handle_timeout(), and resume(). { if (DEBUG_LEVEL > 5) ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("Consumer %d dispatching pending events. Queue size: %d\n"), static_cast<int> (this->proxy ()->id ()), this->pending_events().size () )); // lock ourselves in memory for the duration TAO_Notify_Consumer::Ptr self_grd (this); // dispatch events until: 1) the queue is empty; 2) the proxy shuts down, or 3) the dispatch fails ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, *this->proxy_lock ()); bool ok = true; while (ok && !this->proxy_supplier ()->has_shutdown () && !this->pending_events().is_empty ()) { if (! dispatch_from_queue ( this->pending_events(), ace_mon)) { this->schedule_timer (true); ok = false; } } }

TAO_Notify_Consumer::DispatchStatus TAO_Notify_Consumer::dispatch_request (  TAO_Notify_Method_Request_Event *  request  )  [protected]

Definition at line 222 of file Consumer.cpp. References ACE_CATCH, ACE_CATCHANY, ACE_DEBUG, ACE_DECLARE_NEW_ENV, ACE_ENDTRY, ACE_ENV_ARG_PARAMETER, ACE_ERROR, ACE_TEXT(), ACE_TRY, ACE_TRY_CHECK, DEBUG_LEVEL, DISPATCH_DISCARD, DISPATCH_FAIL, DISPATCH_RETRY, DISPATCH_SUCCESS, DispatchStatus, TAO_Notify_Method_Request_Event::event(), LM_DEBUG, LM_ERROR, TAO_Notify_Event::push(), TAO_Notify_Method_Request_Event::sequence(), TAO_Notify_Method_Request_Event::should_retry(), TAO_INVOCATION_SEND_REQUEST_MINOR_CODE, TAO_POA_DISCARDING, and TAO_POA_HOLDING. Referenced by deliver(), and dispatch_from_queue(). { DispatchStatus result = DISPATCH_SUCCESS; ACE_DECLARE_NEW_ENV; ACE_TRY { request->event ()->push (this ACE_ENV_ARG_PARAMETER); ACE_TRY_CHECK; if (DEBUG_LEVEL > 8) ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("Consumer %d dispatched single event %d.\n"), static_cast<int> (this->proxy ()->id ()), request->sequence () )); } ACE_CATCH (CORBA::OBJECT_NOT_EXIST, ex) { if (DEBUG_LEVEL > 0) { ACE_DEBUG ((LM_ERROR, ACE_TEXT ("(%P|%t) TAO_Notify_Consumer %d::push " "(request) %s\n"), static_cast<int> (this->proxy ()->id ()), ex._info ().c_str () )); } result = DISPATCH_FAIL; } ACE_CATCH (CORBA::TRANSIENT, ex) { if (DEBUG_LEVEL > 0) ACE_DEBUG ((LM_ERROR, ACE_TEXT ("(%P|%t) TAO_Notify_Consumer %d::push " "(request) Transient (minor=%d) %s\n"), static_cast<int> (this->proxy ()->id ()), ex.minor (), ex._info ().c_str () )); const CORBA::ULong BITS_5_THRU_12_MASK = 0x00000f80u; switch (ex.minor () & 0xfffff000u) { case CORBA::OMGVMCID: switch (ex.minor () & 0x00000fffu) { case 2: // No usable profile case 3: // Request cancelled case 4: // POA destroyed result = DISPATCH_FAIL; break; default: result = DISPATCH_DISCARD; } break; case TAO::VMCID: default: switch (ex.minor () & BITS_5_THRU_12_MASK) { case TAO_INVOCATION_SEND_REQUEST_MINOR_CODE: result = DISPATCH_FAIL; break; case TAO_POA_DISCARDING: case TAO_POA_HOLDING: default: result = DISPATCH_RETRY; } break; } } ACE_CATCH (CORBA::TIMEOUT, ex) { if (DEBUG_LEVEL > 0) ACE_DEBUG ((LM_ERROR, ACE_TEXT ("(%P|%t) TAO_Notify_Consumer %u::push " "(request) %s\n"), this->proxy ()->id (), ex._info().c_str () )); result = DISPATCH_FAIL; } ACE_CATCH (CORBA::COMM_FAILURE, ex) { if (DEBUG_LEVEL > 0) ACE_DEBUG ((LM_ERROR, ACE_TEXT ("(%P|%t) TAO_Notify_Consumer %u::push " "(request) %s\n"), this->proxy ()->id (), ex._info().c_str () )); result = DISPATCH_FAIL; } ACE_CATCH (CORBA::SystemException, ex) { if (DEBUG_LEVEL > 0) { ACE_DEBUG ((LM_ERROR, ACE_TEXT ("(%P|%t) TAO_Notify_Consumer %d::push " "(request) SystemException %s\n"), static_cast<int> (this->proxy ()->id ()), ex._info ().c_str () )); } result = DISPATCH_DISCARD; } ACE_CATCHANY { ACE_ERROR ( (LM_ERROR, ACE_TEXT ("(%P|%t) TAO_Notify_Consumer %d::push " "(request) Caught unexpected exception " "pushing event to consumer.\n"), static_cast<int> (this->proxy ()->id ()) )); result = DISPATCH_DISCARD; } ACE_ENDTRY; // for persistent events that haven't timed out // convert "FAIL" & "DISCARD" to "RETRY" // for transient events, convert RETRY to DISCARD (hey, best_effort.) if (result == DISPATCH_FAIL || result == DISPATCH_DISCARD) { if (request->should_retry ()) { result = DISPATCH_RETRY; } } else if (result == DISPATCH_RETRY) { if (! request->should_retry ()) { result = DISPATCH_DISCARD; } } return result; }

void TAO_Notify_Consumer::dispatch_updates_i (  const CosNotification::EventTypeSeq &  added, const CosNotification::EventTypeSeq &  removed )  [protected, virtual]

Implementation of Peer specific dispatch_updates. Implements TAO_Notify_Peer. Definition at line 683 of file Consumer.cpp. References ACE_ENV_ARG_PARAMETER, CosNotification::EventTypeSeq, CORBA::is_nil(), and publish_. { if (!CORBA::is_nil (this->publish_.in ())) this->publish_->offer_change (added, removed ACE_ENV_ARG_PARAMETER); }

bool TAO_Notify_Consumer::enqueue_if_necessary (  TAO_Notify_Method_Request_Event *  request  )  [protected, virtual]

Add request to a queue if necessary. Overridden by sequence consumer to "always" put incoming events into the queue. Returns: true the request has been enqueued; false the request should be handled now. Reimplemented in TAO_Notify_SequencePushConsumer. Definition at line 101 of file Consumer.cpp. References ACE_CHECK_RETURN, ACE_DEBUG, ACE_ENV_SINGLE_ARG_PARAMETER, ACE_GUARD_RETURN, ACE_NEW_THROW_EX, ACE_TEXT(), DEBUG_LEVEL, ACE_Unbounded_Queue< T >::enqueue_tail(), ACE_Unbounded_Queue< T >::is_empty(), is_suspended_, LM_DEBUG, pending_events(), schedule_timer(), and TAO_SYNCH_MUTEX. Referenced by deliver(). { ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, ace_mon, *this->proxy_lock (), false); if (! this->pending_events().is_empty ()) { if (DEBUG_LEVEL > 3) ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("Consumer %d: enqueuing another event. %d\n"), static_cast<int> (this->proxy ()->id ()), request->sequence () )); TAO_Notify_Event::Ptr event ( request->event ()->queueable_copy (ACE_ENV_SINGLE_ARG_PARAMETER)); ACE_CHECK_RETURN (false); TAO_Notify_Method_Request_Event_Queueable * queue_entry; ACE_NEW_THROW_EX (queue_entry, TAO_Notify_Method_Request_Event_Queueable (*request, event), CORBA::NO_MEMORY ()); ACE_CHECK_RETURN (false); this->pending_events().enqueue_tail (queue_entry); this->schedule_timer (false); return true; } if (this->is_suspended_ == 1) { if (DEBUG_LEVEL > 3) ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("Suspended Consumer %d enqueing event. %d\n"), static_cast<int> (this->proxy ()->id ()), request->sequence () )); TAO_Notify_Event::Ptr event ( request->event ()->queueable_copy (ACE_ENV_SINGLE_ARG_PARAMETER)); ACE_CHECK_RETURN (false); TAO_Notify_Method_Request_Event_Queueable * queue_entry; ACE_NEW_THROW_EX (queue_entry, TAO_Notify_Method_Request_Event_Queueable (*request, event), CORBA::NO_MEMORY ()); ACE_CHECK_RETURN (false); this->pending_events().enqueue_tail (queue_entry); this->schedule_timer (false); return true; } return false; }

void TAO_Notify_Consumer::enqueue_request (  TAO_Notify_Method_Request_Event *  request  )  [protected]

Definition at line 76 of file Consumer.cpp. References ACE_CHECK, ACE_DEBUG, ACE_ENV_SINGLE_ARG_PARAMETER, ACE_GUARD, ACE_NEW_THROW_EX, ACE_TEXT(), DEBUG_LEVEL, ACE_Unbounded_Queue< T >::enqueue_tail(), LM_DEBUG, pending_events(), TAO_Notify_Method_Request_Event::sequence(), and TAO_SYNCH_MUTEX. Referenced by deliver(), and TAO_Notify_SequencePushConsumer::enqueue_if_necessary(). { TAO_Notify_Event::Ptr event ( request->event ()->queueable_copy (ACE_ENV_SINGLE_ARG_PARAMETER)); ACE_CHECK; TAO_Notify_Method_Request_Event_Queueable * queue_entry; ACE_NEW_THROW_EX (queue_entry, TAO_Notify_Method_Request_Event_Queueable (*request, event), CORBA::NO_MEMORY ()); ACE_CHECK; if (DEBUG_LEVEL > 3) ACE_DEBUG ( (LM_DEBUG, ACE_TEXT ("Consumer %d: enqueue_request (%d) @%@.\n"), static_cast<int> (this->proxy ()->id ()), request->sequence (), request )); ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, *this->proxy_lock ()); this->pending_events().enqueue_tail (queue_entry); }

int TAO_Notify_Consumer::handle_timeout (  const ACE_Time_Value &  current_time, const void *  act = 0 )  [protected, virtual]

Reimplemented from ACE_Event_Handler. Definition at line 654 of file Consumer.cpp. References ACE_CATCHALL, ACE_DECLARE_NEW_ENV, ACE_ENDTRY, ACE_ENV_SINGLE_ARG_PARAMETER, ACE_TRY, ACE_TRY_CHECK, and dispatch_pending(). { TAO_Notify_Consumer::Ptr grd (this); this->timer_id_ = -1; // This must come first, because dispatch_pending may try to resched ACE_DECLARE_NEW_ENV; ACE_TRY { this->dispatch_pending (ACE_ENV_SINGLE_ARG_PARAMETER); ACE_TRY_CHECK; } ACE_CATCHALL { } ACE_ENDTRY; return 0; }

ACE_INLINE CORBA::Boolean TAO_Notify_Consumer::is_suspended (  void   )

Is the connection suspended? Definition at line 16 of file Consumer.inl. References is_suspended_. Referenced by schedule_timer(). { return this->is_suspended_; }

TAO_BEGIN_VERSIONED_NAMESPACE_DECL ACE_INLINE TAO_Notify_Consumer::Request_Queue & TAO_Notify_Consumer::pending_events (   )  [protected]

= Protected Data Members Definition at line 9 of file Consumer.inl. References ACE_ASSERT, ACE_Auto_Basic_Ptr< X >::get(), and pending_events_. Referenced by dispatch_pending(), TAO_Notify_SequencePushConsumer::enqueue_if_necessary(), enqueue_if_necessary(), enqueue_request(), and TAO_Notify_Consumer(). { ACE_ASSERT( pending_events_.get() != 0 ); return *pending_events_; }

TAO_Notify_Proxy * TAO_Notify_Consumer::proxy (  void   )  [virtual]

Access Base Proxy. Implements TAO_Notify_Peer. Definition at line 56 of file Consumer.cpp. References proxy_supplier(). { return this->proxy_supplier (); }

TAO_SYNCH_MUTEX * TAO_Notify_Consumer::proxy_lock (  void   )  [protected]

Get the shared Proxy Lock. Definition at line 691 of file Consumer.cpp. References TAO_Notify_Object::lock_. { return &this->proxy_->lock_; }

TAO_Notify_ProxySupplier * TAO_Notify_Consumer::proxy_supplier (  void   )

Access Specific Proxy. Definition at line 697 of file Consumer.cpp. Referenced by deliver(), TAO_Notify_SequencePushConsumer::dispatch_from_queue(), dispatch_from_queue(), dispatch_pending(), and proxy(). { return this->proxy_; }

virtual void TAO_Notify_Consumer::push (  const CosNotification::EventBatch &  event  )  [pure virtual]

Push a batch of events to this consumer. Implemented in TAO_Notify_PushConsumer, TAO_Notify_SequencePushConsumer, and TAO_Notify_StructuredPushConsumer.

virtual void TAO_Notify_Consumer::push (  const CosNotification::StructuredEvent &  event  )  [pure virtual]

Push to this consumer. Implemented in TAO_Notify_PushConsumer, TAO_Notify_SequencePushConsumer, and TAO_Notify_StructuredPushConsumer.

virtual void TAO_Notify_Consumer::push (  const CORBA::Any &  event  )  [pure virtual]

Push to this consumer. Implemented in TAO_Notify_PushConsumer, TAO_Notify_SequencePushConsumer, and TAO_Notify_StructuredPushConsumer. Referenced by dispatch_batch().

void TAO_Notify_Consumer::qos_changed (  const TAO_Notify_QoSProperties &  qos_properties  )  [virtual]

Override, Peer::qos_changed. Reimplemented from TAO_Notify_Peer. Definition at line 62 of file Consumer.cpp. References max_batch_size_, and TAO_Notify_QoSProperties::maximum_batch_size(). { this->max_batch_size_ = qos_properties.maximum_batch_size (); }

virtual void TAO_Notify_Consumer::reconnect_from_consumer (  TAO_Notify_Consumer *  old_consumer  )  [pure virtual]

on reconnect we need to move events from the old consumer to the new one Implemented in TAO_Notify_PushConsumer, TAO_Notify_SequencePushConsumer, and TAO_Notify_StructuredPushConsumer.

void TAO_Notify_Consumer::resume (   )

Resume Connection. Definition at line 68 of file Consumer.cpp. References ACE_ENV_SINGLE_ARG_PARAMETER, dispatch_pending(), and is_suspended_. { this->is_suspended_ = 0; this->dispatch_pending (ACE_ENV_SINGLE_ARG_PARAMETER); }

void TAO_Notify_Consumer::schedule_timer (  bool  is_error = false  )  [protected]

Schedule timer. Definition at line 591 of file Consumer.cpp. References ACE_ASSERT, ACE_DEBUG, ACE_ERROR, ACE_TEXT(), DEBUG_LEVEL, DEFAULT_RETRY_TIMEOUT, is_suspended(), TAO_Notify_PropertyBase_T< TYPE >::is_valid(), LM_DEBUG, LM_ERROR, ACE_Time_Value::msec(), and pacing_. Referenced by deliver(), dispatch_pending(), TAO_Notify_SequencePushConsumer::enqueue_if_necessary(), enqueue_if_necessary(), TAO_Notify_StructuredPushConsumer::reconnect_from_consumer(), TAO_Notify_SequencePushConsumer::reconnect_from_consumer(), and TAO_Notify_PushConsumer::reconnect_from_consumer(). { if (this->timer_id_ != -1) { return; // We only want a single timeout scheduled. } // don't schedule timer if there's nothing that can be done if (this->is_suspended ()) { return; } ACE_ASSERT (this->timer_.get() != 0); // If we're scheduling the timer due to an error then we want to // use the retry timeout, otherwise we'll assume that the pacing // interval is sufficient for now. ACE_Time_Value tv (DEFAULT_RETRY_TIMEOUT); if (! is_error) { if (this->pacing_.is_valid ()) { tv = ORBSVCS_Time::to_Time_Value (this->pacing_.value ()); } } if (DEBUG_LEVEL > 5) { ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("Consumer %d: scheduling pacing/retry for %dms.\n"), static_cast<int> (this->proxy ()->id ()), tv.msec ())); } this->timer_id_ = this->timer_->schedule_timer (this, tv, ACE_Time_Value::zero); if (this->timer_id_ == -1) { ACE_ERROR ((LM_ERROR, ACE_TEXT ("TAO_Notify_Consumer %d::schedule_timer () " "Error scheduling timer.\n"), static_cast<int> (this->proxy ()->id ()) )); } }

void TAO_Notify_Consumer::shutdown (   )  [virtual]

Shutdown the consumer. Reimplemented from TAO_Notify_Peer. Definition at line 673 of file Consumer.cpp. References cancel_timer(). { if (this->timer_.isSet ()) { this->cancel_timer (); this->timer_.reset (); } }

void TAO_Notify_Consumer::suspend (   )

Suspend Connection.

---

Member Data Documentation

CORBA::Boolean TAO_Notify_Consumer::is_suspended_ [protected]

Suspended Flag. Definition at line 164 of file Consumer.h. Referenced by enqueue_if_necessary(), is_suspended(), and resume().

TAO_Notify_Property_Long TAO_Notify_Consumer::max_batch_size_ [protected]

Max. batch size. Definition at line 173 of file Consumer.h. Referenced by qos_changed().

const TAO_Notify_Property_Time& TAO_Notify_Consumer::pacing_ [protected]

The Pacing Interval. Definition at line 170 of file Consumer.h. Referenced by schedule_timer().

ACE_Auto_Ptr< Request_Queue > TAO_Notify_Consumer::pending_events_ [private]

Events pending to be delivered. Definition at line 184 of file Consumer.h. Referenced by pending_events(), and TAO_Notify_Consumer().

TAO_Notify_ProxySupplier* TAO_Notify_Consumer::proxy_ [protected]

The Proxy that we associate with. Definition at line 161 of file Consumer.h.

CosNotifyComm::NotifyPublish_var TAO_Notify_Consumer::publish_ [protected]

Interface that accepts offer_changes. Definition at line 167 of file Consumer.h. Referenced by dispatch_updates_i().

TAO_Notify_Timer::Ptr TAO_Notify_Consumer::timer_ [protected]

The Timer Manager that we use. Definition at line 179 of file Consumer.h.

long TAO_Notify_Consumer::timer_id_ [protected]

Timer Id. Definition at line 176 of file Consumer.h.

---

The documentation for this class was generated from the following files:

• Consumer.h
• Consumer.cpp
• Consumer.inl

---