OpenDDS::DCPS::MessageTracker Class Reference

#include <MessageTracker.h>

Collaboration diagram for OpenDDS::DCPS::MessageTracker:

Public Member Functions
	MessageTracker (const OPENDDS_STRING &msg_src)
void	message_sent ()
void	message_delivered ()
void	message_dropped ()
bool	pending_messages () const
void	wait_messages_pending (const char *caller)
void	wait_messages_pending (const char *caller, const MonotonicTimePoint &deadline)
int	dropped_count () const

Private Types
typedef ConditionVariable< ACE_Thread_Mutex >	ConditionVariableType
	All messages have been transported condition variable. More...

Private Member Functions
bool	pending_messages_i () const

Private Attributes
const OPENDDS_STRING	msg_src_
int	dropped_count_
int	delivered_count_
int	sent_count_
ACE_Thread_Mutex	lock_
ConditionVariableType	done_condition_

Detailed Description

A simple message tracker to use to wait until all messages have been accounted for being continuing processing.

Definition at line 27 of file MessageTracker.h.

Member Typedef Documentation

ConditionVariableType

typedef ConditionVariable<ACE_Thread_Mutex> OpenDDS::DCPS::MessageTracker::ConditionVariableType

private

All messages have been transported condition variable.

Definition at line 79 of file MessageTracker.h.

Constructor & Destructor Documentation

MessageTracker()

OpenDDS::DCPS::MessageTracker::MessageTracker

(

const OPENDDS_STRING &

msg_src

)

Definition at line 21 of file MessageTracker.cpp.

: msg_src_(msg_src)
, dropped_count_(0)
, delivered_count_(0)
, sent_count_(0)
, done_condition_(lock_)
{
}

Member Function Documentation

dropped_count()

int OpenDDS::DCPS::MessageTracker::dropped_count

(

)

const

For testing.

Definition at line 129 of file MessageTracker.cpp.

References ACE_GUARD_RETURN, dropped_count_, lock_, and OPENDDS_END_VERSIONED_NAMESPACE_DECL.

{
  ACE_GUARD_RETURN(ACE_Thread_Mutex, guard, lock_, 0);
  return dropped_count_;
}

message_delivered()

void OpenDDS::DCPS::MessageTracker::message_delivered

(

)

Indicate that a message has been delivered by the transport layer.

Definition at line 51 of file MessageTracker.cpp.

References ACE_GUARD, delivered_count_, done_condition_, lock_, OpenDDS::DCPS::ConditionVariable< Mutex >::notify_all(), and pending_messages_i().

Referenced by OpenDDS::DCPS::SendResponseListener::control_delivered(), OpenDDS::DCPS::DataWriterImpl::control_delivered(), OpenDDS::DCPS::SendResponseListener::data_delivered(), and OpenDDS::DCPS::WriteDataContainer::data_delivered().

{
  ACE_GUARD(ACE_Thread_Mutex, guard, lock_);
  ++delivered_count_;

  if (!pending_messages_i()) {
    done_condition_.notify_all();
  }
}

message_dropped()

void OpenDDS::DCPS::MessageTracker::message_dropped

(

)

Indicate that a message has been dropped by the transport layer.

Definition at line 62 of file MessageTracker.cpp.

References ACE_GUARD, done_condition_, dropped_count_, lock_, OpenDDS::DCPS::ConditionVariable< Mutex >::notify_all(), and pending_messages_i().

Referenced by OpenDDS::DCPS::DataWriterImpl::association_complete_i(), OpenDDS::DCPS::SendResponseListener::control_dropped(), OpenDDS::DCPS::DataWriterImpl::control_dropped(), OpenDDS::DCPS::SendResponseListener::data_dropped(), OpenDDS::DCPS::WriteDataContainer::data_dropped(), OpenDDS::DCPS::DataWriterImpl::replay_durable_data_for(), and OpenDDS::DCPS::DataWriterImpl::send_control().

{
  ACE_GUARD(ACE_Thread_Mutex, guard, lock_);
  ++dropped_count_;

  if (!pending_messages_i()) {
    done_condition_.notify_all();
  }
}

message_sent()

void OpenDDS::DCPS::MessageTracker::message_sent

(

)

Indicate that a message has been to the transport layer.

Definition at line 44 of file MessageTracker.cpp.

References ACE_GUARD, lock_, and sent_count_.

Referenced by OpenDDS::DCPS::DataWriterImpl::association_complete_i(), OpenDDS::DCPS::DataWriterImpl::replay_durable_data_for(), OpenDDS::DCPS::DataWriterImpl::send_all_to_flush_control(), OpenDDS::DCPS::DataWriterImpl::send_control(), and OpenDDS::DCPS::SendResponseListener::track_message().

{
  ACE_GUARD(ACE_Thread_Mutex, guard, lock_);
  ++sent_count_;
}

pending_messages()

bool OpenDDS::DCPS::MessageTracker::pending_messages

(

)

const

Answer if there are any messages that have not been accounted for.

Definition at line 31 of file MessageTracker.cpp.

References ACE_GUARD_RETURN, lock_, and pending_messages_i().

{
  ACE_GUARD_RETURN(ACE_Thread_Mutex, guard, lock_, false);
  return pending_messages_i();
}

