Implementing Pluggable Protocol in TAO AV Streaming Service

Nagarajan Surendran and Yamuna Krishnamurthy

Center for Distributed Object Computing
Washington University in St.Louis

Overview

Traditionally DOC Middleware has provided facilities to design application-level objects at a higher level of abstraction, away from low-level details of network programming and distribution. The aspect of multimedia application that could benefit from this is an interoperable standard design of multimedia interfaces defined in IDL. The OMG A/V Specification defines such a set of interfaces which helps in type-safe ,robust and flexible applications. Also as proved in our earlier work,ORBs with IIOP implementation cannot be used for streaming multimedia data since they're optimized for request-response semantics. To circumvent this the application developer will have to use Out of Band data connection to stream the data while using the ORB to negotiate QoS, bind type-safe devices and explore their properties.

Also it is important for a Multimedia framework to have the ability to support different transports and be able to plug and configure them dynamically and statically .For example, a wireless link based application may want to use the Wireless link protocol to stream the data. RTP is gaining wide significance as the Transport protocol for streaming Audio and Video Data over the Internet. Our framework provides an implementation of RTP/RTCP for the audio/video profile. RTP is internet-centric and doesn't provide facilities to transport  IDL-defined flows. SFP is a Simple Flow Protocol specified by the OMG for A/V Streaming and it can be mapped to different transports like UDP, RTP and ATM AAL5. Our framework provides an implementation of SFP that runs over UDP and multicast UDP.

The OMG specification defines the flow specification syntax that is to be used for the bind_devs calls for connection establishment. It defines the protocol names and also the syntax for specifying the transport/flow protocol information but it doesn't define any interfaces for the protocol implementation as such. Our framework bridges this gap by defining a uniform API  for the different flow protocols like RTP and SFP while taking care of the variations using a Policy interface. We solve this with our Pluggable Protocol Framework using design patterns like Layer,
Acceptor, Connector, Facade and Abstract Factory. Please look at this paper for more documentation on the TAO AV Service Pluggable Protocol Framework.

Implementing the Pluggable Protocol

In order to add a new pluggable transport and flow protocol to the AVStreams framework, without making changes to the framework itself, the pluggable protocol must implement the following components of the Pluggable Protocol framework:

The Transport Protocol Components

Acceptor and Connector: 

These are implementations of the Acceptor and Connector design patterns. In the case of connectionless protocols like UDP, calling accept and connect will result in the creation of handlers immediately.

The pluggable transport protocol implementation should inherit from the following abstract classes declared in <orbsvcs/orbsvcs/AV/Transport.h>:

/**
 * @class TAO_AV_Acceptor
 * @brief
 */
class TAO_AV_Export TAO_AV_Acceptor
{
public:
  TAO_AV_Acceptor (void);
  virtual ~TAO_AV_Acceptor (void);
  virtual int open (TAO_Base_StreamEndPoint *endpoint,
                              TAO_AV_Core *av_core,
                              TAO_FlowSpec_Entry *entry,
                              TAO_AV_Flow_Protocol_Factory *factory) = 0;

  virtual int open_default (TAO_Base_StreamEndPoint *endpoint,
                             TAO_AV_Core *av_core,
                             TAO_FlowSpec_Entry *entry,
                             TAO_AV_Flow_Protocol_Factory *factory) = 0;

  const char *flowname ();
  virtual int close (void) = 0;
protected:
  ACE_CString flowname_;
  TAO_AV_Core *av_core_;
  ACE_Addr *address_;
};
 

/**
 * @class TAO_AV_Connector
 * @brief
 */
class TAO_AV_Export TAO_AV_Connector
{
public:
  TAO_AV_Connector (void);
  virtual ~TAO_AV_Connector (void);
  const char *flowname (void);

  virtual int open (TAO_Base_StreamEndPoint *endpoint,
                    TAO_AV_Core *av_core,
                    TAO_AV_Flow_Protocol_Factory *factory) = 0;

  virtual int connect (TAO_FlowSpec_Entry *entry,
                       TAO_AV_Transport *&transport) = 0;

  virtual int close (void) = 0;
protected:
  ACE_CString flowname_;
};
 
 

Transport and Transport Factory:

The transport class provides a common interface for different protocols for performing operations like open, close, send and recv. The trasnport factory is an interface for creating acceptors and connectors.

