OpenDDS::DCPS::UdpReceiveStrategy Class Reference

#include <UdpReceiveStrategy.h>

Inheritance diagram for OpenDDS::DCPS::UdpReceiveStrategy:

[legend]Collaboration diagram for OpenDDS::DCPS::UdpReceiveStrategy:


Public Member Functions
	UdpReceiveStrategy (UdpDataLink *link)
virtual ACE_HANDLE	get_handle () const
virtual int	handle_input (ACE_HANDLE fd)
Protected 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.
virtual int	start_i ()
	Let the subclass start.
virtual void	stop_i ()
	Let the subclass stop.
virtual bool	check_header (const TransportHeader &header)
virtual bool	check_header (const DataSampleHeader &header)
Private Types
typedef std::pair< TransportReassembly, SequenceNumber >	ReassemblyInfo
Private Member Functions
virtual bool	reassemble (ReceivedDataSample &data)
typedef	OPENDDS_MAP (ACE_INET_Addr, ReassemblyInfo) ReassemblyMap
Private Attributes
UdpDataLink *	link_
SequenceNumber	expected_
ACE_INET_Addr	remote_address_
ReassemblyMap	reassembly_

Detailed Description

Definition at line 25 of file UdpReceiveStrategy.h.

Member Typedef Documentation

typedef std::pair<TransportReassembly, SequenceNumber> OpenDDS::DCPS::UdpReceiveStrategy::ReassemblyInfo [private]

Definition at line 61 of file UdpReceiveStrategy.h.

Constructor & Destructor Documentation

OpenDDS::DCPS::UdpReceiveStrategy::UdpReceiveStrategy ( UdpDataLink * link ) [explicit]

Definition at line 16 of file UdpReceiveStrategy.cpp.

00017   : link_(link)
00018   , expected_(SequenceNumber::SEQUENCENUMBER_UNKNOWN())
00019 {
00020 }

Member Function Documentation

virtual bool OpenDDS::DCPS::UdpReceiveStrategy::check_header ( const DataSampleHeader & header ) [inline, protected, virtual]

Definition at line 48 of file UdpReceiveStrategy.h.

References header.

00049   {
00050     return TransportReceiveStrategy<>::check_header(header);
00051   }

bool OpenDDS::DCPS::UdpReceiveStrategy::check_header ( const TransportHeader & header ) [protected, virtual]

Definition at line 108 of file UdpReceiveStrategy.cpp.

References expected_, header, reassembly_, remote_address_, OpenDDS::DCPS::SequenceNumber::SEQUENCENUMBER_UNKNOWN(), and VDBG_LVL.

00109 {
00110   ReassemblyInfo& info = reassembly_[remote_address_];
00111 
00112   if (header.sequence_ != info.second &&
00113       expected_ != SequenceNumber::SEQUENCENUMBER_UNKNOWN()) {
00114     VDBG_LVL((LM_WARNING,
00115                ACE_TEXT("(%P|%t) WARNING: UdpReceiveStrategy::check_header ")
00116                ACE_TEXT("expected %q received %q\n"),
00117                info.second.getValue(), header.sequence_.getValue()), 2);
00118     SequenceRange range(info.second, header.sequence_.previous());
00119     info.first.data_unavailable(range);
00120   }
00121 
00122   info.second = header.sequence_;
00123   ++info.second;
00124   return true;
00125 }

void OpenDDS::DCPS::UdpReceiveStrategy::deliver_sample	(	ReceivedDataSample &	sample,
		const ACE_INET_Addr &	remote_address
	)			`[protected, virtual]`

Called when there is a ReceivedDataSample to be delivered.

Implements OpenDDS::DCPS::TransportReceiveStrategy< TH, DSH >.

Definition at line 47 of file UdpReceiveStrategy.cpp.

References OpenDDS::DCPS::UdpDataLink::control_received(), OpenDDS::DCPS::DataLink::data_received(), OpenDDS::DCPS::ReceivedDataSample::header_, link_, OpenDDS::DCPS::DataSampleHeader::message_id_, and OpenDDS::DCPS::TRANSPORT_CONTROL.

00049 {
00050   switch (sample.header_.message_id_) {
00051 
00052   case TRANSPORT_CONTROL:
00053     this->link_->control_received(sample, remote_address);
00054     break;
00055 
00056   default:
00057     this->link_->data_received(sample);
00058     break;
00059   }
00060 }

ACE_HANDLE OpenDDS::DCPS::UdpReceiveStrategy::get_handle ( ) const [virtual]

Definition at line 23 of file UdpReceiveStrategy.cpp.

