*** empty log message ***

10 files changed, 537 insertions, 95 deletions

diff --git a/ace/Makefile b/ace/Makefile
index 058faa2089b..7bc9e82ec79 100644
--- a/ace/Makefile
+++ b/ace/Makefile
@@ -198,7 +198,8 @@ OTHER_FILES = \
   Remote_Name_Space \
   QoS_Session_Impl \
   QoS_Session_Factory \
-  QoS_Manager
+  QoS_Manager \
+  QoS_Decorator
 
 TEMPLATE_FILES = \
            Acceptor \
diff --git a/ace/QoS_Decorator.cpp b/ace/QoS_Decorator.cpp
new file mode 100644
index 00000000000..7291cd38ef8
--- /dev/null
+++ b/ace/QoS_Decorator.cpp
@@ -0,0 +1,167 @@
+// QoS_Decorator.cpp
+// $Id$
+
+#include "ace/QoS_Decorator.h"
+
+ACE_RCSID(ace, QoS_Decorator, "$Id $")
+
+ACE_ALLOC_HOOK_DEFINE(ACE_QOS_DECORATOR)
+
+// Constructor.
+ACE_QoS_Decorator_Base::ACE_QoS_Decorator_Base (void)
+{}
+
+// Constructor.
+ACE_QoS_Decorator_Base::ACE_QoS_Decorator_Base (ACE_Event_Handler 
+                                                *event_handler)
+  : event_handler_ (event_handler)
+{
+}
+  
+// Destructor.
+ACE_QoS_Decorator_Base::~ACE_QoS_Decorator_Base (void)
+{
+}
+
+// Forward the call to ACE_Event_Handler component.
+ACE_HANDLE
+ACE_QoS_Decorator_Base::get_handle (void) const
+{
+ return this->event_handler_->get_handle ();
+}
+
+// Forward the call to ACE_Event_Handler component.
+int
+ACE_QoS_Decorator_Base::handle_input (ACE_HANDLE fd)
+{
+  return this->event_handler_->handle_input (fd);
+}
+
+// Forward the call to ACE_Event_Handler component.
+int
+ACE_QoS_Decorator_Base::handle_qos (ACE_HANDLE fd)
+{
+  return this->event_handler_->handle_qos (fd);
+}
+
+// Constructor.
+ACE_QoS_Decorator::ACE_QoS_Decorator (void)
+{}
+
+// Constructor.
+ACE_QoS_Decorator::ACE_QoS_Decorator (ACE_Event_Handler *event_handler,
+                                      ACE_QoS_Session *qos_session,
+                                      ACE_Reactor *reactor)
+  : qos_session_ (qos_session), 
+    reactor_ (reactor)
+{
+  ACE_NEW (this->decorator_base_,
+           ACE_QoS_Decorator_Base (event_handler));
+
+  ACE_NEW (this->qos_event_handler_,
+           ACE_QoS_Event_Handler (this->decorator_base_));
+}
+
+// Destructor.
+ACE_QoS_Decorator::~ACE_QoS_Decorator (void)
+{
+  delete this->decorator_base_;
+  delete this->qos_event_handler_;
+}
+
+// Implements the undecorated functionality. This is sufficient for
+// GQoS. RAPI needs additional QoS decoration. This is done by the
+// ACE_QoS_Event_Handler class.
+ACE_HANDLE
+ACE_QoS_Decorator::get_handle (void) const
+{
+  return this->decorator_base_->get_handle ();
+}
+
+// Implements the undecorated functionality. This is sufficient for
+// GQoS. RAPI needs additional QoS decoration. This is done by the
+// ACE_QoS_Event_Handler class.
+int
+ACE_QoS_Decorator::handle_input (ACE_HANDLE fd)
+{
+  return this->decorator_base_->handle_input (fd);
+}
+
+// Implements the undecorated functionality. This is sufficient for
+// GQoS. RAPI needs additional QoS decoration. This is done by the
+// ACE_QoS_Event_Handler class.
+int
+ACE_QoS_Decorator::handle_qos (ACE_HANDLE fd)
+{
+  return this->decorator_base_->handle_qos (fd);
+}
+
+// This method registers the RAPI QoS event handler with the reactor
+// if the application is using RAPI. Note that it is a no-op for GQoS
+// because an extra socket for handling QoS events is not required.
+int
+ACE_QoS_Decorator::init (void)
+{
+#if defined (ACE_HAS_RAPI)
+
+  // Pass the QoS session to QoS Event Handler.
+  this->qos_event_handler_->qos_session (this->qos_session_);
+
+  // Register the QoS Event Handler with the Reactor.
+  return this->reactor_->register_handler (this->qos_event_handler_,
+                                           ACE_Event_Handler::READ_MASK);
+#endif
+  return 0;
+
+}
+
+// Constructor.
+ACE_QoS_Event_Handler::ACE_QoS_Event_Handler (void)
+{
+}
+
+// Constructor.
+ACE_QoS_Event_Handler::ACE_QoS_Event_Handler (ACE_QoS_Decorator_Base 
+                                              *decorator_base)
+  : decorator_base_ (decorator_base)
+{
+};
+  
+// Destructor.
+ACE_QoS_Event_Handler::~ACE_QoS_Event_Handler (void)
+{
+}
+
+// Set the QoS session.
+void
+ACE_QoS_Event_Handler::qos_session (ACE_QoS_Session *qos_session)
+{
+  this->qos_session_ = qos_session;
+}
+
+// Returns the RAPI file descriptor for listening to RAPI evnets.
+ACE_HANDLE
+ACE_QoS_Event_Handler::get_handle (void) const
+{
+  return this->qos_session_->rsvp_events_handle ();
+}
+
+// Note, here the handle_input () calls the handle_qos () of the
+// Decorator Base which then calls handle_qos () of the
+// ACE_Event_Handler component within it. This helps to translate the
+// normal read events into qos events in case of RAPI so the
+// application using the API is oblivious to the fact that in RAPI,
+// QoS events are received on a different socket. This helps to
+// maintain a uniform design for the application irrespective of
+// whether it is using RAPI or GQoS.
+int
+ACE_QoS_Event_Handler::handle_input (ACE_HANDLE fd)
+{
+  return this->decorator_base_->handle_qos (fd);
+}
+
+
+
+
+
+
diff --git a/ace/QoS_Decorator.h b/ace/QoS_Decorator.h
new file mode 100644
index 00000000000..3fe6e38398c
--- /dev/null
+++ b/ace/QoS_Decorator.h
@@ -0,0 +1,173 @@
+/* -*- C++ -*- */
+// $Id$
+
+// ============================================================================
+//
+// = LIBRARY
+//    ACE_wrappers/ace
+//
+// = FILENAME
+//    QOS_Decorator.h
+//
+// = AUTHOR
+//    Vishal Kachroo <vishal@cs.wustl.edu>
+//
+// ============================================================================
+
+#ifndef QOS_DECORATOR_H
+#define QOS_DECORATOR_H
+#include "ace/pre.h"
+
+#include "ace/Reactor.h"
+
+#if !defined (ACE_LACKS_PRAGMA_ONCE)
+# pragma once
+#endif /* ACE_LACKS_PRAGMA_ONCE */
+
+#include "ace/INET_Addr.h"
+#include "ace/Event_Handler.h"
+#include "ace/SOCK_Dgram_Mcast_QoS.h"
+
+ACE_RCSID(QOS_Decorator, QOS_Decorator, "$Id$")
+  
+class ACE_QoS_Decorator_Base : public ACE_Event_Handler
+{
+  // = TITLE
+  //     This class is the Decorator Pattern Base class for decorating
+  //     ACE_Event_Handler.
+  //
+  // = DESCRIPTION
+  //       It simply forwards the requests for get_handle (), 
+  //       handle_input () and handle_qos () to its event_handler_ 
+  //       component. Concrete decorators for ACE_Event_Handler will use 
+  //       this class to access the basic event handler functionality and 
+  //       decorate that by their own implementation.
+  
+public:
+  
+  // Initialization and termination methods.
+  ACE_QoS_Decorator_Base (void);
+  // Constructor.
+
+  ACE_QoS_Decorator_Base (ACE_Event_Handler *event_handler);
+  // Constructor.
+
+  ~ACE_QoS_Decorator_Base (void);
+  // Destructor.
+
+  virtual ACE_HANDLE get_handle (void) const;
+  // Forwards the request to its event_handler_ component.
+
+  virtual int handle_input (ACE_HANDLE fd);
+  // Forwards the request to its event_handler_ component.
+
+  virtual int handle_qos (ACE_HANDLE fd);
+  // Forwards the request to its event_handler_ component.
+
+private:
+  
+  ACE_Event_Handler *event_handler_;
+  // The event handler that is decorated by this class.
+
+};
+
+class ACE_QoS_Event_Handler : public ACE_Event_Handler 
+{
+  // = TITLE
+  //     This Handler is registered with the Reactor for QoS events.
+  //
+  // = DESCRIPTION
+  //       Concrete QoS decorator uses this class to receive QoS events 
+  //       for RAPI.  It hides the application from knowing that it is 
+  //       receiving QoS events on a different socket so the application
+  //       doesnt have to be designed differently for RAPI and GQoS.
+
+  ~ACE_QoS_Event_Handler (void);
+  // Destructor.
+
+  virtual ACE_HANDLE get_handle (void) const;
+  // Returns the RAPI file descriptor for receiving QoS events.
+
+  virtual int handle_input (ACE_HANDLE fd);
+  // Calls the base class handle_input ().
+
+  void qos_session (ACE_QoS_Session *qos_session);
+  // Sets the QoS session.
+
+  friend class ACE_QoS_Decorator;
+
+private:
+
+  ACE_QoS_Event_Handler (void);
+  // Constructor is private because only ACE_QoS_Decorator should
+  // create this object.
+
+  ACE_QoS_Event_Handler (ACE_QoS_Decorator_Base *decorator_base);
+  // The QoS Decorator passes in its base for this handler to use.
+
+  ACE_QoS_Session *qos_session_;
+  // Used to get to the RAPI file descriptor for QoS Events.
+
+  ACE_QoS_Decorator_Base *decorator_base_;
+  // Requests on the class are forwarded to this base class;
+
+};
+
+class ACE_QoS_Decorator : public ACE_QoS_Decorator_Base
+{
+  // = TITLE
+  //     Concrete QoS Decorator.
+  //
+  // = DESCRIPTION
+  //       Decorates the ACE_Event_Handler to additionally handle QoS 
+  //       events uniformly for different QoS mechanisms like RAPI and 
+  //       GQoS.
+
+public:
+
+  // Initialization and termination methods.
+  ACE_QoS_Decorator (void);
+  // Constructor.
+
+  ACE_QoS_Decorator (ACE_Event_Handler *event_handler,
+                     ACE_QoS_Session *qos_session,
+                     ACE_Reactor *reactor = ACE_Reactor::instance ());
+  // Constructor.
+
+  ~ACE_QoS_Decorator (void);
+  // Destructor.
+
+  virtual ACE_HANDLE get_handle (void) const;
+  // Calls the base class get_handle ().
+
+  virtual int handle_input (ACE_HANDLE fd);
+  // Calls the base class handle_input ().
+
+  virtual int handle_qos (ACE_HANDLE fd);
+  // Calls the base class handle_qos ().
+  
+  int init (void);
+  // This method registers the QoS Event Handler with the Reactor
+  // to receive RAPI events.
+
+private:
+
+  ACE_QoS_Decorator_Base *decorator_base_;
+  // Requests on the class are forwarded to this base class;
+
+  ACE_QoS_Event_Handler *qos_event_handler_;
+  // Handles the QoS events and in that sense decorates the usual
+  // ACE_Event_Handler.
+
+  ACE_QoS_Session *qos_session_;
+  // Passed to the ACE_QoS_Event_Handler for retrieving the RAPI
+  // session specific information like rapi_fd.
+
+  ACE_Reactor *reactor_;
+  // If the application wants to use an instance of Reactor other
+  // than the Singleton one.
+
+};
+
+#include "ace/post.h"
+#endif  /* QOS_DECORATOR_H */
diff --git a/ace/QoS_Session.h b/ace/QoS_Session.h
index 0e145c432c4..cdbef4db6a8 100644
--- a/ace/QoS_Session.h
+++ b/ace/QoS_Session.h
@@ -49,7 +49,8 @@ public:
     ACE_QOS_RECEIVER,
     ACE_QOS_BOTH
   };
