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
	DISPATCH_SUCCESS
	DISPATCH_RETRY
	DISPATCH_DISCARD
	DISPATCH_FAIL
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)
	Constructor.
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 event to this consumer.
virtual void	push (const CosNotification::StructuredEvent &event)=0
	Push event 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 (void)
	Dispatch the pending events.
CORBA::Boolean	is_suspended (void)
	Is the connection suspended?
void	suspend (void)
	Suspend Connection.
void	resume (void)
	Resume Connection.
virtual void	shutdown (void)
	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.
void	assume_pending_events (TAO_Notify_Consumer &rhs)
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.
bool	have_not_yet_verified_publish_
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 113 of file Consumer.h.

---

Member Enumeration Documentation

enum TAO_Notify_Consumer::DispatchStatus

Status returned from dispatch attempts.

Enumerator:

DISPATCH_SUCCESS
DISPATCH_RETRY
DISPATCH_DISCARD
DISPATCH_FAIL

Definition at line 51 of file Consumer.h.

                      {
    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

)

Constructor.

Definition at line 31 of file Consumer.cpp.

References ACE_NEW, pending_events(), pending_events_, TAO_Notify_Refcountable_Guard_T< T >::reset(), and timer_.

: proxy_ (proxy)
, is_suspended_ (0)
, have_not_yet_verified_publish_ (true)
, 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 47 of file Consumer.cpp.

References TAO_Notify_Refcountable_Guard_T< T >::reset(), and timer_.

{
  if (this->timer_.isSet())
  {
    this->cancel_timer ();
    this->timer_.reset ();
  }
}

---

Member Function Documentation

void TAO_Notify_Consumer::assume_pending_events

(

TAO_Notify_Consumer &

rhs

)

Take the pending queue from the rhs, cancel it's timer and schedule our timer. The caller should have locked the proxy lock before calling this method.

Definition at line 684 of file Consumer.cpp.

References cancel_timer(), ACE_Unbounded_Queue< T >::is_empty(), TAO_Notify_Refcountable_Guard_T< T >::isSet(), pending_events(), pending_events_, schedule_timer(), and timer_.

{
  // No need to lock the this proxy's lock.  It should have been locked
  // by the caller.

  // If the original consumer has pending events
  if (!rhs.pending_events ().is_empty ())
    {
      // We will take them away and cancel it's timer
      this->pending_events_.reset (rhs.pending_events_.release ());
      if (rhs.timer_.isSet ())
        {
          rhs.cancel_timer ();
        }

      // Schedule a new timer for us, which will use the default
      // timer value (unless we have a valid pacing interval).
      this->schedule_timer ();
    }
}

void TAO_Notify_Consumer::cancel_timer

(

void

)

[protected]

Cancel timer.

Definition at line 617 of file Consumer.cpp.

References ACE_DEBUG, ACE_TEXT(), DEBUG_LEVEL, LM_DEBUG, timer_, and timer_id_.

Referenced by assume_pending_events().

{
  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 144 of file Consumer.cpp.

References ACE_DEBUG, ACE_TEXT(), TAO_Notify_Method_Request_Event::complete(), DEBUG_LEVEL, TAO_Notify_ProxySupplier::destroy(), DISPATCH_DISCARD, DISPATCH_FAIL, dispatch_request(), DISPATCH_RETRY, DISPATCH_SUCCESS, enqueue_if_necessary(), enqueue_request(), LM_DEBUG, proxy_supplier(), schedule_timer(), and TAO_Notify_Method_Request_Event::sequence().

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);
  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);
            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 ();
            try
              {
                this->proxy_supplier ()->destroy ();
              }
            catch (const CORBA::Exception&)
              {
                // todo is there something meaningful we can do here?
                ;
              }
            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 344 of file Consumer.cpp.

References CORBA::SystemException::_info(), ACE_DEBUG, ACE_ERROR, ACE_TEXT(), ACE_String_Base< CHAR >::c_str(), DEBUG_LEVEL, DISPATCH_DISCARD, DISPATCH_FAIL, DISPATCH_RETRY, DISPATCH_SUCCESS, LM_ERROR, CORBA::OMGVMCID, push(), TAO_INVOCATION_SEND_REQUEST_MINOR_CODE, TAO_POA_DISCARDING, TAO_POA_HOLDING, and TAO::VMCID.

Referenced by TAO_Notify_SequencePushConsumer::dispatch_from_queue().

