diff options
Diffstat (limited to 'ace/RMCast')
33 files changed, 792 insertions, 84 deletions
diff --git a/ace/RMCast/Makefile b/ace/RMCast/Makefile index 868795fa82c..fb270696c5d 100644 --- a/ace/RMCast/Makefile +++ b/ace/RMCast/Makefile @@ -19,10 +19,16 @@ FILES= \ RMCast_Membership \ RMCast_Retransmission \ RMCast_Reordering \ + RMCast_Reliable_Factory \ + RMcast_Singleton_Factory \ + RMCast_Control_Splitter \ + RMcast_Resend_Handler \ \ RMCast_IO_UDP \ RMCast_UDP_Event_Handler \ - RMCast_UDP_Proxy + RMCast_UDP_Proxy \ + RMCast_UDP_Reliable_Receiver \ + RMCast_UDP_Reliable_Sender #---------------------------------------------------------------------------- # Include macros and targets diff --git a/ace/RMCast/RMCast.dsp b/ace/RMCast/RMCast.dsp index d160812fee6..e6cbc7af35a 100644 --- a/ace/RMCast/RMCast.dsp +++ b/ace/RMCast/RMCast.dsp @@ -98,6 +98,10 @@ SOURCE=.\RMCast.cpp # End Source File
# Begin Source File
+SOURCE=.\RMCast_Control_Splitter.cpp
+# End Source File
+# Begin Source File
+
SOURCE=.\RMCast_Fragment.cpp
# End Source File
# Begin Source File
@@ -130,20 +134,40 @@ SOURCE=.\RMCast_Reassembly.cpp # End Source File
# Begin Source File
+SOURCE=.\RMCast_Reliable_Factory.cpp
+# End Source File
+# Begin Source File
+
SOURCE=.\RMCast_Reordering.cpp
# End Source File
# Begin Source File
+SOURCE=.\RMCast_Resend_Handler.cpp
+# End Source File
+# Begin Source File
+
SOURCE=.\RMCast_Retransmission.cpp
# End Source File
# Begin Source File
+SOURCE=.\RMCast_Singleton_Factory.cpp
+# End Source File
+# Begin Source File
+
SOURCE=.\RMCast_UDP_Event_Handler.cpp
# End Source File
# Begin Source File
SOURCE=.\RMCast_UDP_Proxy.cpp
# End Source File
+# Begin Source File
+
+SOURCE=.\RMCast_UDP_Reliable_Receiver.cpp
+# End Source File
+# Begin Source File
+
+SOURCE=.\RMCast_UDP_Reliable_Sender.cpp
+# End Source File
# End Group
# Begin Group "Header Files"
@@ -154,6 +178,10 @@ SOURCE=.\RMCast.h # End Source File
# Begin Source File
+SOURCE=.\RMCast_Control_Splitter.h
+# End Source File
+# Begin Source File
+
SOURCE=.\RMCast_Export.h
# End Source File
# Begin Source File
@@ -190,20 +218,40 @@ SOURCE=.\RMCast_Reassembly.h # End Source File
# Begin Source File
+SOURCE=.\RMCast_Reliable_Factory.h
+# End Source File
+# Begin Source File
+
SOURCE=.\RMCast_Reordering.h
# End Source File
# Begin Source File
+SOURCE=.\RMCast_Resend_Handler.h
+# End Source File
+# Begin Source File
+
SOURCE=.\RMCast_Retransmission.h
# End Source File
# Begin Source File
+SOURCE=.\RMCast_Singleton_Factory.h
+# End Source File
+# Begin Source File
+
SOURCE=.\RMCast_UDP_Event_Handler.h
# End Source File
# Begin Source File
SOURCE=.\RMCast_UDP_Proxy.h
# End Source File
+# Begin Source File
+
+SOURCE=.\RMCast_UDP_Reliable_Receiver.h
+# End Source File
+# Begin Source File
+
+SOURCE=.\RMCast_UDP_Reliable_Sender.h
+# End Source File
# End Group
# Begin Group "Inline Files"
@@ -214,6 +262,10 @@ SOURCE=.\RMCast.i # End Source File
# Begin Source File
+SOURCE=.\RMCast_Control_Splitter.i
+# End Source File
+# Begin Source File
+
SOURCE=.\RMCast_Fragment.i
# End Source File
# Begin Source File
@@ -246,20 +298,40 @@ SOURCE=.\RMCast_Reassembly.i # End Source File
# Begin Source File
+SOURCE=.\RMCast_Reliable_Factory.i
+# End Source File
+# Begin Source File
+
SOURCE=.\RMCast_Reordering.i
# End Source File
# Begin Source File
+SOURCE=.\RMCast_Resend_Handler.i
+# End Source File
+# Begin Source File
+
SOURCE=.\RMCast_Retransmission.i
# End Source File
# Begin Source File
+SOURCE=.\RMCast_Singleton_Factory.i
+# End Source File
+# Begin Source File
+
SOURCE=.\RMCast_UDP_Event_Handler.i
# End Source File
# Begin Source File
SOURCE=.\RMCast_UDP_Proxy.i
# End Source File
+# Begin Source File
+
+SOURCE=.\RMCast_UDP_Reliable_Receiver.i
+# End Source File
+# Begin Source File
+
+SOURCE=.\RMCast_UDP_Reliable_Sender.i
+# End Source File
# End Group
# Begin Group "Template Files"
diff --git a/ace/RMCast/RMCast_Control_Splitter.cpp b/ace/RMCast/RMCast_Control_Splitter.cpp new file mode 100644 index 00000000000..b06e9bfc467 --- /dev/null +++ b/ace/RMCast/RMCast_Control_Splitter.cpp @@ -0,0 +1,37 @@ +// $Id$ + +#include "RMCast_Control_Splitter.h" + +#if !defined (__ACE_INLINE__) +# include "RMCast_Control_Splitter.i" +#endif /* ! __ACE_INLINE__ */ + +ACE_RCSID(ace, RMCast_Control_Splitter, "$Id$") + +ACE_RMCast_Control_Splitter::~ACE_RMCast_Control_Splitter (void) +{ +} + +int +ACE_RMCast_Control_Splitter::ack (ACE_RMCast::Ack &ack) +{ + if (this->control_module () != 0) + return this->control_module ()->ack (ack); + return 0; +} + +int +ACE_RMCast_Control_Splitter::join (ACE_RMCast::Join &join) +{ + if (this->control_module () != 0) + return this->control_module ()->join (join); + return 0; +} + +int +ACE_RMCast_Control_Splitter::leave (ACE_RMCast::Leave &leave) +{ + if (this->control_module () != 0) + return this->control_module ()->leave (leave); + return 0; +} diff --git a/ace/RMCast/RMCast_Control_Splitter.h b/ace/RMCast/RMCast_Control_Splitter.h new file mode 100644 index 00000000000..d1c0b5df3a0 --- /dev/null +++ b/ace/RMCast/RMCast_Control_Splitter.h @@ -0,0 +1,49 @@ +/* -*- C++ -*- */ +// $Id$ + +#ifndef ACE_RMCAST_CONTROL_SPLITTER_H +#define ACE_RMCAST_CONTROL_SPLITTER_H +#include "ace/pre.h" + +#include "RMCast_Module.h" + +#if !defined (ACE_LACKS_PRAGMA_ONCE) +# pragma once +#endif /* ACE_LACKS_PRAGMA_ONCE */ + +/// Split control and data messages different modules in the stack +/** + * In some instances the control messages must go a destination + * different from the regular data flow. This class performs this + * separation. + */ +class ACE_RMCast_Export ACE_RMCast_Control_Splitter : public ACE_RMCast_Module +{ +public: + //! Constructor + ACE_RMCast_Control_Splitter (void); + + //! Destructor + virtual ~ACE_RMCast_Control_Splitter (void); + + /// Set the control module, all incoming control messages go to it + void control_module (ACE_RMCast_Module *module); + + /// Return the current control module + ACE_RMCast_Module *control_module (void) const; + + virtual int ack (ACE_RMCast::Ack &); + virtual int join (ACE_RMCast::Join &); + virtual int leave (ACE_RMCast::Leave &); + +private: + /// The control module + ACE_RMCast_Module *control_module_; +}; + +#if defined (__ACE_INLINE__) +#include "RMCast_Control_Splitter.i" +#endif /* __ACE_INLINE__ */ + +#include "ace/post.h" +#endif /* ACE_RMCAST_CONTROL_SPLITTER_H */ diff --git a/ace/RMCast/RMCast_Control_Splitter.i b/ace/RMCast/RMCast_Control_Splitter.i new file mode 100644 index 00000000000..7cc8205fcce --- /dev/null +++ b/ace/RMCast/RMCast_Control_Splitter.i @@ -0,0 +1,21 @@ +// $Id$ + +ACE_INLINE +ACE_RMCast_Control_Splitter::ACE_RMCast_Control_Splitter (void) + : ACE_RMCast_Module () + , control_module_ (0) +{ +} + +ACE_INLINE void +ACE_RMCast_Control_Splitter::control_module (ACE_RMCast_Module *module) +{ + this->control_module_ = module; +} + + +ACE_INLINE ACE_RMCast_Module * +ACE_RMCast_Control_Splitter::control_module (void) const +{ + return this->control_module_; +} diff --git a/ace/RMCast/RMCast_Copy_On_Write.cpp b/ace/RMCast/RMCast_Copy_On_Write.cpp index eeb3b9422c2..39bf16f88a2 100644 --- a/ace/RMCast/RMCast_Copy_On_Write.cpp +++ b/ace/RMCast/RMCast_Copy_On_Write.cpp @@ -87,6 +87,16 @@ ACE_RMCast_Copy_On_Write<KEY,ITEM,C,ITERATOR>::first_key (void) return (*begin).key (); } +template<class KEY, class ITEM, class C, class ITERATOR> int +ACE_RMCast_Copy_On_Write<KEY,ITEM,C,ITERATOR>::empty (void) +{ + Read_Guard ace_mon (*this); + ITERATOR end = ace_mon.collection->collection.end (); + ITERATOR begin = ace_mon.collection->collection.begin (); + + return end == begin; +} + template<class KEY, class ITEM, class C, class I> int ACE_RMCast_Copy_On_Write<KEY,ITEM,C,I>::bind (KEY const & k, ITEM const & i) diff --git a/ace/RMCast/RMCast_Copy_On_Write.h b/ace/RMCast/RMCast_Copy_On_Write.h index d3513a05d2d..605970c7c85 100644 --- a/ace/RMCast/RMCast_Copy_On_Write.h +++ b/ace/RMCast/RMCast_Copy_On_Write.h @@ -181,6 +181,9 @@ public: //! Get the first key KEY first_key (void); + /// Return non-zero if the collection is empty + int empty (void); + //! Add a new element int bind (KEY const & key, ITEM const & item); diff --git a/ace/RMCast/RMCast_IO_UDP.cpp b/ace/RMCast/RMCast_IO_UDP.cpp index 421982d5ad6..4fe1488bad0 100644 --- a/ace/RMCast/RMCast_IO_UDP.cpp +++ b/ace/RMCast/RMCast_IO_UDP.cpp @@ -20,6 +20,19 @@ ACE_RMCast_IO_UDP::~ACE_RMCast_IO_UDP (void) } int +ACE_RMCast_IO_UDP::open (const ACE_INET_Addr &mcast_group, + const ACE_Addr &local, + int protocol_family, + int protocol, + int reuse_addr) +{ + this->mcast_group_ = mcast_group; + + ACE_SOCK_Dgram &dgram = this->dgram_; + return dgram.open (local, protocol_family, protocol, reuse_addr); +} + +int ACE_RMCast_IO_UDP::subscribe (const ACE_INET_Addr &mcast_addr, int reuse_addr, const ACE_TCHAR *net_if, @@ -65,28 +78,6 @@ ACE_RMCast_IO_UDP::handle_events (ACE_Time_Value *tv) } int -ACE_RMCast_IO_UDP::register_handlers (ACE_Reactor *reactor) -{ - this->eh_.reactor (reactor); - return reactor->register_handler (&this->eh_, - ACE_Event_Handler::READ_MASK); -} - -int -ACE_RMCast_IO_UDP::remove_handlers (void) -{ - ACE_Reactor *r = this->eh_.reactor (); - if (r != 0) - { - r->remove_handler (&this->eh_, - ACE_Event_Handler::ALL_EVENTS_MASK - | ACE_Event_Handler::DONT_CALL); - this->eh_.reactor (0); - } - return 0; -} - -int ACE_RMCast_IO_UDP::handle_input (ACE_HANDLE) { // @@ We should use a system constant instead of this literal @@ -102,7 +93,7 @@ ACE_RMCast_IO_UDP::handle_input (ACE_HANDLE) // @@ LOG?? ACE_DEBUG ((LM_DEBUG, "RMCast_IO_UDP::handle_input () - " - "error in recv\n")); + "error in recv %p\n", "")); return -1; } @@ -156,7 +147,7 @@ ACE_RMCast_IO_UDP::handle_input (ACE_HANDLE) // The message type is valid, we must create a new proxy, // initially in the JOINING state... - ACE_RMCast_Module *module = this->factory_->create (this); + ACE_RMCast_Module *module = this->factory_->create (); if (module == 0) { // @@ LOG?? diff --git a/ace/RMCast/RMCast_IO_UDP.h b/ace/RMCast/RMCast_IO_UDP.h index 5af403bf994..e68b360576d 100644 --- a/ace/RMCast/RMCast_IO_UDP.h +++ b/ace/RMCast/RMCast_IO_UDP.h @@ -15,7 +15,6 @@ #include "ace/pre.h" #include "RMCast_Module.h" -#include "RMCast_UDP_Event_Handler.h" #include "ace/SOCK_Dgram_Mcast.h" #include "ace/Hash_Map_Manager.h" #include "ace/Synch.h" @@ -27,7 +26,6 @@ class ACE_RMCast_UDP_Proxy; class ACE_RMCast_Module_Factory; -class ACE_Reactor; class ACE_Time_Value; class ACE_RMCast_Export ACE_RMCast_IO_UDP : public ACE_RMCast_Module @@ -46,6 +44,20 @@ public: //! Destructor ~ACE_RMCast_IO_UDP (void); + /// Open the internal socket, but only to send multicast data. + /** + * It is not clear to me if this is a valid configuration. Maybe it + * would be a problem to expose two separate, incompatible + * interfaces (check the subscribe() method). However, the + * alternative would be to implement almost identical class for + * outgoing and incoming UDP I/O + */ + int open (const ACE_INET_Addr &mcast_group, + const ACE_Addr &local, + int protocol_family = PF_INET, + int protocol = 0, + int reuse_addr = 0); + //! Join a new multicast group /*! * Start receiving data for the <mcast_addr> multicast group. @@ -65,20 +77,6 @@ public: //! forever. int handle_events (ACE_Time_Value *tv = 0); - //! Register any event handlers into <reactor> - /*! - * @@TODO: This should be left for the clients of the class, there - * is no reason why this class must know about reactors. - */ - int register_handlers (ACE_Reactor *reactor); - - //! Remove all the handlers from the reactor - /*! - * @@TODO: This should be left for the clients of the class, there - * is no reason why this class must know about reactors. - */ - int remove_handlers (void); - //! There is data to read, read it and process it. int handle_input (ACE_HANDLE h); @@ -124,9 +122,6 @@ private: typedef ACE_Hash_Map_Manager<ACE_INET_Addr,ACE_RMCast_UDP_Proxy*,ACE_Null_Mutex> Map; //! The collection of proxies Map map_; - - //! The event handler adapter - ACE_RMCast_UDP_Event_Handler eh_; }; #if defined (__ACE_INLINE__) diff --git a/ace/RMCast/RMCast_IO_UDP.i b/ace/RMCast/RMCast_IO_UDP.i index ddacc5694ad..f542e85df70 100644 --- a/ace/RMCast/RMCast_IO_UDP.i +++ b/ace/RMCast/RMCast_IO_UDP.i @@ -4,6 +4,5 @@ ACE_INLINE ACE_RMCast_IO_UDP:: ACE_RMCast_IO_UDP (ACE_RMCast_Module_Factory *factory) : factory_ (factory) - , eh_ (this) { } diff --git a/ace/RMCast/RMCast_Module.cpp b/ace/RMCast/RMCast_Module.cpp index 632d905f900..08ff971f039 100644 --- a/ace/RMCast/RMCast_Module.cpp +++ b/ace/RMCast/RMCast_Module.cpp @@ -30,21 +30,6 @@ ACE_RMCast_Module::next (void) const } int -ACE_RMCast_Module::prev (ACE_RMCast_Module *prev) -{ - if (this->prev_ != 0 && prev != 0) - return 1; - this->prev_ = prev; - return 0; -} - -ACE_RMCast_Module * -ACE_RMCast_Module::prev (void) const -{ - return this->prev_; -} - -int ACE_RMCast_Module::open (void) { return 0; diff --git a/ace/RMCast/RMCast_Module.h b/ace/RMCast/RMCast_Module.h index fad76caac53..d362dac2366 100644 --- a/ace/RMCast/RMCast_Module.h +++ b/ace/RMCast/RMCast_Module.h @@ -48,12 +48,6 @@ public: //! Accesor for the next element in the stack virtual ACE_RMCast_Module* next (void) const; - //! Modifier for the previous element in the stack - virtual int prev (ACE_RMCast_Module *prev); - - //! Accesor for the previous element in the stack - virtual ACE_RMCast_Module* prev (void) const; - //! Initialize the module, setting up the next module virtual int open (void); @@ -84,9 +78,6 @@ public: private: //! The next element in the stack ACE_RMCast_Module *next_; - - //! The previous element in the stack - ACE_RMCast_Module *prev_; }; #if defined (__ACE_INLINE__) diff --git a/ace/RMCast/RMCast_Module.i b/ace/RMCast/RMCast_Module.i index 61099903d20..af299d706e3 100644 --- a/ace/RMCast/RMCast_Module.i +++ b/ace/RMCast/RMCast_Module.i @@ -3,6 +3,5 @@ ACE_INLINE ACE_RMCast_Module::ACE_RMCast_Module (void) : next_ (0) - , prev_ (0) { } diff --git a/ace/RMCast/RMCast_Module_Factory.h b/ace/RMCast/RMCast_Module_Factory.h index f0ea58df0e5..97e6ce29235 100644 --- a/ace/RMCast/RMCast_Module_Factory.h +++ b/ace/RMCast/RMCast_Module_Factory.h @@ -52,7 +52,7 @@ public: virtual ~ACE_RMCast_Module_Factory (void); //! Create a new proxy - virtual ACE_RMCast_Module *create (ACE_RMCast_IO_UDP *) = 0; + virtual ACE_RMCast_Module *create (void) = 0; //! Destroy a proxy /*! diff --git a/ace/RMCast/RMCast_Reliable_Factory.cpp b/ace/RMCast/RMCast_Reliable_Factory.cpp new file mode 100644 index 00000000000..b06b9d0953d --- /dev/null +++ b/ace/RMCast/RMCast_Reliable_Factory.cpp @@ -0,0 +1,46 @@ +// $Id$ + +#include "RMCast_Reliable_Factory.h" +#include "RMCast_Reassembly.h" +#include "RMCast_Reordering.h" + +#if !defined (__ACE_INLINE__) +# include "RMCast_Reliable_Factory.i" +#endif /* ! __ACE_INLINE__ */ + +ACE_RCSID(ace, RMCast_Reliable_Factory, "$Id$") + +ACE_RMCast_Reliable_Factory::~ACE_RMCast_Reliable_Factory (void) +{ +} + +ACE_RMCast_Module* +ACE_RMCast_Reliable_Factory::create (void) +{ + ACE_RMCast_Module *reassembly; + ACE_NEW_RETURN (reassembly, ACE_RMCast_Reassembly, 0); + + ACE_RMCast_Module *reordering; + ACE_NEW_RETURN (reordering, ACE_RMCast_Reordering, 0); + + ACE_RMCast_Module *user = this->factory_->create (); + if (user == 0) + { + delete reordering; + delete reassembly; + return 0; + } + reassembly->next (reordering); + reordering->next (user); + return reassembly; +} + +void +ACE_RMCast_Reliable_Factory::destroy (ACE_RMCast_Module *reassembly) +{ + ACE_RMCast_Module *reordering = reassembly->next (); + ACE_RMCast_Module *user = reordering->next (); + this->factory_->destroy (user); + delete reordering; + delete reassembly; +} diff --git a/ace/RMCast/RMCast_Reliable_Factory.h b/ace/RMCast/RMCast_Reliable_Factory.h new file mode 100644 index 00000000000..e41c7fd9e74 --- /dev/null +++ b/ace/RMCast/RMCast_Reliable_Factory.h @@ -0,0 +1,60 @@ +/* -*- C++ -*- */ +// $Id$ + +// ============================================================================ +// +// = LIBRARY +// ace +// +// = FILENAME +// RMCast_Reliable_Factory.h +// +// = AUTHOR +// Carlos O'Ryan <coryan@uci.edu> +// +// ============================================================================ + +#ifndef ACE_RMCAST_RELIABLE_FACTORY_H +#define ACE_RMCAST_RELIABLE_FACTORY_H +#include "ace/pre.h" + +#include "RMCast_Module_Factory.h" + +#if !defined (ACE_LACKS_PRAGMA_ONCE) +# pragma once +#endif /* ACE_LACKS_PRAGMA_ONCE */ + +/// Implement an ACE_RMCast_Module_Factory that "creates" a single +/// object. +/** + * Many applications (and even some internal components), will use a + * single ACE_RMCast_Module to process all the events, for example, a + * receiver may decide to use the same ACE_RMCast_Module to process + * all incoming events, instead of using one per remote sender. + */ +class ACE_RMCast_Export ACE_RMCast_Reliable_Factory : public ACE_RMCast_Module_Factory +{ +public: + /// Constructor + /** + * The create() method will return always \param reliable. + */ + ACE_RMCast_Reliable_Factory (ACE_RMCast_Module_Factory *factory); + + //! Destructor + virtual ~ACE_RMCast_Reliable_Factory (void); + + virtual ACE_RMCast_Module *create (void); + virtual void destroy (ACE_RMCast_Module *); + +private: + /// Delegate on another factory to create the user module + ACE_RMCast_Module_Factory *factory_; +}; + +#if defined (__ACE_INLINE__) +#include "RMCast_Reliable_Factory.i" +#endif /* __ACE_INLINE__ */ + +#include "ace/post.h" +#endif /* ACE_RMCAST_RELIABLE_FACTORY_H */ diff --git a/ace/RMCast/RMCast_Reliable_Factory.i b/ace/RMCast/RMCast_Reliable_Factory.i new file mode 100644 index 00000000000..47ba22754ec --- /dev/null +++ b/ace/RMCast/RMCast_Reliable_Factory.i @@ -0,0 +1,8 @@ +// $Id$ + +ACE_INLINE +ACE_RMCast_Reliable_Factory:: + ACE_RMCast_Reliable_Factory (ACE_RMCast_Module_Factory *factory) + : factory_ (factory) +{ +} diff --git a/ace/RMCast/RMCast_Resend_Handler.cpp b/ace/RMCast/RMCast_Resend_Handler.cpp new file mode 100644 index 00000000000..130a1d6b5ca --- /dev/null +++ b/ace/RMCast/RMCast_Resend_Handler.cpp @@ -0,0 +1,22 @@ +// $Id$ + +#include "RMCast_Resend_Handler.h" +#include "RMCast_Retransmission.h" + +#if !defined (__ACE_INLINE__) +# include "RMCast_Resend_Handler.i" +#endif /* ! __ACE_INLINE__ */ + +ACE_RCSID(ace, RMCast_Resend_Handler, "$Id$") + +ACE_RMCast_Resend_Handler::~ACE_RMCast_Resend_Handler (void) +{ +} + +int +ACE_RMCast_Resend_Handler::handle_timeout (const ACE_Time_Value &, + const void *) +{ + (void) this->retransmission_->resend (0 /* @@ TODO */); + return 0; +} diff --git a/ace/RMCast/RMCast_Resend_Handler.h b/ace/RMCast/RMCast_Resend_Handler.h new file mode 100644 index 00000000000..0d2ec0fe13c --- /dev/null +++ b/ace/RMCast/RMCast_Resend_Handler.h @@ -0,0 +1,44 @@ +// $Id$ + +#ifndef ACE_RMCAST_RESEND_HANDLER_H +#define ACE_RMCAST_RESEND_HANDLER_H +#include "ace/pre.h" + +#include "RMCast_Export.h" +#include "ace/Event_Handler.h" + +#if !defined (ACE_LACKS_PRAGMA_ONCE) +# pragma once +#endif /* ACE_LACKS_PRAGMA_ONCE */ + +class ACE_RMCast_Retransmission; + +/// Implement an adapter to resend messages in the +/// ACE_RMCast_Retransmission layer, but based on Reactor based +/// timeouts. +class ACE_RMCast_Export ACE_RMCast_Resend_Handler : public ACE_Event_Handler +{ +public: + /// Constructor, save io_udp as the Adaptee in the Adapter pattern. + ACE_RMCast_Resend_Handler (ACE_RMCast_Retransmission *retransmission); + + /// Destructor + ~ACE_RMCast_Resend_Handler (void); + + //@{ + //! Documented in ACE_Event_Handler class + virtual int handle_timeout (const ACE_Time_Value ¤t_time, + const void *act = 0); + //@} + +private: + //! The adaptee + ACE_RMCast_Retransmission *retransmission_; +}; + +#if defined (__ACE_INLINE__) +#include "RMCast_Resend_Handler.i" +#endif /* __ACE_INLINE__ */ + +#include "ace/post.h" +#endif /* ACE_RMCAST_RESEND_HANDLER_H */ diff --git a/ace/RMCast/RMCast_Resend_Handler.i b/ace/RMCast/RMCast_Resend_Handler.i new file mode 100644 index 00000000000..9ad6cd8870d --- /dev/null +++ b/ace/RMCast/RMCast_Resend_Handler.i @@ -0,0 +1,8 @@ +// $Id$ + +ACE_INLINE +ACE_RMCast_Resend_Handler:: +ACE_RMCast_Resend_Handler (ACE_RMCast_Retransmission *r) + : retransmission_ (r) +{ +} diff --git a/ace/RMCast/RMCast_Retransmission.cpp b/ace/RMCast/RMCast_Retransmission.cpp index c9db70cbf63..a23829cd54e 100644 --- a/ace/RMCast/RMCast_Retransmission.cpp +++ b/ace/RMCast/RMCast_Retransmission.cpp @@ -67,6 +67,26 @@ ACE_RMCast_Retransmission::resend (ACE_UINT32 max_sequence_number) } int +ACE_RMCast_Retransmission::resend_all (void) +{ + if (this->next () == 0) + return 0; + + ACE_RMCast_Resend_Worker worker (this->next (), ACE_UINT32_MAX); + + if (this->messages_.for_each (&worker) == -1) + return -1; + + return worker.n; +} + +int +ACE_RMCast_Retransmission::has_data (void) +{ + return !this->messages_.empty (); +} + +int ACE_RMCast_Retransmission::close (void) { // @@ diff --git a/ace/RMCast/RMCast_Retransmission.h b/ace/RMCast/RMCast_Retransmission.h index b7bc20d2914..f49c0a73f9c 100644 --- a/ace/RMCast/RMCast_Retransmission.h +++ b/ace/RMCast/RMCast_Retransmission.h @@ -59,6 +59,15 @@ public: */ int resend (ACE_UINT32 max_sequence_number); + //! Resend all messages + /*! + * Resends all the messages currently in the queue. + */ + int resend_all (void); + + /// Return 0 if there is no pending data to send + int has_data (void); + //! Cleanup all the stored messages virtual int close (void); diff --git a/ace/RMCast/RMCast_Singleton_Factory.cpp b/ace/RMCast/RMCast_Singleton_Factory.cpp new file mode 100644 index 00000000000..403b87060e2 --- /dev/null +++ b/ace/RMCast/RMCast_Singleton_Factory.cpp @@ -0,0 +1,24 @@ +// $Id$ + +#include "RMCast_Singleton_Factory.h" + +#if !defined (__ACE_INLINE__) +# include "RMCast_Singleton_Factory.i" +#endif /* ! __ACE_INLINE__ */ + +ACE_RCSID(ace, RMCast_Singleton_Factory, "$Id$") + +ACE_RMCast_Singleton_Factory::~ACE_RMCast_Singleton_Factory (void) +{ +} + +ACE_RMCast_Module* +ACE_RMCast_Singleton_Factory::create (void) +{ + return this->singleton_; +} + +void +ACE_RMCast_Singleton_Factory::destroy (ACE_RMCast_Module *) +{ +} diff --git a/ace/RMCast/RMCast_Singleton_Factory.h b/ace/RMCast/RMCast_Singleton_Factory.h new file mode 100644 index 00000000000..5e8b27a8e0b --- /dev/null +++ b/ace/RMCast/RMCast_Singleton_Factory.h @@ -0,0 +1,60 @@ +/* -*- C++ -*- */ +// $Id$ + +// ============================================================================ +// +// = LIBRARY +// ace +// +// = FILENAME +// RMCast_Singleton_Factory.h +// +// = AUTHOR +// Carlos O'Ryan <coryan@uci.edu> +// +// ============================================================================ + +#ifndef ACE_RMCAST_SINGLETON_FACTORY_H +#define ACE_RMCAST_SINGLETON_FACTORY_H +#include "ace/pre.h" + +#include "RMCast_Module_Factory.h" + +#if !defined (ACE_LACKS_PRAGMA_ONCE) +# pragma once +#endif /* ACE_LACKS_PRAGMA_ONCE */ + +/// Implement an ACE_RMCast_Module_Factory that "creates" a single +/// object. +/** + * Many applications (and even some internal components), will use a + * single ACE_RMCast_Module to process all the events, for example, a + * receiver may decide to use the same ACE_RMCast_Module to process + * all incoming events, instead of using one per remote sender. + */ +class ACE_RMCast_Export ACE_RMCast_Singleton_Factory : public ACE_RMCast_Module_Factory +{ +public: + /// Constructor + /** + * The create() method will return always \param singleton. + */ + ACE_RMCast_Singleton_Factory (ACE_RMCast_Module *singleton); + + //! Destructor + virtual ~ACE_RMCast_Singleton_Factory (void); + + virtual ACE_RMCast_Module *create (void); + virtual void destroy (ACE_RMCast_Module *); + +private: + /// The singleton object + ACE_RMCast_Module *singleton_; +}; + +#if defined (__ACE_INLINE__) +#include "RMCast_Singleton_Factory.i" +#endif /* __ACE_INLINE__ */ + +#include "ace/post.h" +#endif /* ACE_RMCAST_SINGLETON_FACTORY_H */ diff --git a/ace/RMCast/RMCast_Singleton_Factory.i b/ace/RMCast/RMCast_Singleton_Factory.i new file mode 100644 index 00000000000..a45837dfa93 --- /dev/null +++ b/ace/RMCast/RMCast_Singleton_Factory.i @@ -0,0 +1,8 @@ +// $Id$ + +ACE_INLINE +ACE_RMCast_Singleton_Factory:: + ACE_RMCast_Singleton_Factory (ACE_RMCast_Module *singleton) + : singleton_ (singleton) +{ +} diff --git a/ace/RMCast/RMCast_UDP_Event_Handler.cpp b/ace/RMCast/RMCast_UDP_Event_Handler.cpp index e5ff8da2761..963928845f7 100644 --- a/ace/RMCast/RMCast_UDP_Event_Handler.cpp +++ b/ace/RMCast/RMCast_UDP_Event_Handler.cpp @@ -9,7 +9,7 @@ # include "RMCast_UDP_Event_Handler.i" #endif /* ! __ACE_INLINE__ */ -ACE_RCSID(ace, RMCast_Event_Handler, "$Id$") +ACE_RCSID(ace, RMCast_UDP_Event_Handler, "$Id$") ACE_RMCast_UDP_Event_Handler::~ACE_RMCast_UDP_Event_Handler (void) { diff --git a/ace/RMCast/RMCast_UDP_Proxy.h b/ace/RMCast/RMCast_UDP_Proxy.h index 02d4e3409a9..f1a7f77a25a 100644 --- a/ace/RMCast/RMCast_UDP_Proxy.h +++ b/ace/RMCast/RMCast_UDP_Proxy.h @@ -28,28 +28,41 @@ class ACE_RMCast_IO_UDP; +/// Define the proxy implementation for UDP based communication +/** + * Proxy objects are transport specific, they are responsible for + * storing the remote peer addressing information. + * This class implements the UDP version of a proxy. + */ class ACE_RMCast_Export ACE_RMCast_UDP_Proxy : public ACE_RMCast_Proxy { - // = TITLE - // Proxy for UDP-based RMCast - // - // = DESCRIPTION - // public: + /// Constructor + /** + * The \param io_udp argument is kept to send the replys through the + * right socket. + * The \param peer_addr is the address used byu the peer to receive + * responses. + */ ACE_RMCast_UDP_Proxy (ACE_RMCast_IO_UDP *io_udp, const ACE_INET_Addr &peer_addr); - // Constructor + /// Destructor virtual ~ACE_RMCast_UDP_Proxy (void); - // Destructor + /// Receive a message, parse and send it upstream in the right + /// format. int receive_message (char *buffer, size_t size); - // Receive the message + /// Make the peer address available const ACE_INET_Addr &peer_addr (void) const; - // The address of the peer - // = The ACE_RMCast_Proxy methods + //@{ + /** + * Implement the ACE_RMCast_Proxy methods, in this case we use the + * \param io_udp_ object to send the data, using the address of our + * remote peer. + */ virtual int reply_data (ACE_RMCast::Data &); virtual int reply_poll (ACE_RMCast::Poll &); virtual int reply_ack_join (ACE_RMCast::Ack_Join &); @@ -57,13 +70,14 @@ public: virtual int reply_ack (ACE_RMCast::Ack &); virtual int reply_join (ACE_RMCast::Join &); virtual int reply_leave (ACE_RMCast::Leave &); + //@} private: + /// The IO facade ACE_RMCast_IO_UDP *io_udp_; - // The IO facade + /// The remote peer's address ACE_INET_Addr peer_addr_; - // The address of the peer }; #if defined (__ACE_INLINE__) diff --git a/ace/RMCast/RMCast_UDP_Reliable_Receiver.cpp b/ace/RMCast/RMCast_UDP_Reliable_Receiver.cpp new file mode 100644 index 00000000000..8a99b4cee6b --- /dev/null +++ b/ace/RMCast/RMCast_UDP_Reliable_Receiver.cpp @@ -0,0 +1,31 @@ +// $Id$ + +#include "RMCast_UDP_Reliable_Receiver.h" +#include "RMCast_UDP_Event_Handler.h" + +#if !defined (__ACE_INLINE__) +# include "RMCast_UDP_Reliable_Receiver.i" +#endif /* ! __ACE_INLINE__ */ + +ACE_RCSID(ace, RMCast_UDP_Reliable_Receiver, "$Id$") + +ACE_RMCast_UDP_Reliable_Receiver::ACE_RMCast_UDP_Reliable_Receiver (ACE_RMCast_Module *user_module) + : user_factory_ (user_module) + , factory_ (&user_factory_) + , io_udp_ (&factory_) +{ +} + +ACE_RMCast_UDP_Reliable_Receiver::~ACE_RMCast_UDP_Reliable_Receiver (void) +{ +} + +void +ACE_RMCast_UDP_Reliable_Receiver::reactive_incoming_messages (ACE_Reactor *reactor) +{ + ACE_RMCast_UDP_Event_Handler *eh; + ACE_NEW (eh, ACE_RMCast_UDP_Event_Handler (&this->io_udp_)); + + /// @@ TODO Make sure it is removed from the Reactor at some point + (void) reactor->register_handler (eh, ACE_Event_Handler::READ_MASK); +} diff --git a/ace/RMCast/RMCast_UDP_Reliable_Receiver.h b/ace/RMCast/RMCast_UDP_Reliable_Receiver.h new file mode 100644 index 00000000000..2a580afbed8 --- /dev/null +++ b/ace/RMCast/RMCast_UDP_Reliable_Receiver.h @@ -0,0 +1,49 @@ +/* -*- C++ -*- */ +// $Id$ + +#ifndef ACE_RMCAST_UDP_RELIABLE_RECEIVER_H +#define ACE_RMCAST_UDP_RELIABLE_RECEIVER_H +#include "ace/pre.h" + +#include "RMCast_Singleton_Factory.h" +#include "RMCast_Reliable_Factory.h" +#include "RMCast_IO_UDP.h" + +#if !defined (ACE_LACKS_PRAGMA_ONCE) +# pragma once +#endif /* ACE_LACKS_PRAGMA_ONCE */ + +class ACE_Reactor; + +class ACE_RMCast_Export ACE_RMCast_UDP_Reliable_Receiver : public ACE_RMCast_Module +{ +public: + /// Constructor + ACE_RMCast_UDP_Reliable_Receiver (ACE_RMCast_Module *user_control); + + /// Destructor + virtual ~ACE_RMCast_UDP_Reliable_Receiver (void); + + /// Open the UDP I/O module. + int open (const ACE_INET_Addr &mcast_group); + + /// Use the reactor to handle incoming messages + void reactive_incoming_messages (ACE_Reactor *reactor); + +private: + /// All the proxys give their messages to user module + ACE_RMCast_Singleton_Factory user_factory_; + + /// This factory creates the per-proxy stack + ACE_RMCast_Reliable_Factory factory_; + + /// Handle all the UDP I/O + ACE_RMCast_IO_UDP io_udp_; +}; + +#if defined (__ACE_INLINE__) +#include "RMCast_UDP_Reliable_Receiver.i" +#endif /* __ACE_INLINE__ */ + +#include "ace/post.h" +#endif /* ACE_RMCAST_UDP_RELIABLE_RECEIVER_H */ diff --git a/ace/RMCast/RMCast_UDP_Reliable_Receiver.i b/ace/RMCast/RMCast_UDP_Reliable_Receiver.i new file mode 100644 index 00000000000..29a212adfeb --- /dev/null +++ b/ace/RMCast/RMCast_UDP_Reliable_Receiver.i @@ -0,0 +1,7 @@ +// $Id$ + +ACE_INLINE int +ACE_RMCast_UDP_Reliable_Receiver::open (const ACE_INET_Addr &mcast_group) +{ + return this->io_udp_.subscribe (mcast_group); +} diff --git a/ace/RMCast/RMCast_UDP_Reliable_Sender.cpp b/ace/RMCast/RMCast_UDP_Reliable_Sender.cpp new file mode 100644 index 00000000000..4659050ebb3 --- /dev/null +++ b/ace/RMCast/RMCast_UDP_Reliable_Sender.cpp @@ -0,0 +1,66 @@ +// $Id$ + +#include "RMCast_UDP_Reliable_Sender.h" +#include "RMCast_UDP_Event_Handler.h" +#include "RMCast_Resend_Handler.h" + +#if !defined (__ACE_INLINE__) +# include "RMCast_UDP_Reliable_Sender.i" +#endif /* ! __ACE_INLINE__ */ + +ACE_RCSID(ace, RMCast_UDP_Reliable_Sender, "$Id$") + +ACE_RMCast_UDP_Reliable_Sender::ACE_RMCast_UDP_Reliable_Sender (ACE_RMCast_Module *user_control) + : user_control_ (user_control) + + // We use a singleton factory, all proxys send their messages to the + // retransmission module. There should be only control messages + // coming back, so this is OK. + , factory_ (&retransmission_) + , io_udp_ (&factory_) +{ + // Messages are passed down to the retransmission module. + this->next (&this->retransmission_); + + // Then to the splitter, at this point control messages are sent + // back to the user, other messages continue down to the + // fragmentation layer. + this->retransmission_.next (&this->splitter_); + this->splitter_.next (&this->fragment_); + this->splitter_.control_module (this->user_control_); + + // The fragmentation layer delegates all messages to the UDP I/O + // module, that sends every message back to the application. + this->fragment_.next (&this->io_udp_); +} + +ACE_RMCast_UDP_Reliable_Sender::~ACE_RMCast_UDP_Reliable_Sender (void) +{ +} + +void +ACE_RMCast_UDP_Reliable_Sender::reactive_incoming_messages (ACE_Reactor *reactor) +{ + ACE_RMCast_UDP_Event_Handler *eh; + ACE_NEW (eh, ACE_RMCast_UDP_Event_Handler (&this->io_udp_)); + + /// @@ TODO Make sure it is removed from the Reactor at some point + (void) reactor->register_handler (eh, ACE_Event_Handler::READ_MASK); +} + +void +ACE_RMCast_UDP_Reliable_Sender::reactive_resends (ACE_Reactor *reactor, + const ACE_Time_Value &period) +{ + ACE_RMCast_Resend_Handler *eh; + ACE_NEW (eh, ACE_RMCast_Resend_Handler (&this->retransmission_)); + + /// @@ TODO make sure it is removed from the Reactor at some point + (void) reactor->schedule_timer (eh, 0, period, period); +} + +int +ACE_RMCast_UDP_Reliable_Sender::has_data (void) +{ + return this->retransmission_.has_data (); +} diff --git a/ace/RMCast/RMCast_UDP_Reliable_Sender.h b/ace/RMCast/RMCast_UDP_Reliable_Sender.h new file mode 100644 index 00000000000..3269c7cb304 --- /dev/null +++ b/ace/RMCast/RMCast_UDP_Reliable_Sender.h @@ -0,0 +1,67 @@ +/* -*- C++ -*- */ +// $Id$ + +#ifndef ACE_RMCAST_UDP_RELIABLE_SENDER_H +#define ACE_RMCAST_UDP_RELIABLE_SENDER_H +#include "ace/pre.h" + +#include "RMCast_Singleton_Factory.h" +#include "RMCast_IO_UDP.h" +#include "RMCast_Retransmission.h" +#include "RMCast_Fragment.h" +#include "RMCast_Control_Splitter.h" + +#if !defined (ACE_LACKS_PRAGMA_ONCE) +# pragma once +#endif /* ACE_LACKS_PRAGMA_ONCE */ + +class ACE_Reactor; + +class ACE_RMCast_Export ACE_RMCast_UDP_Reliable_Sender : public ACE_RMCast_Module +{ +public: + /// Constructor + ACE_RMCast_UDP_Reliable_Sender (ACE_RMCast_Module *user_control); + + /// Destructor + virtual ~ACE_RMCast_UDP_Reliable_Sender (void); + + /// Open the UDP I/O module. + int open (const ACE_INET_Addr &mcast_group); + + /// Use the reactor to handle incoming messages + void reactive_incoming_messages (ACE_Reactor *reactor); + + /// Use the reactor to periodically resend messages + void reactive_resends (ACE_Reactor *reactor, + const ACE_Time_Value &period); + + /// Check if there is still some messages to send, return 0 if not. + int has_data (void); + +private: + /// The application-level control module + ACE_RMCast_Module *user_control_; + + /// The retransmission module + ACE_RMCast_Retransmission retransmission_; + + /// All the proxys give their messages to the retransmission module + ACE_RMCast_Singleton_Factory factory_; + + /// Handle all the UDP I/O + ACE_RMCast_IO_UDP io_udp_; + + /// The fragmentation module + ACE_RMCast_Fragment fragment_; + + /// Redirect control messages to the user supplied module + ACE_RMCast_Control_Splitter splitter_; +}; + +#if defined (__ACE_INLINE__) +#include "RMCast_UDP_Reliable_Sender.i" +#endif /* __ACE_INLINE__ */ + +#include "ace/post.h" +#endif /* ACE_RMCAST_UDP_RELIABLE_SENDER_H */ diff --git a/ace/RMCast/RMCast_UDP_Reliable_Sender.i b/ace/RMCast/RMCast_UDP_Reliable_Sender.i new file mode 100644 index 00000000000..bab679aa0f2 --- /dev/null +++ b/ace/RMCast/RMCast_UDP_Reliable_Sender.i @@ -0,0 +1,7 @@ +// $Id$ + +ACE_INLINE int +ACE_RMCast_UDP_Reliable_Sender::open (const ACE_INET_Addr &mcast_group) +{ + return this->io_udp_.open (mcast_group, ACE_Addr::sap_any); +} |