-  // A flag to indicate if this endpoint is a sender or a receiver or both.
+  // A flag to indicate if this endpoint is a sender or a receiver or
+  // both.
 
   virtual ~ACE_QoS_Session (void) {};
   // to shutup g++.
@@ -67,21 +68,23 @@ public:
   virtual int qos (ACE_SOCK *socket,
                    ACE_QoS_Manager *qos_manager,
                    const ACE_QoS &ace_qos) = 0;
-  // Set QoS for the current session. The socket parameter is used to confirm if
-  // this QoS session was subscribed to by the socket.
-
+  // Set QoS for the current session. The qos manager is used to
+  // confirm if this QoS session was subscribed to by the socket.
+  
   virtual void qos (const ACE_QoS &ace_qos) = 0;
-  // Sets the QoS for this session object to ace_qos. Does not interfere with the 
-  // QoS in the underlying socket. This call is useful to update the QoS object 
-  // when the underlying socket QoS is being set through a mechanism other than
-  // the previous qos () method e.g. inside the dgram_mcast.subscribe () where the 
-  // QoS for the socket is set through ACE_OS::join_leaf ().
+  // Sets the QoS for this session object to ace_qos. Does not
+  // interfere with the QoS in the underlying socket. This call is
+  // useful to update the QoS object when the underlying socket QoS is
+  // being set through a mechanism other than the previous qos ()
+  // method e.g. inside the dgram_mcast.subscribe () where the QoS for
+  // the socket is set through ACE_OS::join_leaf ().
 
   virtual int update_qos (void) = 0;