pending_messages_i()

bool OpenDDS::DCPS::MessageTracker::pending_messages_i ( ) const

private

Definition at line 38 of file MessageTracker.cpp.

References delivered_count_, dropped_count_, and sent_count_.

Referenced by message_delivered(), message_dropped(), pending_messages(), and wait_messages_pending().

{
  return sent_count_ > delivered_count_ + dropped_count_;
}

wait_messages_pending() [1/2]

void OpenDDS::DCPS::MessageTracker::wait_messages_pending

(

const char *

caller

)

Block until all messages have been accounted for or timeouts out based on PendingTimeout.

Definition at line 72 of file MessageTracker.cpp.

References OpenDDS::DCPS::TimePoint_T< MonotonicClock >::now(), and TheServiceParticipant.

Referenced by OpenDDS::DCPS::DataWriterImpl::wait_pending(), and OpenDDS::DCPS::SendResponseListener::~SendResponseListener().

{
  const TimeDuration pending_timeout(TheServiceParticipant->pending_timeout());
  wait_messages_pending(caller, pending_timeout.is_zero() ?
    MonotonicTimePoint() : MonotonicTimePoint::now() + pending_timeout);
}

wait_messages_pending() [2/2]

void OpenDDS::DCPS::MessageTracker::wait_messages_pending	(	const char *	caller,
		const MonotonicTimePoint &	deadline
	)

Block until all messages have been accounted for or the deadline supplied has passed. Blocks indefinitely if deadline is zero.

Definition at line 79 of file MessageTracker.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_GUARD, ACE_TEXT(), OpenDDS::DCPS::CvStatus_Error, OpenDDS::DCPS::CvStatus_NoTimeout, OpenDDS::DCPS::CvStatus_Timeout, OpenDDS::DCPS::DCPS_debug_level, done_condition_, OpenDDS::DCPS::TimePoint_T< AceClock >::is_zero(), LM_DEBUG, LM_ERROR, lock_, msg_src_, pending_messages_i(), TheServiceParticipant, OpenDDS::DCPS::TimePoint_T< AceClock >::value(), and OpenDDS::DCPS::ConditionVariable< Mutex >::wait_until().

{
  const bool use_deadline = deadline.is_zero();
  ACE_GUARD(ACE_Thread_Mutex, guard, this->lock_);
  const bool report = DCPS_debug_level > 0 && pending_messages_i();
  if (report) {
    if (use_deadline) {
      ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) MessageTracker::wait_messages_pending ")
                ACE_TEXT("from source=%C will wait until %#T.\n"),
                msg_src_.c_str(), &deadline.value()));
    } else {
      ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) MessageTracker::wait_messages_pending ")
                ACE_TEXT("from source=%C will wait with no timeout.\n"),
                msg_src_.c_str()));
    }
  }
  bool loop = true;
  ThreadStatusManager& thread_status_manager = TheServiceParticipant->get_thread_status_manager();
  while (loop && pending_messages_i()) {
    switch (done_condition_.wait_until(deadline, thread_status_manager)) {
    case CvStatus_Timeout:
      if (DCPS_debug_level && pending_messages_i()) {
        ACE_DEBUG((LM_DEBUG,
                   "(%P|%t) MessageTracker::wait_messages_pending: "
                   "Timed out waiting for messages to be transported (caller: %C)\n",
                   caller));
      }
      loop = false;
      break;

    case CvStatus_NoTimeout:
      break;

    case CvStatus_Error:
      if (DCPS_debug_level) {
        ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: MessageTracker::wait_messages_pending: "
          "error in wait_until\n"));
      }
      loop = false;
      return;
    }
  }
  if (report) {
    ACE_DEBUG((LM_DEBUG, "(%P|%t) MessageTracker::wait_messages_pending %T done\n"));
  }
}

Member Data Documentation

delivered_count_

int OpenDDS::DCPS::MessageTracker::delivered_count_

private

Definition at line 73 of file MessageTracker.h.

Referenced by message_delivered(), and pending_messages_i().

done_condition_

ConditionVariableType OpenDDS::DCPS::MessageTracker::done_condition_

private

Definition at line 80 of file MessageTracker.h.

Referenced by message_delivered(), message_dropped(), and wait_messages_pending().

dropped_count_

int OpenDDS::DCPS::MessageTracker::dropped_count_

private

Definition at line 72 of file MessageTracker.h.

Referenced by dropped_count(), message_dropped(), and pending_messages_i().

lock_

ACE_Thread_Mutex OpenDDS::DCPS::MessageTracker::lock_

mutableprivate

Definition at line 76 of file MessageTracker.h.

Referenced by dropped_count(), message_delivered(), message_dropped(), message_sent(), pending_messages(), and wait_messages_pending().

msg_src_

const OPENDDS_STRING OpenDDS::DCPS::MessageTracker::msg_src_

private

Definition at line 71 of file MessageTracker.h.

Referenced by wait_messages_pending().

sent_count_

int OpenDDS::DCPS::MessageTracker::sent_count_

private

Definition at line 74 of file MessageTracker.h.

Referenced by message_sent(), and pending_messages_i().

---

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

• MessageTracker.h
• MessageTracker.cpp