The pluggable transport protocol implementation should inherit from the following abstract classes declared in  <orbsvcs/orbsvcs/AV/Transport.h>:

/**
 * @class TAO_AV_Transport
 * @brief A Base class for the different transport protocols.
 *        All the different transports should derive and implement
 *        the open,close,send and recv methods.
 */
class TAO_AV_Export TAO_AV_Transport
{
public:
  TAO_AV_Transport (void);

  virtual ~TAO_AV_Transport (void);

  virtual int open (ACE_Addr *address) = 0;

  virtual int close (void) = 0;

  virtual int mtu (void) = 0;
  virtual ACE_Addr *get_peer_addr (void) = 0;
  virtual ACE_Addr *get_local_addr (void);

  virtual ssize_t send (const ACE_Message_Block *mblk,
                        ACE_Time_Value *s = 0) = 0;

  /// Write the contents of the buffer of length len to the connection.
  virtual ssize_t send (const char *buf,
                        size_t len,
                        ACE_Time_Value *s = 0) = 0;

  /// Write the contents of iovcnt iovec's to the connection.
  virtual ssize_t send (const iovec *iov,
                        int iovcnt,
                        ACE_Time_Value *s = 0) = 0;

  /// Read len bytes from into buf.
  virtual ssize_t recv (char *buf,
                        size_t len,
                        ACE_Time_Value *s = 0) = 0;

  /// Read len bytes from into buf using flags.
  virtual ssize_t recv (char *buf,
                        size_t len,
                        int flags,
                        ACE_Time_Value *s = 0) = 0;

  ///  Read received data into the iovec buffers.
  virtual ssize_t recv (iovec *iov,
                        int iovcnt,
                        ACE_Time_Value *s = 0) = 0;

};
 

/**
 * @class TAO_AV_Transport_Factory
 * @brief
 */
class TAO_AV_Export TAO_AV_Transport_Factory : public ACE_Service_Object
{
public:
  /// Initialization hook.
  TAO_AV_Transport_Factory (void);
  virtual ~TAO_AV_Transport_Factory (void);
  virtual int init (int argc, char *argv[]);
  virtual int match_protocol (const char *protocol_string);
  virtual TAO_AV_Acceptor *make_acceptor (void);
  virtual TAO_AV_Connector *make_connector (void);
};
 

Flow handler :

All the transport handlers derive from the  TAO_AV_Flow_Handler which has methods to start and stop and provide flow specific functionality for timeout upcalls to the Callbacks.

The pluggable transport protocol implementation should inherit from the following abstract class declared in <orbsvcs/orbsvcs/AV/Transport.h>:

/**
 * @class TAO_AV_Flow_Handler
 * @brief
 */
class TAO_AV_Export TAO_AV_Flow_Handler
{
public:
  /// Constructor.
  TAO_AV_Flow_Handler (void);

  /// Start/stop the flow handler.
  virtual int start (TAO_FlowSpec_Entry::Role role);
  virtual int stop  (TAO_FlowSpec_Entry::Role role);

  /// Schedule timer. Uses the get_timeout method on the callback.
  virtual int schedule_timer (void);

  /// get the transport.
  TAO_AV_Transport *transport (void);

  /// set/get protocol_object.
  TAO_AV_Protocol_Object* protocol_object (void);
  void protocol_object (TAO_AV_Protocol_Object *protocol_object);

  /// set the callback.
  void callback (TAO_AV_Callback *callback);

  /// Handle timeout. called from reactor.
  int handle_timeout (const ACE_Time_Value &tv, const void *arg = 0);

  /// set the remote address.
  virtual int set_remote_address (ACE_Addr *address);

  /// get the underlying event handler. To be overridden by the derived clases.
  virtual ACE_Event_Handler* event_handler (void) = 0;

  virtual int change_qos (AVStreams::QoS);

protected:
  TAO_AV_Transport *transport_;
  TAO_AV_Callback *callback_;
  TAO_AV_Protocol_Object *protocol_object_;
  long timer_id_;
  ACE_Reactor *reactor_;
  void *timeout_arg_;
};
 

The Flow Protocol Components

Flow Protocol Factory:

The  flow protocol factory is an interface for creating flow protocol objects.

The pluggable flow protocol implementation should inherit from the following abstract class declared in <orbsvcs/orbsvcs/AV/Protocol_Factory>:

/**
 * @class TAO_AV_Flow_Protocol_Factory
 * @brief
 */
class TAO_AV_Export TAO_AV_Flow_Protocol_Factory : public ACE_Service_Object
{
public:
  /// Initialization hook.
  TAO_AV_Flow_Protocol_Factory (void);
  virtual ~TAO_AV_Flow_Protocol_Factory (void);
  virtual int init (int argc, char *argv[]);
  virtual int match_protocol (const char *flow_string);
  virtual TAO_AV_Protocol_Object* make_protocol_object (TAO_FlowSpec_Entry *entry,
                                                        TAO_Base_StreamEndPoint *endpoint,
                                                        TAO_AV_Flow_Handler *handler,
                                                        TAO_AV_Transport *transport);
  virtual const char *control_flow_factory (void);
};
 

Protocol Object:

This is a concrete implementation providing the flow protocol functionality. Applications use this interface to send frames and the Protocol_Object uses application specified  Callback objects to deliver frames.

The pluggable flow protocol implementation should inherit from the following abstract class declared in <orbsvcs/orbsvcs/AV/Protocol_Factory>:

/**
 * @class TAO_AV_Protocol_Object
 * @brief
 */
class TAO_AV_Export TAO_AV_Protocol_Object
{
public:
  TAO_AV_Protocol_Object (void);

  /// constructor.
  TAO_AV_Protocol_Object (TAO_AV_Callback *callback,
                          TAO_AV_Transport *transport);

  /// Destructor
  virtual ~TAO_AV_Protocol_Object (void);

  virtual int open (TAO_AV_Callback *callback,
                    TAO_AV_Transport *transport);

  virtual int handle_input (void) = 0;

  /// Called on a control object.
  virtual int handle_control_input (ACE_Message_Block *control_frame,
                                    const ACE_Addr &peer_address);

  /// set/get policies.
  virtual int set_policies (const TAO_AV_PolicyList &policy_list);
  virtual TAO_AV_PolicyList get_policies (void);

  /// start/stop the flow.
  virtual int start (void);
  virtual int stop (void);

  /// send a data frame.
  virtual int send_frame (ACE_Message_Block *frame,
                          TAO_AV_frame_info *frame_info = 0) = 0;

  /// send a frame in iovecs.
  virtual int send_frame (const iovec *iov,
                          int iovcnt,
                          TAO_AV_frame_info *frame_info = 0) = 0;

  virtual int send_frame (const char *buf,
                          size_t len) = 0;

  /// end the stream.
  virtual void control_object (TAO_AV_Protocol_Object *object);
  virtual int destroy (void) = 0;
  TAO_AV_Transport *transport (void);
protected:
  TAO_AV_Transport *transport_;
  TAO_AV_PolicyList policy_list_;
  TAO_AV_Callback *callback_;
};

For an example of how all the above components are implemented refer to the implementation of the UDP transport <orbsvcs/orbsvcs/AV/UDP.h>  and <orbsvcs/orbsvcs/AV/UDP.cpp>

Using a Pluggable Protocol

Once the TAO AV pluggable transport and flow protocols are implemented they can be loaded into the AV Streams framework by adding their entries to the Service Configurator file (eg. svc.conf) for that protocol:

A typical svc.conf will look like this:

dynamic TAO_AV_Resource_Factory Service_Object * TAO_AV:_make_TAO_AV_Resource_Factory() ""
dynamic ATM_Factory Service_Object * TAO_AV:_make_TAO_AV_ATM_Factory() ""
static TAO_AV_Resource_Factory "-AVTransportFactory ATM_Factory"
dynamic SFP_Factory Service_Object * TAO_AV:_make_TAO_AV_SFP_Factory() ""
static TAO_AV_Resource_Factory "-AVFlowProtocolFactory SFP_Factory"

The TAO_AV_Resource_Factory is a the default resource factory that helps to load the different transport and flow protocols to the corresponding factory sets.

NOTE: The TAO_AV_Resource_Factory must be loaded in order to load the pluggable protocols, ie. the first statement shown above is mandatory to load the protocols.

The above entries show how to add a transport factory eg. ATM Factory and a flow protocol factory eg. SFP Factory. The -AVTransportFactory option causes the specified transport protocol factory to be loaded into the AV Core and the -AVFlowProtocolFactory option causes the specified flow protocol factory to be loaded into the AV Core.