-  // This is called from handle_qos () method of the the QoS Event Handler.
-  // Invoking this method is an indication of a QoS event occurring, that may have 
-  // resulted in a change of QoS for the underlying session. This method updates 
-  // the QoS object associated with this session.
+  // This is called from handle_qos () method of the the QoS Event
+  // Handler.  Invoking this method is an indication of a QoS event
+  // occurring, that may have resulted in a change of QoS for the
+  // underlying session. This method updates the QoS object associated
+  // with this session.
 
   virtual ACE_End_Point_Type flags (void) const = 0;
   virtual void flags (const ACE_End_Point_Type flags) = 0;
@@ -93,24 +96,37 @@ public:
   virtual void session_id (const int session_id) = 0;
   // Set the session id.
 
+  virtual ACE_HANDLE rsvp_events_handle (void) = 0;
+  // Get the file descriptor on which RSVP events will occur.
+
   virtual ACE_INET_Addr dest_addr (void) const = 0;
   // Get the destination address for this session.
 
   virtual void dest_addr (const ACE_INET_Addr &dest_addr) = 0;
   // Set the destination address for this session.
 
+  virtual u_short source_port (void) const = 0;
+  // Get the source port for this session.
+  
+  virtual void source_port (const u_short &source_port) = 0;
+  // Set the source port for this session.
+  
   virtual int version (void) = 0;
   // Returns the version of the underlying RSVP implementation. Is
