OpenDDS::DCPS::MessageTracker Class Reference

#include <MessageTracker.h>

List of all members.

Public Member Functions
	MessageTracker (const OPENDDS_STRING &msg_src)
void	message_sent ()
void	message_delivered ()
void	message_dropped ()
bool	pending_messages ()
void	wait_messages_pending (OPENDDS_STRING &caller_message)
int	dropped_count ()
Static Public Member Functions
static ACE_TCHAR *	timestamp (const ACE_Time_Value &time_value, ACE_TCHAR date_and_time[], size_t date_and_timelen)
Private Attributes
const OPENDDS_STRING	msg_src_
int	dropped_count_
int	delivered_count_
int	sent_count_
ACE_Thread_Mutex	lock_
ACE_Condition_Thread_Mutex	done_condition_
	All messages have been transported condition variable.

---

Detailed Description

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

Definition at line 24 of file MessageTracker.h.

---

Constructor & Destructor Documentation

MessageTracker::MessageTracker

(

const OPENDDS_STRING &

msg_src

)

Definition at line 13 of file MessageTracker.cpp.

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

---

Member Function Documentation

int MessageTracker::dropped_count

(

)

For testing.

Definition at line 143 of file MessageTracker.cpp.

References dropped_count_.

{
  return dropped_count_;
}

void MessageTracker::message_delivered

(

)

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

Definition at line 39 of file MessageTracker.cpp.

References delivered_count_, done_condition_, lock_, and pending_messages().

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

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

  if (!pending_messages())
    done_condition_.broadcast();
}

void MessageTracker::message_dropped

(

)

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

Definition at line 49 of file MessageTracker.cpp.

References done_condition_, dropped_count_, lock_, and pending_messages().

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

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

  if (!pending_messages())
    done_condition_.broadcast();
}

void MessageTracker::message_sent

(

)

Indicate that a message has been to the transport layer.

Definition at line 32 of file MessageTracker.cpp.

References lock_, and sent_count_.

Referenced by OpenDDS::DCPS::DataWriterImpl::association_complete_i(), 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_;
}

bool MessageTracker::pending_messages

(

)

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

Definition at line 23 of file MessageTracker.cpp.

References delivered_count_, dropped_count_, and sent_count_.

Referenced by message_delivered(), message_dropped(), OpenDDS::DCPS::DataWriterImpl::pending_control(), and wait_messages_pending().

{
  if (sent_count_ > delivered_count_ + dropped_count_) {
    return true;
  }
  return false;
}

ACE_TCHAR * MessageTracker::timestamp	(	const ACE_Time_Value &	time_value,
		ACE_TCHAR	date_and_time[],
		size_t	date_and_timelen
	)			[static]

Provide a timestamp for the passed in time. DEPRECATED: remove and replace with ACE::timestamp once TAO 1.6a is no longer supported

Definition at line 107 of file MessageTracker.cpp.

Referenced by wait_messages_pending(), and OpenDDS::DCPS::WriteDataContainer::wait_pending().

{
  //ACE_TRACE ("ACE::timestamp");

  // This magic number is from the formatting statement
  // farther down this routine.
  if (date_and_timelen < 27)
    {
      errno = EINVAL;
      return 0;
    }

  ACE_Time_Value cur_time =
    (time_value == ACE_Time_Value::zero) ?
        ACE_Time_Value (ACE_OS::gettimeofday ()) : time_value;
  time_t secs = cur_time.sec ();
  struct tm tms;
  ACE_OS::localtime_r (&secs, &tms);
  ACE_OS::snprintf (date_and_time,
                    date_and_timelen,
                    ACE_TEXT ("%4.4d-%2.2d-%2.2d %2.2d:%2.2d:%2.2d.%06ld"),
                    tms.tm_year + 1900,
                    tms.tm_mon + 1,
                    tms.tm_mday,
                    tms.tm_hour,
                    tms.tm_min,
                    tms.tm_sec,
                    static_cast<long> (cur_time.usec()));
  date_and_time[date_and_timelen - 1] = '\0';
  return &date_and_time[11];
}

void MessageTracker::wait_messages_pending

(

OPENDDS_STRING &

caller_message

)

Block until all messages have been account for.

Definition at line 59 of file MessageTracker.cpp.

References OpenDDS::DCPS::DCPS_debug_level, done_condition_, msg_src_, pending_messages(), TheServiceParticipant, and timestamp().

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

{
  ACE_Time_Value pending_timeout =
    TheServiceParticipant->pending_timeout();

  ACE_Time_Value* pTimeout = 0;

  if (pending_timeout != ACE_Time_Value::zero) {
    pTimeout = &pending_timeout;
    pending_timeout += ACE_OS::gettimeofday();
  }

  ACE_GUARD(ACE_Thread_Mutex, guard, this->lock_);
  const bool report = DCPS_debug_level > 0 && pending_messages();
  if (report) {
    ACE_TCHAR date_time[50];
    ACE_TCHAR* const time =
      MessageTracker::timestamp(pending_timeout,
                                date_time,
                                50);
    ACE_DEBUG((LM_DEBUG,
               ACE_TEXT("%T (%P|%t) MessageTracker::wait_messages_pending ")
               ACE_TEXT("from source=%C will wait until %s.\n"),
               msg_src_.c_str(),
               (pTimeout == 0 ? ACE_TEXT("(no timeout)") : time)));
  }
  while (true) {
    if (!pending_messages())
      break;

    if (done_condition_.wait(pTimeout) == -1 && pending_messages()) {
      if (DCPS_debug_level) {
        ACE_DEBUG((LM_INFO,
                   ACE_TEXT("(%P|%t) %T MessageTracker::")
                   ACE_TEXT("wait_messages_pending (Redmine Issue# 1446) %p (caller: %s)\n"),
                   ACE_TEXT("Timed out waiting for messages to be transported"),
                   caller_message.c_str()));
      }
      break;
    }
  }
  if (report) {
    ACE_DEBUG((LM_DEBUG,
               "%T (%P|%t) MessageTracker::wait_messages_pending done\n"));
  }
}

---

Member Data Documentation

int OpenDDS::DCPS::MessageTracker::delivered_count_ [private]

Definition at line 72 of file MessageTracker.h.

Referenced by message_delivered(), and pending_messages().

ACE_Condition_Thread_Mutex OpenDDS::DCPS::MessageTracker::done_condition_ [private]

All messages have been transported condition variable.

Definition at line 78 of file MessageTracker.h.

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

int OpenDDS::DCPS::MessageTracker::dropped_count_ [private]

Definition at line 71 of file MessageTracker.h.

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

ACE_Thread_Mutex OpenDDS::DCPS::MessageTracker::lock_ [private]

Definition at line 75 of file MessageTracker.h.

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

const OPENDDS_STRING OpenDDS::DCPS::MessageTracker::msg_src_ [private]

Definition at line 70 of file MessageTracker.h.

Referenced by wait_messages_pending().

int OpenDDS::DCPS::MessageTracker::sent_count_ [private]

Definition at line 73 of file MessageTracker.h.

Referenced by message_sent(), and pending_messages().

---

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

• MessageTracker.h
• MessageTracker.cpp

---