OpenDDS::DCPS::RtpsUdpReceiveStrategy Class Reference

#include <RtpsUdpReceiveStrategy.h>

Inheritance diagram for OpenDDS::DCPS::RtpsUdpReceiveStrategy:

[legend]

Collaboration diagram for OpenDDS::DCPS::RtpsUdpReceiveStrategy:

[legend]

List of all members.

Classes
struct	MessageReceiver
Public Types
typedef std::pair < SequenceNumber, RTPS::FragmentNumberSet >	SeqFragPair
Public Member Functions
	RtpsUdpReceiveStrategy (RtpsUdpDataLink *link, const GuidPrefix_t &local_prefix)
virtual int	handle_input (ACE_HANDLE fd)
bool	remove_frags_from_bitmap (CORBA::Long bitmap[], CORBA::ULong num_bits, const SequenceNumber &base, const RepoId &pub_id)
void	remove_fragments (const SequenceRange &range, const RepoId &pub_id)
typedef	OPENDDS_VECTOR (SeqFragPair) FragmentInfo
bool	has_fragments (const SequenceRange &range, const RepoId &pub_id, FragmentInfo *frag_info=0)
const ReceivedDataSample *	withhold_data_from (const RepoId &sub_id)
void	do_not_withhold_data_from (const RepoId &sub_id)
Private Member Functions
virtual ssize_t	receive_bytes (iovec iov[], int n, ACE_INET_Addr &remote_address, ACE_HANDLE fd)
	Only our subclass knows how to do this.
virtual void	deliver_sample (ReceivedDataSample &sample, const ACE_INET_Addr &remote_address)
	Called when there is a ReceivedDataSample to be delivered.
void	deliver_sample_i (ReceivedDataSample &sample, const RTPS::Submessage &submessage)
virtual int	start_i ()
	Let the subclass start.
virtual void	stop_i ()
	Let the subclass stop.
virtual bool	check_header (const RtpsTransportHeader &header)
	Check the transport header for suitability.
virtual bool	check_header (const RtpsSampleHeader &header)
	Check the data sample header for suitability.
virtual bool	reassemble (ReceivedDataSample &data)
Private Attributes
RtpsUdpDataLink *	link_
SequenceNumber	last_received_
const ReceivedDataSample *	recvd_sample_
RepoIdSet	readers_withheld_
RepoIdSet	readers_selected_
SequenceRange	frags_
TransportReassembly	reassembly_
MessageReceiver	receiver_
ACE_INET_Addr	remote_address_

---

Detailed Description

Definition at line 32 of file RtpsUdpReceiveStrategy.h.

---

Member Typedef Documentation

typedef std::pair<SequenceNumber, RTPS::FragmentNumberSet> OpenDDS::DCPS::RtpsUdpReceiveStrategy::SeqFragPair

Definition at line 53 of file RtpsUdpReceiveStrategy.h.

---

Constructor & Destructor Documentation

OpenDDS::DCPS::RtpsUdpReceiveStrategy::RtpsUdpReceiveStrategy	(	RtpsUdpDataLink *	link,
		const GuidPrefix_t &	local_prefix
	)			[explicit]

Definition at line 24 of file RtpsUdpReceiveStrategy.cpp.

References defined(), if(), and OpenDDS::DCPS::SUBMESSAGE_NONE.

  : link_(link)
  , last_received_()
  , recvd_sample_(0)
  , receiver_(local_prefix)
#if defined(OPENDDS_SECURITY)
  , secure_sample_(0)
#endif
{
#if defined(OPENDDS_SECURITY)
  secure_prefix_.smHeader.submessageId = SUBMESSAGE_NONE;
#endif
}

Here is the call graph for this function:

---

Member Function Documentation

bool OpenDDS::DCPS::RtpsUdpReceiveStrategy::check_header

(

const RtpsSampleHeader &

header

)

[private, virtual]

Check the data sample header for suitability.

Reimplemented from OpenDDS::DCPS::TransportReceiveStrategy< RtpsTransportHeader, RtpsSampleHeader >.

Definition at line 538 of file RtpsUdpReceiveStrategy.cpp.