-  // meaningful only when the underlying implementation has versioning.
+  // meaningful only when the underlying implementation has
+  // versioning.
    
 protected:
 
+  u_short source_port_;
+  // Source port if this is a Sender session. Used for rapi_sender ().
+
   int session_id_;
   // session id for the session.
   
   ACE_INET_Addr dest_addr_;
   // Destination address for this session.
-  
+
   ACE_Protocol_ID protocol_id_;
   // Is this a TCP or a UDP session.
 
diff --git a/ace/QoS_Session_Impl.cpp b/ace/QoS_Session_Impl.cpp
index 35c64294b0e..08e900cacb1 100644
--- a/ace/QoS_Session_Impl.cpp
+++ b/ace/QoS_Session_Impl.cpp
@@ -19,9 +19,9 @@ ACE_ALLOC_HOOK_DEFINE(ACE_QoS_Session_Impl)
 
 int ACE_RAPI_Session::rsvp_error = 0;
 
-// Call back function used by RAPI to report RSVP events. This function translates 
-// the RAPI QoS parameters into the more generic ACE_QoS parameters for the 
-// underlying RAPI session.
+// Call back function used by RAPI to report RSVP events. This
+// function translates the RAPI QoS parameters into the more generic
+// ACE_QoS parameters for the underlying RAPI session.
 int 
 rsvp_callback (rapi_sid_t sid,
                rapi_eventinfo_t eventype,
@@ -39,43 +39,49 @@ rsvp_callback (rapi_sid_t sid,
                void *args
                )
 {
-
-  ACE_QoS_Session * qos_session = (ACE_QoS_Session *) args;
+  
+  if (args == 0)
+    ACE_DEBUG ((LM_DEBUG,
+                "Argument in the call back function is null\n\n"));
+  
+  ACE_QoS_Session *qos_session = (ACE_QoS_Session *) args;
 
   // Extended Legacy format.
   qos_flowspecx_t *csxp = &flow_spec_list->specbody_qosx;
-  
-  ACE_Flow_Spec sending_fs (csxp->xspec_r,
-                            csxp->xspec_b,
-                            csxp->xspec_p,
-                            0,
-                            0,
-                            0,
-                            csxp->xspec_M,
-                            csxp->xspec_m,
-                            25,
-                            0);
 
+  ACE_QoS ace_qos = qos_session->qos ();
+  ACE_Flow_Spec receiving_flow = ace_qos.receiving_flowspec ();
+  
   switch(eventype)
     {
     case RAPI_PATH_EVENT:
-      ACE_DEBUG ((LM_DEBUG,
-                  "RSVP PATH Event received\n"));
-      
-      ACE_DEBUG ((LM_DEBUG,
-                  "No. of TSpecs received : %d\n",
-                  flow_spec_no));
-      
-      // Set the sending flowspec QoS of the given session. 
-      qos_session->qos ().sending_flowspec (sending_fs);
-      
-      break;
+      {
+        ACE_DEBUG ((LM_DEBUG,
+                    "RSVP PATH Event received\n"
+                    "No. of TSpecs received : %d\n",
+                    flow_spec_no));
+        
+        ACE_Flow_Spec sending_fs (csxp->xspec_r,
+                                  csxp->xspec_b,
+                                  csxp->xspec_p,
+                                  0,
+                                  0,
+                                  0,
+                                  csxp->xspec_M,
+                                  csxp->xspec_m,
+                                  25,
+                                  0);
+        
+        // Set the sending flowspec QoS of the given session. 
+        ace_qos.sending_flowspec (sending_fs);
+        
+      }
+    
+    break;
       
     case RAPI_RESV_EVENT:
       ACE_DEBUG ((LM_DEBUG,
-                  "RSVP RESV Event received\n"));
-      
-      ACE_DEBUG ((LM_DEBUG,
+                  "RSVP RESV Event received\n"
                   "No. of FlowSpecs received : %d\n",
                   flow_spec_no));
       
@@ -84,25 +90,25 @@ rsvp_callback (rapi_sid_t sid,
         {
         case QOS_GUARANTEEDX:
           // Slack term in MICROSECONDS
-          qos_session->qos ().receiving_flowspec ().delay_variation (csxp->xspec_S);
-          
+          receiving_flow.delay_variation (csxp->xspec_S);
+
           // @@How does the guaranteed rate parameter map to the ACE_Flow_Spec.
           // Note there is no break !!
           
         case QOS_CNTR_LOAD:
-
+          
           // qos_service_type.
-          qos_session->qos ().receiving_flowspec ().service_type (csxp->spec_type); 
+          receiving_flow.service_type (csxp->spec_type); 
           // Token Bucket Average Rate (B/s)
-          qos_session->qos ().receiving_flowspec ().token_rate (csxp->xspec_r);
+          receiving_flow.token_rate (csxp->xspec_r);
           // Token Bucket Rate (B)
-          qos_session->qos ().receiving_flowspec ().token_bucket_size (csxp->xspec_b);
+          receiving_flow.token_bucket_size (csxp->xspec_b);
           // Peak Data Rate (B/s)
-          qos_session->qos ().receiving_flowspec ().peak_bandwidth (csxp->xspec_p);
+          receiving_flow.peak_bandwidth (csxp->xspec_p);
           // Minimum Policed Unit (B)
-          qos_session->qos ().receiving_flowspec ().minimum_policed_size (csxp->xspec_m);
+          receiving_flow.minimum_policed_size (csxp->xspec_m);
           // Max Packet Size (B)
-          qos_session->qos ().receiving_flowspec ().max_sdu_size (csxp->xspec_M);
+          receiving_flow.max_sdu_size (csxp->xspec_M);
 
           break;
           
@@ -112,6 +118,8 @@ rsvp_callback (rapi_sid_t sid,
                             0);
         };
       
+      ace_qos.receiving_flowspec (receiving_flow);
+
       break;
 
     case RAPI_PATH_ERROR:
@@ -138,14 +146,23 @@ rsvp_callback (rapi_sid_t sid,
                   "RESV CONFIRM Event received\n"));
       break;
       
+    default:
+      ACE_DEBUG ((LM_DEBUG,
+                  "Unknown RSVP Event Received\n"));
+      break;
+      
     }
-  
+
+  // Set the updated ACE_QoS for the RSVP callback argument(QoS session).
+  qos_session->qos (ace_qos);
+        
 } 
 
 // Constructor.
 ACE_RAPI_Session::ACE_RAPI_Session (void)
 {
   ACE_TRACE ("ACE_RAPI_Session::ACE_RAPI_Session");
+  this->source_port (DEFAULT_SOURCE_SENDER_PORT);
 }
 
 // Open a RAPI QoS session [dest IP, dest port, Protocol ID].
@@ -156,8 +173,6 @@ ACE_RAPI_Session::open (ACE_INET_Addr dest_addr,
   this->dest_addr_ = dest_addr;
   this->protocol_id_ = protocol_id;
   
-  rapi_eventinfo_t RSVP_arg;      /*RSVP callback argument*/
-
   // Open a RAPI session. Note "this" is being passed as an argument to 
   // the callback function. The callback function uses this argument to 
   // update the QoS of this session based on the RSVP event it receives.
@@ -166,15 +181,14 @@ ACE_RAPI_Session::open (ACE_INET_Addr dest_addr,
                                         protocol_id,
                                         0, 
                                         rsvp_callback, 
-                                        //(void *) this, 
-                                        (void *) &RSVP_arg,
+                                        (void *) this, 
                                         &rsvp_error)) == NULL_SID)
     ACE_ERROR_RETURN ((LM_ERROR,
                        "rapi_session () call fails. Error\n"),
                       -1);
   else
     ACE_DEBUG ((LM_DEBUG,
-                "rapi_session () call succeeds\n"
+                "rapi_session () call succeeds. "
                 "Session ID = %d\n",
                 this->session_id_));
   