{
  DispatchStatus result = DISPATCH_SUCCESS;
  try
    {
      this->push (batch);
    }
  catch (const 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;
    }
  catch (const 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;
        }
    }
  catch (const 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;
    }
  catch (const 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;
    }
  catch (const 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;
    }
  catch (const CORBA::Exception&)
    {
      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;
    }
  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.

Definition at line 481 of file Consumer.cpp.

References ACE_DEBUG, ACE_TEXT(), 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, ACE_Unbounded_Queue< T >::enqueue_head(), LM_DEBUG, proxy_supplier(), ACE_Message_Block::release(), ACE_Guard< ACE_LOCK >::release(), 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 ();
            try
              {
                this->proxy_supplier ()->destroy ();
              }
            catch (const CORBA::Exception&)
              {
                // todo is there something reasonable to do here?
              }
            ace_mon.acquire ();
            result = true;
            break;
          }
        default:
          {
            ace_mon.acquire ();
            result = false;
            break;
          }
        }
    }
  return result;
}

void TAO_Notify_Consumer::dispatch_pending

(

void

)

Dispatch the pending events.

Definition at line 451 of file Consumer.cpp.

References ACE_DEBUG, ACE_GUARD, ACE_TEXT(), DEBUG_LEVEL, dispatch_from_queue(), ACE_Unbounded_Queue< T >::is_empty(), LM_DEBUG, pending_events(), schedule_timer(), and TAO_SYNCH_MUTEX.

Referenced by 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 210 of file Consumer.cpp.

References CORBA::SystemException::_info(), ACE_DEBUG, ACE_ERROR, ACE_TEXT(), ACE_String_Base< CHAR >::c_str(), DEBUG_LEVEL, DISPATCH_DISCARD, DISPATCH_FAIL, DISPATCH_RETRY, DISPATCH_SUCCESS, TAO_Notify_Method_Request_Event::event(), LM_DEBUG, LM_ERROR, CORBA::OMGVMCID, 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, TAO_POA_HOLDING, and TAO::VMCID.

Referenced by deliver(), and dispatch_from_queue().

{
  DispatchStatus result = DISPATCH_SUCCESS;
  try
    {
      request->event ()->push (this);
      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 ()
                    ));
    }
  catch (const 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;
    }
  catch (const 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;
        }
    }
  catch (const 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;
    }
  catch (const 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;
    }
  catch (const 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;
    }
  catch (const CORBA::Exception&)
    {
      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;
    }

  // 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 659 of file Consumer.cpp.

References have_not_yet_verified_publish_, CORBA::is_nil(), and publish_.

{
  if (this->have_not_yet_verified_publish_)
    {
      this->have_not_yet_verified_publish_ = false; // no need to check again
      if (! this->publish_->_is_a ("IDL:omg.org/CosNotifyComm/NotifyPublish:1.0"))
        this->publish_ = CosNotifyComm::NotifyPublish::_nil();
    }
  if (! CORBA::is_nil (this->publish_.in ()))
    this->publish_->offer_change (added, removed);
}

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 99 of file Consumer.cpp.

References ACE_DEBUG, ACE_GUARD_RETURN, ACE_NEW_THROW_EX, ACE_TEXT(), DEBUG_LEVEL, ACE_Unbounded_Queue< T >::enqueue_tail(), TAO_Notify_Method_Request_Event::event(), LM_DEBUG, pending_events(), TAO_Notify_Event::queueable_copy(), schedule_timer(), TAO_Notify_Method_Request_Event::sequence(), 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 ());
      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 ());
      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 ());
      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 ());
      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 77 of file Consumer.cpp.

References ACE_DEBUG, ACE_GUARD, ACE_NEW_THROW_EX, ACE_TEXT(), DEBUG_LEVEL, ACE_Unbounded_Queue< T >::enqueue_tail(), TAO_Notify_Method_Request_Event::event(), LM_DEBUG, pending_events(), TAO_Notify_Event::queueable_copy(), 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 ());

  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 ());

  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 633 of file Consumer.cpp.

References dispatch_pending(), and timer_id_.