References OpenDDS::RTPS::DATA_FRAG, OpenDDS::RTPS::Submessage::data_frag_sm, OpenDDS::RTPS::DataFragSubmessage::fragmentsInSubmessage, OpenDDS::RTPS::DataFragSubmessage::fragmentStartingNum, frags_, receiver_, OpenDDS::DCPS::RtpsSampleHeader::submessage_, OpenDDS::DCPS::RtpsUdpReceiveStrategy::MessageReceiver::submsg(), and OpenDDS::DCPS::RtpsSampleHeader::valid().

{

#if defined(OPENDDS_SECURITY)
  if (secure_prefix_.smHeader.submessageId) {
    return header.valid();
  }
#endif

  receiver_.submsg(header.submessage_);

  // save fragmentation details for use in reassemble()
  if (header.valid() && header.submessage_._d() == RTPS::DATA_FRAG) {
    const RTPS::DataFragSubmessage& rtps = header.submessage_.data_frag_sm();
    frags_.first = rtps.fragmentStartingNum.value;
    frags_.second = frags_.first + (rtps.fragmentsInSubmessage - 1);
  }

  return header.valid();
}

Here is the call graph for this function:

bool OpenDDS::DCPS::RtpsUdpReceiveStrategy::check_header

(

const RtpsTransportHeader &

header

)

[private, virtual]

Check the transport header for suitability.

Reimplemented from OpenDDS::DCPS::TransportReceiveStrategy< RtpsTransportHeader, RtpsSampleHeader >.

Definition at line 526 of file RtpsUdpReceiveStrategy.cpp.

References OpenDDS::DCPS::RtpsTransportHeader::header_, receiver_, remote_address_, OpenDDS::DCPS::RtpsUdpReceiveStrategy::MessageReceiver::reset(), OpenDDS::DCPS::SUBMESSAGE_NONE, and OpenDDS::DCPS::RtpsTransportHeader::valid().

{
  receiver_.reset(remote_address_, header.header_);

#if defined(OPENDDS_SECURITY)
  secure_prefix_.smHeader.submessageId = SUBMESSAGE_NONE;
#endif

  return header.valid();
}

Here is the call graph for this function:

void OpenDDS::DCPS::RtpsUdpReceiveStrategy::deliver_sample	(	ReceivedDataSample &	sample,
		const ACE_INET_Addr &	remote_address
	)			[private, virtual]

Called when there is a ReceivedDataSample to be delivered.

Implements OpenDDS::DCPS::TransportReceiveStrategy< RtpsTransportHeader, RtpsSampleHeader >.

Definition at line 73 of file RtpsUdpReceiveStrategy.cpp.

References OpenDDS::RTPS::DATA, deliver_sample_i(), OpenDDS::DCPS::RtpsUdpReceiveStrategy::MessageReceiver::dest_guid_prefix_, link_, OpenDDS::DCPS::RtpsUdpDataLink::local_prefix(), OpenDDS::DCPS::TransportReceiveStrategy< RtpsTransportHeader, RtpsSampleHeader >::received_sample_header(), receiver_, OpenDDS::RTPS::SEC_POSTFIX, OpenDDS::RTPS::SEC_PREFIX, OpenDDS::RTPS::SRTPS_POSTFIX, OpenDDS::RTPS::SRTPS_PREFIX, and OpenDDS::DCPS::RtpsSampleHeader::submessage_.

{
  using namespace RTPS;

  if (std::memcmp(receiver_.dest_guid_prefix_, link_->local_prefix(),
                  sizeof(GuidPrefix_t))) {
    // Not our message, we may be on multicast listening to all the others.
    return;
  }

  const RtpsSampleHeader& rsh = received_sample_header();

#if defined(OPENDDS_SECURITY)
  const SubmessageKind kind = rsh.submessage_._d();

  if ((secure_prefix_.smHeader.submessageId == SRTPS_PREFIX
       && kind != SRTPS_POSTFIX) ||
      (secure_prefix_.smHeader.submessageId == SEC_PREFIX
       && kind != SEC_POSTFIX)) {
    // secure envelope in progress, defer processing
    secure_submessages_.push_back(rsh.submessage_);
    if (kind == DATA) {
      // SRTPS: once full-message protection is supported, this technique will
      // need to be extended to support > 1 data payload (auth. only)
      secure_sample_ = sample;
    }
    return;
  }
#endif

  deliver_sample_i(sample, rsh.submessage_);
}

Here is the call graph for this function:

void OpenDDS::DCPS::RtpsUdpReceiveStrategy::deliver_sample_i	(	ReceivedDataSample &	sample,
		const RTPS::Submessage &	submessage
	)			[private]

Definition at line 108 of file RtpsUdpReceiveStrategy.cpp.

References OpenDDS::RTPS::ACKNACK, OpenDDS::RTPS::Submessage::acknack_sm, OpenDDS::RTPS::DATA, OpenDDS::DCPS::DataLink::data_received(), OpenDDS::DCPS::DataLink::data_received_include(), OpenDDS::RTPS::Submessage::data_sm, OpenDDS::DCPS::DATAWRITER_LIVELINESS, OpenDDS::DCPS::GUID_t::entityId, OpenDDS::DCPS::ENTITYID_UNKNOWN, OpenDDS::DCPS::RtpsUdpReceiveStrategy::MessageReceiver::fill_header(), OpenDDS::RTPS::FLAG_L, OpenDDS::RTPS::GAP, OpenDDS::RTPS::Submessage::gap_sm, OpenDDS::DCPS::SequenceNumber::getValue(), OpenDDS::DCPS::GUID_t::guidPrefix, OpenDDS::RTPS::Submessage::hb_frag_sm, OpenDDS::DCPS::ReceivedDataSample::header_, OpenDDS::RTPS::HEARTBEAT, OpenDDS::RTPS::HEARTBEAT_FRAG, OpenDDS::RTPS::Submessage::heartbeat_sm, OpenDDS::RTPS::INFO_DST, OpenDDS::RTPS::INFO_REPLY, OpenDDS::RTPS::INFO_REPLY_IP4, OpenDDS::RTPS::INFO_SRC, OpenDDS::RTPS::INFO_TS, link_, LM_DEBUG, LM_ERROR, OpenDDS::DCPS::RtpsUdpDataLink::local_prefix(), OpenDDS::DCPS::DataSampleHeader::message_id_, OpenDDS::RTPS::NACK_FRAG, OpenDDS::RTPS::Submessage::nack_frag_sm, OPENDDS_STRING, OpenDDS::DCPS::DataSampleHeader::publication_id_, readers_selected_, readers_withheld_, OpenDDS::DCPS::RtpsUdpDataLink::received(), receiver_, recvd_sample_, OpenDDS::RTPS::SEC_POSTFIX, OpenDDS::RTPS::SEC_PREFIX, OpenDDS::RTPS::Submessage::security_sm, OpenDDS::DCPS::DataSampleHeader::sequence_, OpenDDS::DCPS::RtpsUdpReceiveStrategy::MessageReceiver::source_guid_prefix_, OpenDDS::RTPS::SRTPS_POSTFIX, OpenDDS::RTPS::SRTPS_PREFIX, OpenDDS::DCPS::SUBMESSAGE_NONE, and OpenDDS::DCPS::Transport_debug_level.

Referenced by deliver_sample().