@@ -203,7 +217,6 @@ ACE_RAPI_Session::qos (ACE_SOCK *socket,
                        const ACE_QoS &ace_qos)
 {
   ACE_UNUSED_ARG (socket);
-  ACE_UNUSED_ARG (qos_manager);
   
   // If sender : call sending_qos ()
   // If receiver : call receiving_qos ()
@@ -251,7 +264,7 @@ ACE_RAPI_Session::sending_qos (const ACE_QoS &ace_qos)
               "\t Peak = %f\n"
               "\t MPU = %d\n"
               "\t MDU = %d\n"
-              "\t\t TTL = %d\n",
+              "\t TTL = %d\n",
               t_spec->tspecbody_qosx.spec_type,
               t_spec->tspecbody_qosx.xtspec_r,
               t_spec->tspecbody_qosx.xtspec_b,
@@ -260,8 +273,8 @@ ACE_RAPI_Session::sending_qos (const ACE_QoS &ace_qos)
               t_spec->tspecbody_qosx.xtspec_M,
               sending_flowspec.ttl ()));
   
-  // @@Hardcoded port. This should be changed later.
-  ACE_INET_Addr sender_addr (9090); 
+  // This the source sender port.
+  ACE_INET_Addr sender_addr (this->source_port ()); 
 
   ACE_DEBUG ((LM_DEBUG,
               "Making the rapi_sender () call\n"));