{
  TAO_Notify_Consumer::Ptr grd (this);
  this->timer_id_ = -1;  // This must come first, because dispatch_pending may try to resched
  try
    {
      this->dispatch_pending ();
    }
  catch (...)
    {
    }

  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_.

{
  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, and pending_events_.

Referenced by assume_pending_events(), dispatch_pending(), 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 57 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 672 of file Consumer.cpp.

References TAO_Notify_Object::lock_, and proxy_.

{
  return &this->proxy_->lock_;
}

TAO_Notify_ProxySupplier * TAO_Notify_Consumer::proxy_supplier

(

void

)

Access Specific Proxy.

Definition at line 678 of file Consumer.cpp.

References proxy_.

Referenced by deliver(), TAO_Notify_SequencePushConsumer::dispatch_from_queue(), dispatch_from_queue(), 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 event 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 event to this consumer.

Implemented in TAO_Notify_PushConsumer, TAO_Notify_SequencePushConsumer, and TAO_Notify_StructuredPushConsumer.

Referenced by dispatch_batch(), TAO_Notify_StructuredEvent_No_Copy::push(), and TAO_Notify_AnyEvent_No_Copy::push().

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 63 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

(

void

)

Resume Connection.

Definition at line 69 of file Consumer.cpp.

References dispatch_pending(), and is_suspended_.

Referenced by TAO_Notify_ProxySupplier_T< SERVANT_TYPE >::resume_connection().

{
  this->is_suspended_ = 0;

  this->dispatch_pending ();
}

void TAO_Notify_Consumer::schedule_timer

(

bool

is_error = false

)

[protected]

Schedule timer.

Definition at line 570 of file Consumer.cpp.

References ACE_ASSERT, ACE_DEBUG, ACE_ERROR, ACE_TEXT(), DEBUG_LEVEL, LM_DEBUG, LM_ERROR, timer_, timer_id_, and ACE_Time_Value::zero.

Referenced by assume_pending_events(), 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

(

void

)

[virtual]

Shutdown the consumer.

Reimplemented from TAO_Notify_Peer.

Definition at line 649 of file Consumer.cpp.

References TAO_Notify_Refcountable_Guard_T< T >::reset(), and timer_.

{
  if (this->timer_.isSet ())
    {
      this->cancel_timer ();
      this->timer_.reset ();
    }
}

ACE_INLINE void TAO_Notify_Consumer::suspend

(

void

)

Suspend Connection.

Definition at line 22 of file Consumer.inl.

References is_suspended_.

Referenced by TAO_Notify_ProxySupplier_T< SERVANT_TYPE >::suspend_connection().

{
  this->is_suspended_ = 1;
}

---

Member Data Documentation

bool TAO_Notify_Consumer::have_not_yet_verified_publish_ [protected]

Definition at line 169 of file Consumer.h.

Referenced by dispatch_updates_i().

CORBA::Boolean TAO_Notify_Consumer::is_suspended_ [protected]

Suspended Flag.

Definition at line 165 of file Consumer.h.

Referenced by is_suspended(), resume(), and suspend().

TAO_Notify_Property_Long TAO_Notify_Consumer::max_batch_size_ [protected]

Max. batch size.

Definition at line 175 of file Consumer.h.

Referenced by TAO_Notify_SequencePushConsumer::enqueue_if_necessary(), and qos_changed().

const TAO_Notify_Property_Time& TAO_Notify_Consumer::pacing_ [protected]

The Pacing Interval.

Definition at line 172 of file Consumer.h.

Referenced by TAO_Notify_SequencePushConsumer::enqueue_if_necessary().

ACE_Auto_Ptr< Request_Queue > TAO_Notify_Consumer::pending_events_ [private]

Events pending to be delivered.

Definition at line 186 of file Consumer.h.

Referenced by assume_pending_events(), pending_events(), and TAO_Notify_Consumer().

TAO_Notify_ProxySupplier* TAO_Notify_Consumer::proxy_ [protected]

The Proxy that we associate with.

Definition at line 162 of file Consumer.h.

Referenced by proxy_lock(), and proxy_supplier().

CosNotifyComm::NotifyPublish_var TAO_Notify_Consumer::publish_ [protected]

Interface that accepts offer_changes.

Definition at line 168 of file Consumer.h.

Referenced by dispatch_updates_i(), TAO_Notify_StructuredPushConsumer::init(), TAO_Notify_SequencePushConsumer::init(), and TAO_Notify_PushConsumer::init().

TAO_Notify_Timer::Ptr TAO_Notify_Consumer::timer_ [protected]

The Timer Manager that we use.

Definition at line 181 of file Consumer.h.

Referenced by assume_pending_events(), cancel_timer(), schedule_timer(), shutdown(), TAO_Notify_Consumer(), and ~TAO_Notify_Consumer().

long TAO_Notify_Consumer::timer_id_ [protected]

Timer Id.

Definition at line 178 of file Consumer.h.

Referenced by cancel_timer(), handle_timeout(), and schedule_timer().

---

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

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

---