{
  using namespace RTPS;
  const SubmessageKind kind = submessage._d();

  switch (kind) {
  case INFO_SRC:
  case INFO_REPLY_IP4:
  case INFO_DST:
  case INFO_REPLY:
  case INFO_TS:
    // No-op: the INFO_* submessages only modify the state of the
    // MessageReceiver (see check_header()), they are not passed up to DCPS.
    break;

  case DATA: {
    receiver_.fill_header(sample.header_);
    const DataSubmessage& data = submessage.data_sm();
    recvd_sample_ = &sample;
    readers_selected_.clear();
    readers_withheld_.clear();
    // If this sample should be withheld from some readers in order to maintain
    // in-order delivery, link_->received() will add it to readers_withheld_ otherwise
    // it will be added to readers_selected_
    link_->received(data, receiver_.source_guid_prefix_);
    recvd_sample_ = 0;

    if (data.readerId != ENTITYID_UNKNOWN) {
      RepoId reader;
      std::memcpy(reader.guidPrefix, link_->local_prefix(),
                  sizeof(GuidPrefix_t));
      reader.entityId = data.readerId;
      if (!readers_withheld_.count(reader)) {
        if (Transport_debug_level > 5) {
          GuidConverter reader_conv(reader);
          ACE_DEBUG((LM_DEBUG, "(%P|%t) RtpsUdpReceiveStrategy[%@]::deliver_sample - calling DataLink::data_received for seq: %q to reader %C\n", this,
                               sample.header_.sequence_.getValue(),
                               OPENDDS_STRING(reader_conv).c_str()));
        }
#if defined(OPENDDS_SECURITY)
        if (decode_payload(sample, data)) {
          link_->data_received(sample, reader);
        }
#else
        link_->data_received(sample, reader);
#endif

      }

    } else {
      if (Transport_debug_level > 5) {
        OPENDDS_STRING included_ids;
        bool first = true;
        RepoIdSet::iterator iter = readers_selected_.begin();
        while(iter != readers_selected_.end()) {
          included_ids += (first ? "" : "\n") + OPENDDS_STRING(GuidConverter(*iter));
          first = false;
          ++iter;
        }
        OPENDDS_STRING excluded_ids;
        first = true;
        RepoIdSet::iterator iter2 = this->readers_withheld_.begin();
        while(iter2 != readers_withheld_.end()) {
            excluded_ids += (first ? "" : "\n") + OPENDDS_STRING(GuidConverter(*iter2));
          first = false;
          ++iter2;
        }
        ACE_DEBUG((LM_DEBUG, "(%P|%t)  - RtpsUdpReceiveStrategy[%@]::deliver_sample \nreaders_selected ids:\n%C\n", this, included_ids.c_str()));
        ACE_DEBUG((LM_DEBUG, "(%P|%t)  - RtpsUdpReceiveStrategy[%@]::deliver_sample \nreaders_withheld ids:\n%C\n", this, excluded_ids.c_str()));
      }

      if (readers_withheld_.empty() && readers_selected_.empty()) {
        if (Transport_debug_level > 5) {
          ACE_DEBUG((LM_DEBUG, "(%P|%t) RtpsUdpReceiveStrategy[%@]::deliver_sample - calling DataLink::data_received for seq: %q TO ALL, no exclusion or inclusion\n", this,
                               sample.header_.sequence_.getValue()));
        }

#if defined(OPENDDS_SECURITY)
        if (decode_payload(sample, data)) {
          link_->data_received(sample);
        }
#else
        link_->data_received(sample);
#endif

      } else {
        if (Transport_debug_level > 5) {
          ACE_DEBUG((LM_DEBUG, "(%P|%t) RtpsUdpReceiveStrategy[%@]::deliver_sample - calling DataLink::data_received_include for seq: %q to readers_selected_\n", this,
                               sample.header_.sequence_.getValue()));
        }

#if defined(OPENDDS_SECURITY)
        if (decode_payload(sample, data)) {
          link_->data_received_include(sample, readers_selected_);
        }
#else
        link_->data_received_include(sample, readers_selected_);
#endif

      }
    }
    break;
  }
  case GAP:
    link_->received(submessage.gap_sm(), receiver_.source_guid_prefix_);
    break;

  case HEARTBEAT:
    link_->received(submessage.heartbeat_sm(),
                    receiver_.source_guid_prefix_);
    if (submessage.heartbeat_sm().smHeader.flags & FLAG_L) {
      // Liveliness has been asserted.  Create a DATAWRITER_LIVELINESS message.
      sample.header_.message_id_ = DATAWRITER_LIVELINESS;
      receiver_.fill_header(sample.header_);
      sample.header_.publication_id_.entityId = submessage.heartbeat_sm().writerId;
      link_->data_received(sample);
    }
    break;

  case ACKNACK:
    link_->received(submessage.acknack_sm(),
                    receiver_.source_guid_prefix_);
    break;

  case HEARTBEAT_FRAG:
    link_->received(submessage.hb_frag_sm(),
                    receiver_.source_guid_prefix_);
    break;

  case NACK_FRAG:
    link_->received(submessage.nack_frag_sm(),
                    receiver_.source_guid_prefix_);
    break;

  /* no case DATA_FRAG: by the time deliver_sample() is called, reassemble()
     has successfully reassembled the fragments and we now have a DATA submsg
   */

#if defined(OPENDDS_SECURITY)
  case SRTPS_PREFIX:
  case SEC_PREFIX:
    secure_prefix_ = submessage.security_sm();
    break;

  case SRTPS_POSTFIX:
    secure_prefix_.smHeader.submessageId = SUBMESSAGE_NONE;
    secure_sample_ = ReceivedDataSample(0);
    ACE_ERROR((LM_ERROR, "ERROR: RtpsUdpReceiveStrategy SRTPS unsupported.\n"));
    break;

  case SEC_POSTFIX:
    deliver_from_secure(submessage);
    break;
#endif

  default:
    break;
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

void OpenDDS::DCPS::RtpsUdpReceiveStrategy::do_not_withhold_data_from

(

const RepoId &

sub_id

)

Definition at line 567 of file RtpsUdpReceiveStrategy.cpp.

References readers_selected_.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::process_data_i().

{
  readers_selected_.insert(sub_id);
}

Here is the caller graph for this function:

int OpenDDS::DCPS::RtpsUdpReceiveStrategy::handle_input

(

ACE_HANDLE

fd

)

[virtual]

Reimplemented from ACE_Event_Handler.

Definition at line 39 of file RtpsUdpReceiveStrategy.cpp.

References OpenDDS::DCPS::TransportReceiveStrategy< RtpsTransportHeader, RtpsSampleHeader >::handle_dds_input().

{
  return handle_dds_input(fd);
}

Here is the call graph for this function:

bool OpenDDS::DCPS::RtpsUdpReceiveStrategy::has_fragments	(	const SequenceRange &	range,
		const RepoId &	pub_id,
		FragmentInfo *	frag_info = 0
	)

Definition at line 644 of file RtpsUdpReceiveStrategy.cpp.

References OpenDDS::RTPS::FragmentNumberSet::bitmap, OpenDDS::RTPS::FragmentNumberSet::bitmapBase, OpenDDS::DCPS::TransportReassembly::get_gaps(), OpenDDS::DCPS::TransportReassembly::has_frags(), OpenDDS::RTPS::FragmentNumberSet::numBits, and reassembly_.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::generate_nack_frags(), and OpenDDS::DCPS::RtpsUdpDataLink::process_heartbeat_i().

{
  for (SequenceNumber sn = range.first; sn <= range.second; ++sn) {
    if (reassembly_.has_frags(sn, pub_id)) {
      if (frag_info) {
        std::pair<SequenceNumber, RTPS::FragmentNumberSet> p;
        p.first = sn;
        frag_info->push_back(p);
        RTPS::FragmentNumberSet& missing_frags = frag_info->back().second;
        missing_frags.bitmap.length(8); // start at max length
        missing_frags.bitmapBase.value =
          reassembly_.get_gaps(sn, pub_id, missing_frags.bitmap.get_buffer(),
                               8, missing_frags.numBits);
        // reduce length in case get_gaps() didn't need all that room
        missing_frags.bitmap.length((missing_frags.numBits + 31) / 32);
      } else {
        return true;
      }
    }
  }
  return frag_info ? !frag_info->empty() : false;
}

Here is the call graph for this function:

Here is the caller graph for this function:

typedef OpenDDS::DCPS::RtpsUdpReceiveStrategy::OPENDDS_VECTOR

(

SeqFragPair

)

bool OpenDDS::DCPS::RtpsUdpReceiveStrategy::reassemble

(

ReceivedDataSample &

data

)

[private, virtual]

Reimplemented from OpenDDS::DCPS::TransportReceiveStrategy< RtpsTransportHeader, RtpsSampleHeader >.

Definition at line 573 of file RtpsUdpReceiveStrategy.cpp.

References OpenDDS::DCPS::DataSampleHeader::byte_order_, OpenDDS::RTPS::DATA, OpenDDS::RTPS::Submessage::data_frag_sm, OpenDDS::RTPS::DATA_OCTETS_TO_IQOS, OpenDDS::RTPS::Submessage::data_sm, OpenDDS::DCPS::RtpsUdpReceiveStrategy::MessageReceiver::fill_header(), OpenDDS::RTPS::FLAG_D, OpenDDS::RTPS::FLAG_E, OpenDDS::RTPS::FLAG_K_IN_DATA, frags_, OpenDDS::DCPS::ReceivedDataSample::header_, OpenDDS::DCPS::DataSampleHeader::key_fields_only_, OpenDDS::DCPS::TransportReassembly::reassemble(), reassembly_, OpenDDS::DCPS::TransportReceiveStrategy< RtpsTransportHeader, RtpsSampleHeader >::received_sample_header(), receiver_, OpenDDS::DCPS::ReceivedDataSample::sample_, and OpenDDS::DCPS::RtpsSampleHeader::submessage_.

{
  using namespace RTPS;
  receiver_.fill_header(data.header_); // set publication_id_.guidPrefix
  if (reassembly_.reassemble(frags_, data)) {

    // Reassembly was successful, replace DataFrag with Data.  This doesn't have
    // to be a fully-formed DataSubmessage, just enough for this class to use
    // in deliver_sample() which ends up calling RtpsUdpDataLink::received().
    // In particular we will need the SequenceNumber, but ignore the iQoS.

    // Peek at the byte order from the encapsulation containing the payload.
    data.header_.byte_order_ = data.sample_->rd_ptr()[1] & FLAG_E;

    RtpsSampleHeader& rsh = received_sample_header();
    const DataFragSubmessage& dfsm = rsh.submessage_.data_frag_sm();

    const CORBA::Octet data_flags = (data.header_.byte_order_ ? FLAG_E : 0)
      | (data.header_.key_fields_only_ ? FLAG_K_IN_DATA : FLAG_D);
    const DataSubmessage dsm = {
      {DATA, data_flags, 0}, 0, DATA_OCTETS_TO_IQOS,
      dfsm.readerId, dfsm.writerId, dfsm.writerSN, ParameterList()};
    rsh.submessage_.data_sm(dsm);
    return true;
  }
  return false;
}

Here is the call graph for this function:

ssize_t OpenDDS::DCPS::RtpsUdpReceiveStrategy::receive_bytes	(	iovec	iov[],
		int	n,
		ACE_INET_Addr &	remote_address,
		ACE_HANDLE	fd
	)			[private, virtual]

Only our subclass knows how to do this.

Implements OpenDDS::DCPS::TransportReceiveStrategy< RtpsTransportHeader, RtpsSampleHeader >.

Definition at line 45 of file RtpsUdpReceiveStrategy.cpp.

References ACE_IPC_SAP::get_handle(), link_, OpenDDS::DCPS::RtpsUdpDataLink::multicast_socket(), ACE_SOCK_Dgram::recv(), remote_address_, socket(), and OpenDDS::DCPS::RtpsUdpDataLink::unicast_socket().

{
  const ACE_SOCK_Dgram& socket =
    (fd == link_->unicast_socket().get_handle())
    ? link_->unicast_socket() : link_->multicast_socket();
#ifdef ACE_LACKS_SENDMSG
  char buffer[0x10000];
  ssize_t scatter = socket.recv(buffer, sizeof buffer, remote_address);
  char* iter = buffer;
  for (int i = 0; scatter > 0 && i < n; ++i) {
    const size_t chunk = std::min(static_cast<size_t>(iov[i].iov_len), // int on LynxOS
                                  static_cast<size_t>(scatter));
    std::memcpy(iov[i].iov_base, iter, chunk);
    scatter -= chunk;
    iter += chunk;
  }
  const ssize_t ret = (scatter < 0) ? scatter : (iter - buffer);
#else
  const ssize_t ret = socket.recv(iov, n, remote_address);
#endif
  remote_address_ = remote_address;
  return ret;
}

Here is the call graph for this function:

void OpenDDS::DCPS::RtpsUdpReceiveStrategy::remove_fragments	(	const SequenceRange &	range,
		const RepoId &	pub_id
	)

Remove any saved fragments. We do not expect to receive any more fragments with sequence numbers in "range" from publication "pub_id".

Definition at line 635 of file RtpsUdpReceiveStrategy.cpp.

References OpenDDS::DCPS::TransportReassembly::data_unavailable(), and reassembly_.

{
  for (SequenceNumber sn = range.first; sn <= range.second; ++sn) {
    reassembly_.data_unavailable(sn, pub_id);
  }
}

Here is the call graph for this function:

bool OpenDDS::DCPS::RtpsUdpReceiveStrategy::remove_frags_from_bitmap	(	CORBA::Long	bitmap[],
		CORBA::ULong	num_bits,
		const SequenceNumber &	base,
		const RepoId &	pub_id
	)

For each "1" bit in the bitmap, change it to a "0" if there are fragments from publication "pub_id" for the sequence number represented by that position in the bitmap. Returns true if the bitmap was changed.

Definition at line 602 of file RtpsUdpReceiveStrategy.cpp.

References OpenDDS::DCPS::TransportReassembly::has_frags(), and reassembly_.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::send_ack_nacks().

{
  bool modified = false;
  for (CORBA::ULong i = 0, x = 0, bit = 0; i < num_bits; ++i, ++bit) {
    if (bit == 32) bit = 0;

    if (bit == 0) {
      x = static_cast<CORBA::ULong>(bitmap[i / 32]);
      if (x == 0) {
        // skip an entire Long if it's all 0's (adds 32 due to ++i)
        i += 31;
        bit = 31;
        //FUTURE: this could be generalized with something like the x86 "bsr"
        //        instruction using compiler intrinsics, VC++ _BitScanReverse()
        //        and GCC __builtin_clz()
        continue;
      }
    }

    const CORBA::ULong mask = 1 << (31 - bit);
    if ((x & mask) && reassembly_.has_frags(base + i, pub_id)) {
      x &= ~mask;
      bitmap[i / 32] = x;
      modified = true;
    }
  }
  return modified;
}

Here is the call graph for this function:

Here is the caller graph for this function:

int OpenDDS::DCPS::RtpsUdpReceiveStrategy::start_i

(

)

[private, virtual]

Let the subclass start.

Implements OpenDDS::DCPS::TransportReceiveStrategy< RtpsTransportHeader, RtpsSampleHeader >.

Definition at line 458 of file RtpsUdpReceiveStrategy.cpp.

References ACE_TEXT(), OpenDDS::DCPS::RtpsUdpDataLink::config(), ACE_IPC_SAP::control(), ACE_IPC_SAP::get_handle(), OpenDDS::DCPS::RtpsUdpDataLink::get_reactor(), link_, LM_ERROR, OpenDDS::DCPS::RtpsUdpDataLink::multicast_socket(), ACE_Event_Handler::reactor(), ACE_Event_Handler::READ_MASK, ACE_Reactor::register_handler(), OpenDDS::DCPS::RtpsUdpDataLink::unicast_socket(), and OpenDDS::DCPS::RtpsUdpInst::use_multicast_.

{
  ACE_Reactor* reactor = link_->get_reactor();
  if (reactor == 0) {
    ACE_ERROR_RETURN((LM_ERROR,
                      ACE_TEXT("(%P|%t) ERROR: ")
                      ACE_TEXT("RtpsUdpReceiveStrategy::start_i: ")
                      ACE_TEXT("NULL reactor reference!\n")),
                     -1);
  }

#ifdef ACE_WIN32
  // By default Winsock will cause reads to fail with "connection reset"
  // when UDP sends result in ICMP "port unreachable" messages.
  // The transport framework is not set up for this since returning <= 0
  // from our receive_bytes causes the framework to close down the datalink
  // which in this case is used to receive from multiple peers.
  BOOL recv_udp_connreset = FALSE;
  link_->unicast_socket().control(SIO_UDP_CONNRESET, &recv_udp_connreset);
#endif

  if (reactor->register_handler(link_->unicast_socket().get_handle(), this,
                                ACE_Event_Handler::READ_MASK) != 0) {
    ACE_ERROR_RETURN((LM_ERROR,
                      ACE_TEXT("(%P|%t) ERROR: ")
                      ACE_TEXT("RtpsUdpReceiveStrategy::start_i: ")
                      ACE_TEXT("failed to register handler for unicast ")
                      ACE_TEXT("socket %d\n"),
                      link_->unicast_socket().get_handle()),
                     -1);
  }

  if (link_->config().use_multicast_) {
    if (reactor->register_handler(link_->multicast_socket().get_handle(), this,
                                  ACE_Event_Handler::READ_MASK) != 0) {
      ACE_ERROR_RETURN((LM_ERROR,
                        ACE_TEXT("(%P|%t) ERROR: ")
                        ACE_TEXT("RtpsUdpReceiveStrategy::start_i: ")
                        ACE_TEXT("failed to register handler for multicast\n")),
                       -1);
    }
  }

  return 0;
}

Here is the call graph for this function:

void OpenDDS::DCPS::RtpsUdpReceiveStrategy::stop_i

(

)

[private, virtual]

Let the subclass stop.

Implements OpenDDS::DCPS::TransportReceiveStrategy< RtpsTransportHeader, RtpsSampleHeader >.

Definition at line 505 of file RtpsUdpReceiveStrategy.cpp.

References ACE_TEXT(), OpenDDS::DCPS::RtpsUdpDataLink::config(), ACE_IPC_SAP::get_handle(), OpenDDS::DCPS::RtpsUdpDataLink::get_reactor(), link_, LM_ERROR, OpenDDS::DCPS::RtpsUdpDataLink::multicast_socket(), ACE_Event_Handler::reactor(), ACE_Event_Handler::READ_MASK, ACE_Reactor::remove_handler(), OpenDDS::DCPS::RtpsUdpDataLink::unicast_socket(), and OpenDDS::DCPS::RtpsUdpInst::use_multicast_.

{
  ACE_Reactor* reactor = link_->get_reactor();
  if (reactor == 0) {
    ACE_ERROR((LM_ERROR,
               ACE_TEXT("(%P|%t) ERROR: ")
               ACE_TEXT("RtpsUdpReceiveStrategy::stop_i: ")
               ACE_TEXT("NULL reactor reference!\n")));
    return;
  }

  reactor->remove_handler(link_->unicast_socket().get_handle(),
                          ACE_Event_Handler::READ_MASK);

  if (link_->config().use_multicast_) {
    reactor->remove_handler(link_->multicast_socket().get_handle(),
                            ACE_Event_Handler::READ_MASK);
  }
}

Here is the call graph for this function:

const ReceivedDataSample * OpenDDS::DCPS::RtpsUdpReceiveStrategy::withhold_data_from

(

const RepoId &

sub_id

)

Prevent delivery of the currently in-progress data sample to the subscription sub_id. Returns pointer to the in-progress data so it can be stored for later delivery.

Definition at line 560 of file RtpsUdpReceiveStrategy.cpp.

References readers_withheld_, and recvd_sample_.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::process_data_i().

{
  readers_withheld_.insert(sub_id);
  return recvd_sample_;
}

Here is the caller graph for this function:

---

Member Data Documentation

SequenceRange OpenDDS::DCPS::RtpsUdpReceiveStrategy::frags_ [private]

Definition at line 102 of file RtpsUdpReceiveStrategy.h.

Referenced by check_header(), and reassemble().

SequenceNumber OpenDDS::DCPS::RtpsUdpReceiveStrategy::last_received_ [private]

Definition at line 97 of file RtpsUdpReceiveStrategy.h.

RtpsUdpDataLink* OpenDDS::DCPS::RtpsUdpReceiveStrategy::link_ [private]

Definition at line 96 of file RtpsUdpReceiveStrategy.h.

Referenced by deliver_sample(), deliver_sample_i(), receive_bytes(), start_i(), and stop_i().

RepoIdSet OpenDDS::DCPS::RtpsUdpReceiveStrategy::readers_selected_ [private]

Definition at line 100 of file RtpsUdpReceiveStrategy.h.

Referenced by deliver_sample_i(), and do_not_withhold_data_from().

RepoIdSet OpenDDS::DCPS::RtpsUdpReceiveStrategy::readers_withheld_ [private]

Definition at line 100 of file RtpsUdpReceiveStrategy.h.

Referenced by deliver_sample_i(), and withhold_data_from().

TransportReassembly OpenDDS::DCPS::RtpsUdpReceiveStrategy::reassembly_ [private]

Definition at line 103 of file RtpsUdpReceiveStrategy.h.

Referenced by has_fragments(), reassemble(), remove_fragments(), and remove_frags_from_bitmap().

MessageReceiver OpenDDS::DCPS::RtpsUdpReceiveStrategy::receiver_ [private]

Definition at line 131 of file RtpsUdpReceiveStrategy.h.

Referenced by check_header(), deliver_sample(), deliver_sample_i(), and reassemble().

const ReceivedDataSample* OpenDDS::DCPS::RtpsUdpReceiveStrategy::recvd_sample_ [private]

Definition at line 99 of file RtpsUdpReceiveStrategy.h.

Referenced by deliver_sample_i(), and withhold_data_from().

ACE_INET_Addr OpenDDS::DCPS::RtpsUdpReceiveStrategy::remote_address_ [private]

Definition at line 132 of file RtpsUdpReceiveStrategy.h.

Referenced by check_header(), and receive_bytes().

---

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

• RtpsUdpReceiveStrategy.h
• RtpsUdpReceiveStrategy.cpp