@@ -288,16 +301,16 @@ ACE_RAPI_Session::sending_qos (const ACE_QoS &ace_qos)
 int 
 ACE_RAPI_Session::receiving_qos (const ACE_QoS &ace_qos)
 {
-
+  
   ACE_Flow_Spec receiving_flowspec = ace_qos.receiving_flowspec ();
   rapi_flowspec_t *flow_spec = init_flowspec_simplified (receiving_flowspec);
-
+  
   if (flow_spec == 0)
     ACE_ERROR_RETURN ((LM_ERROR,
                        "Error in translating from ACE Flow Spec to"
                        " RAPI FlowSpec\n"),
                       -1);
-
+  
   char buffer[BUFSIZ];
 
   // This formats the flow_spec in a visually intuitive char * that can
@@ -306,7 +319,7 @@ ACE_RAPI_Session::receiving_qos (const ACE_QoS &ace_qos)
   ACE_DEBUG ((LM_DEBUG,
               "\nReceiver FlowSpec : %s\n",
               buffer));
-
+  
   // Print out all the fields separately.
   ACE_DEBUG ((LM_DEBUG,
               "\nFlowSpec :\n"
@@ -323,11 +336,6 @@ ACE_RAPI_Session::receiving_qos (const ACE_QoS &ace_qos)
               flow_spec->specbody_qosx.xspec_m,
               flow_spec->specbody_qosx.xspec_M));
 
-  // @@Hardcoded port. This should be changed later.
-  //  ACE_INET_Addr receiver_addr (8002); 
-
-  //  ACE_INET_Addr receiver_addr; 
-
   sockaddr_in Receiver_host;
 
   Receiver_host.sin_addr.s_addr = INADDR_ANY;
@@ -365,6 +373,7 @@ ACE_RAPI_Session::receiving_qos (const ACE_QoS &ace_qos)
 int 
 ACE_RAPI_Session::update_qos (void)
 {
+  // Update the session QoS Parameters based on the RSVP Event Received.
   if ((rsvp_error = rapi_dispatch ()) != 0)
     ACE_ERROR_RETURN ((LM_ERROR,
                        "Error in rapi_dispatch () : %s\n",
@@ -423,6 +432,7 @@ ACE_RAPI_Session::init_flowspec_simplified(const ACE_Flow_Spec &flow_spec)
   qos_flowspecx_t *csxp = &flowsp->specbody_qosx;
 
   // Choose based on the service type : [QOS_GUARANTEEDX/QOS_CNTR_LOAD].
+
   switch (flow_spec.service_type ()) 
     {
     case QOS_GUARANTEEDX:
@@ -434,14 +444,12 @@ ACE_RAPI_Session::init_flowspec_simplified(const ACE_Flow_Spec &flow_spec)
       // Note there is no break !!
 
     case QOS_CNTR_LOAD:
-      ACE_DEBUG ((LM_DEBUG,
-                  "QOS_CONTROLLED_LOAD\n"));
       csxp->spec_type = flow_spec.service_type ();        // qos_service_type
       csxp->xspec_r = flow_spec.token_rate ();            // Token Bucket Average Rate (B/s)
       csxp->xspec_b = flow_spec.token_bucket_size ();     // Token Bucket Rate (B)
       csxp->xspec_p = flow_spec.peak_bandwidth ();        // Peak Data Rate (B/s)
       csxp->xspec_m = flow_spec.minimum_policed_size ();  // Minimum Policed Unit (B)
-
+      
       // @@Hardcoded Max. Pkt. size.
       csxp->xspec_M = 65535;                              // Max Packet Size (B)
 
@@ -500,11 +508,6 @@ ACE_GQoS_Session::qos (ACE_SOCK *socket,
   // Confirm if the current session is one of the QoS sessions
   // subscribed to by the given socket.
 
-  //if (socket->qos_session_set ().find (this) == -1)
-
-  // @@Vishal : Need to relate the below to the socket (as above)
-  // instead of the QoS Manager.
-
   if (qos_manager->qos_session_set ().find (this) == -1)
     ACE_ERROR_RETURN ((LM_ERROR,
                        ACE_TEXT ("This QoS session was not subscribed to")
diff --git a/ace/QoS_Session_Impl.h b/ace/QoS_Session_Impl.h
index 698ae478f82..a657dbb837b 100644
--- a/ace/QoS_Session_Impl.h
+++ b/ace/QoS_Session_Impl.h
@@ -28,6 +28,8 @@
 #if defined (ACE_HAS_RAPI)
 #include "rapi_lib.h"
 
+#define DEFAULT_SOURCE_SENDER_PORT 10001
+
 class ACE_Export ACE_RAPI_Session : public ACE_QoS_Session
 {
   // = TITLE
@@ -83,12 +85,21 @@ public:
   virtual void session_id (const int session_id);
   // Set the RAPI session id.
 
+  virtual ACE_HANDLE rsvp_events_handle (void);
+  // Get the RAPI file descriptor for RSVP events.
+
   virtual ACE_INET_Addr dest_addr (void) const;
   // Get the destination address for this RAPI session.
 
   virtual void dest_addr (const ACE_INET_Addr &dest_addr);
   // Set the destination address for this RAPI session.
 
+  virtual u_short source_port (void) const;
+  // Get the source port for this session.
+  
+  virtual void source_port (const u_short &source_port);
+  // Set the source port for this session.
+  
   virtual int version ();
   // RAPI version. Returned value = 100 * major-version + minor-version.
 
@@ -175,12 +186,21 @@ public:
   virtual void dest_addr (const ACE_INET_Addr &dest_addr);
   // Set the destination address for this GQoS session.
 
+  virtual u_short source_port (void) const;
+  // Get the source port for this session.
+  
+  virtual void source_port (const u_short &source_port);
+  // Set the source port for this session.
+  
   virtual int session_id (void) const;
   // Get the GQoS session id.
   
   virtual void session_id (const int session_id);
   // Set the GQoS session id.
 
+  virtual ACE_HANDLE rsvp_events_handle (void);
+  // Get the file descriptor of the underlying socket.
+
   virtual int version ();
   // GQoS version.
 
diff --git a/ace/QoS_Session_Impl.i b/ace/QoS_Session_Impl.i
index 72021154d23..6f17f95d776 100644
--- a/ace/QoS_Session_Impl.i
+++ b/ace/QoS_Session_Impl.i
@@ -40,6 +40,22 @@ ACE_RAPI_Session::session_id (const int session_id)
   this->session_id_ = session_id;
 }
 
+// Get the RAPI file desciptor for RSVP events.
+ACE_INLINE ACE_HANDLE
+ACE_RAPI_Session::rsvp_events_handle (void)
+{
+  int rapi_fd = rapi_getfd (this->session_id ());
+  if (rapi_fd == -1)
+    {
+      this->close ();
+      ACE_ERROR_RETURN ((LM_ERROR,
+                         "Error in rapi_getfd ()\n"),
+                        -1);
+    }
+
+  return rapi_fd;
+}
+
 // Get the End Point Type (Sender/Receiver/Both).
 ACE_INLINE ACE_End_Point_Type 
 ACE_RAPI_Session::flags (void) const
@@ -69,6 +85,20 @@ ACE_RAPI_Session::dest_addr (const ACE_INET_Addr &dest_addr)
   this->dest_addr_ = dest_addr;
 }
 
+// Get the source port for this RAPI session.
+ACE_INLINE u_short 
+ACE_RAPI_Session::source_port (void) const
+{
+  return this->source_port_;
+}
+
+// Set the source port for this RAPI session.
+ACE_INLINE void 
+ACE_RAPI_Session::source_port (const u_short &source_port)
+{
+  this->source_port_ = source_port;
+}
+
 // RAPI version. Returned value = 100 * major-version + minor-version.
 ACE_INLINE int 
 ACE_RAPI_Session::version (void)
@@ -113,6 +143,15 @@ ACE_GQoS_Session::session_id (const int session_id)
   this->session_id_ = session_id;
 }
 
+// Get the underlying file desciptor for RSVP events.
+// Currently returns 0 because GQoS does not have a special
+// descriptor for QoS events.
+ACE_INLINE ACE_HANDLE
+ACE_GQoS_Session::rsvp_events_handle (void)
+{
+  return 0;
+}
+
 // Get the End Point Type (Sender/Receiver/Both).
 ACE_INLINE ACE_QoS_Session::ACE_End_Point_Type 
 ACE_GQoS_Session::flags (void) const
@@ -141,6 +180,20 @@ ACE_GQoS_Session::dest_addr (const ACE_INET_Addr &dest_addr)
   this->dest_addr_ = dest_addr;
 }
 
+// Get the source port for this RAPI session.
+ACE_INLINE u_short 
+ACE_GQoS_Session::source_port (void) const
+{
+  return this->source_port_;
+}
+
+// Set the source port for this RAPI session.
+ACE_INLINE void 
+ACE_GQoS_Session::source_port (const u_short &source_port)
+{
+  this->source_port_ = source_port;
+}
+
 // GQoS version.
 ACE_INLINE int 
 ACE_GQoS_Session::version (void)
diff --git a/ace/SOCK_Dgram_Mcast_QoS.cpp b/ace/SOCK_Dgram_Mcast_QoS.cpp
index 412ae9ec6d0..36baf0d9e8f 100644
--- a/ace/SOCK_Dgram_Mcast_QoS.cpp
+++ b/ace/SOCK_Dgram_Mcast_QoS.cpp
@@ -56,7 +56,7 @@ ACE_SOCK_Dgram_Mcast_QoS::open (const ACE_Addr &mcast_addr,
                           flags,
                           reuse_addr) == -1)
         return -1;
-
+      
       int one = 1;
       if (reuse_addr
           && this->ACE_SOCK::set_option (SOL_SOCKET,
@@ -199,7 +199,6 @@ ACE_SOCK_Dgram_Mcast_QoS::subscribe (const ACE_INET_Addr &mcast_addr,
                                      ACE_Protocol_Info *protocolinfo,
                                      ACE_SOCK_GROUP g,
                                      u_long flags,
-                                     ACE_QoS_Manager *qos_manager,
                                      ACE_QoS_Session *qos_session)
 {
   ACE_TRACE ("ACE_SOCK_Dgram_Mcast::subscribe");
@@ -234,26 +233,25 @@ ACE_SOCK_Dgram_Mcast_QoS::subscribe (const ACE_INET_Addr &mcast_addr,
       // same as the QoS session address.
       if (mcast_addr == qos_session->dest_addr ()) 
       {
-
         // Subscribe to the QoS session.
-        ACE_UNUSED_ARG (qos_manager);
-#if 0
-        if (qos_manager->join_qos_session (qos_session) == -1)
+        if (this->qos_manager_.join_qos_session (qos_session) == -1)
           ACE_ERROR_RETURN ((LM_ERROR,
                              "Unable to join QoS Session\n"),
                             -1);
-#endif
-
       } 
       else
-      {
+        {
+          if (this->close () != 0)
+            ACE_ERROR ((LM_ERROR,
+                        "Unable to close socket\n"));
+          
           ACE_ERROR_RETURN ((LM_ERROR,
                              ACE_TEXT ("Dest Addr in the QoS Session does")
                              ACE_TEXT (" not match the address passed into")
                              ACE_TEXT (" subscribe\n")),
                             -1);
-      }
-
+        }
+      
       sockaddr_in mult_addr;
 
       if (protocol_family == ACE_FROM_PROTOCOL_INFO)
diff --git a/ace/SOCK_Dgram_Mcast_QoS.h b/ace/SOCK_Dgram_Mcast_QoS.h
index 9ed2d63ea87..81d32e11b4c 100644
--- a/ace/SOCK_Dgram_Mcast_QoS.h
+++ b/ace/SOCK_Dgram_Mcast_QoS.h
@@ -19,6 +19,7 @@
 #include "ace/pre.h"
 
 #include "ace/SOCK_Dgram_Mcast.h"
+#include "ace/QoS_Manager.h"
 
 #if !defined (ACE_LACKS_PRAGMA_ONCE)
 # pragma once
@@ -31,7 +32,7 @@ class ACE_Export ACE_SOCK_Dgram_Mcast_QoS : public ACE_SOCK_Dgram_Mcast
   //     wrapper for UDP/IP multicast.
 
 public:
-  // = Initialization routine.
+  // = Initialization routines.
   ACE_SOCK_Dgram_Mcast_QoS (void);
 
   // Note that there is no public <open> method.  Therefore, this
@@ -52,7 +53,6 @@ public:
                  ACE_Protocol_Info *protocolinfo = 0,
                  ACE_SOCK_GROUP g = 0,
                  u_long flags = 0,
-                 ACE_QoS_Manager *qos_manager = 0,
                  ACE_QoS_Session *qos_session = 0);
   // This is a QoS-enabled method for joining a multicast group, which
   // passes <qos_params> via <ACE_OS::join_leaf>.  The network
@@ -94,6 +94,9 @@ public:
                 ACE_OVERLAPPED_COMPLETION_FUNC func) const;
   // Send an <n> byte <buf> to the datagram socket (uses <WSASentTo>).
 
+  ACE_QoS_Manager qos_manager (void);
+  // Returns the QoS manager for this socket.
+
   ACE_ALLOC_HOOK_DECLARE;
   // Declare the dynamic allocation hooks.
 
@@ -119,6 +122,9 @@ private:
                      ACE_Protocol_Info *protocolinfo);
   // Subscribe to the multicast interface using QoS-enabled semantics.
 
+  ACE_QoS_Manager qos_manager_;
+  // Manages the QoS sessions that this socket subscribes to.
+
 };
 
 #if !defined (ACE_LACKS_INLINE_FUNCTIONS)
diff --git a/ace/SOCK_Dgram_Mcast_QoS.i b/ace/SOCK_Dgram_Mcast_QoS.i
index 879f50d51fd..a1023979f8b 100644
--- a/ace/SOCK_Dgram_Mcast_QoS.i
+++ b/ace/SOCK_Dgram_Mcast_QoS.i
@@ -45,3 +45,8 @@ ACE_SOCK_Dgram_Mcast_QoS::send (const void *buf,
                                func);
 }
 
+ASYS_INLINE ACE_QoS_Manager
+ACE_SOCK_Dgram_Mcast_QoS::qos_manager (void)
+{
+  return this->qos_manager_;
+}