References link_, and OpenDDS::DCPS::UdpDataLink::socket().

00024 {
00025   ACE_SOCK_Dgram& socket = this->link_->socket();
00026   return socket.get_handle();
00027 }

int OpenDDS::DCPS::UdpReceiveStrategy::handle_input ( ACE_HANDLE fd ) [virtual]

Definition at line 30 of file UdpReceiveStrategy.cpp.

References OpenDDS::DCPS::TransportReceiveStrategy< TH, DSH >::handle_dds_input().

00031 {
00032   return this->handle_dds_input(fd);
00033 }

typedef OpenDDS::DCPS::UdpReceiveStrategy::OPENDDS_MAP	(	ACE_INET_Addr	,
		ReassemblyInfo
	)			`[private]`

bool OpenDDS::DCPS::UdpReceiveStrategy::reassemble ( ReceivedDataSample & data ) [private, virtual]

Reimplemented from OpenDDS::DCPS::TransportReceiveStrategy< TH, DSH >.

Definition at line 128 of file UdpReceiveStrategy.cpp.

References header, reassembly_, OpenDDS::DCPS::TransportReceiveStrategy< TH, DSH >::received_header(), and remote_address_.

00129 {
00130   ReassemblyInfo& info = reassembly_[remote_address_];
00131   const TransportHeader& header = received_header();
00132   return info.first.reassemble(header.sequence_, header.first_fragment_, data);
00133 }

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

Only our subclass knows how to do this.

Implements OpenDDS::DCPS::TransportReceiveStrategy< TH, DSH >.

Definition at line 36 of file UdpReceiveStrategy.cpp.

References link_, remote_address_, and OpenDDS::DCPS::UdpDataLink::socket().

00040 {
00041   const ssize_t ret = this->link_->socket().recv(iov, n, remote_address);
00042   remote_address_ = remote_address;
00043   return ret;
00044 }

int OpenDDS::DCPS::UdpReceiveStrategy::start_i ( ) [protected, virtual]

Let the subclass start.

Implements OpenDDS::DCPS::TransportReceiveStrategy< TH, DSH >.

Definition at line 63 of file UdpReceiveStrategy.cpp.

References OpenDDS::DCPS::UdpDataLink::get_reactor(), link_, OpenDDS::DCPS::UdpDataLink::socket(), and OpenDDS::DCPS::Transport_debug_level.

00064 {
00065   ACE_Reactor* reactor = this->link_->get_reactor();
00066   if (reactor == 0) {
00067     ACE_ERROR_RETURN((LM_ERROR,
00068                       ACE_TEXT("(%P|%t) ERROR: ")
00069                       ACE_TEXT("UdpReceiveStrategy::start_i: ")
00070                       ACE_TEXT("NULL reactor reference!\n")),
00071                      -1);
00072   }
00073 
00074   if (reactor->register_handler(this, ACE_Event_Handler::READ_MASK) != 0) {
00075     ACE_ERROR_RETURN((LM_ERROR,
00076                       ACE_TEXT("(%P|%t) ERROR: ")
00077                       ACE_TEXT("UdpReceiveStrategy::start_i: ")
00078                       ACE_TEXT("failed to register handler for DataLink!\n")),
00079                      -1);
00080   }
00081 
00082   if (Transport_debug_level > 5) {
00083     ACE_INET_Addr addr;
00084     link_->socket().get_local_addr(addr);
00085     ACE_DEBUG((LM_DEBUG, "(%P|%t) UdpReceiveStrategy::start_i: "
00086                "listening on %C:%hu\n",
00087                addr.get_host_addr(), addr.get_port_number()));
00088   }
00089   return 0;
00090 }

void OpenDDS::DCPS::UdpReceiveStrategy::stop_i ( ) [protected, virtual]

Let the subclass stop.

Implements OpenDDS::DCPS::TransportReceiveStrategy< TH, DSH >.

Definition at line 93 of file UdpReceiveStrategy.cpp.

References OpenDDS::DCPS::UdpDataLink::get_reactor(), and link_.

00094 {
00095   ACE_Reactor* reactor = this->link_->get_reactor();
00096   if (reactor == 0) {
00097     ACE_ERROR((LM_ERROR,
00098                ACE_TEXT("(%P|%t) ERROR: ")
00099                ACE_TEXT("UdpReceiveStrategy::stop_i: ")
00100                ACE_TEXT("NULL reactor reference!\n")));
00101     return;
00102   }
00103 
00104   reactor->remove_handler(this, ACE_Event_Handler::READ_MASK);
00105 }