Added all the new files to test the Multipoint binding in AVStreams.

Two instances of server receive the same file from the ftp program and
save them under 2 different filenames using the multipoint bindings of
TAO_StreamCtrl.

5 files changed, 859 insertions, 0 deletions

diff --git a/TAO/orbsvcs/tests/AVStreams/Multicast/Makefile b/TAO/orbsvcs/tests/AVStreams/Multicast/Makefile
new file mode 100644
index 00000000000..ad90e057c35
--- /dev/null
+++ b/TAO/orbsvcs/tests/AVStreams/Multicast/Makefile
@@ -0,0 +1,57 @@
+#----------------------------------------------------------------------------
+#	$Id$
+#
+#	Top-level Makefile for Multipoint Binding testing of AVStreams of TAO.
+#
+#----------------------------------------------------------------------------
+
+#----------------------------------------------------------------------------
+#	Local macros
+#----------------------------------------------------------------------------
+
+LIBS     += -lorbsvcs -lTAO 
+
+SERVER_OBJS = server.o
+CLIENT_OBJS = ftp.o
+
+VLDLIBS = $(LDLIBS:%=%$(VAR))
+
+ifndef TAO_ROOT
+TAO_ROOT = $(ACE_ROOT)/TAO
+endif
+
+FILES = ftp server
+DEFS    = $(addsuffix .h,$(FILES))
+LSRC	= $(addsuffix .cpp,$(FILES))
+
+BIN2 = server ftp
+
+#### If the TAO orbsvcs library wasn't built with sufficient components,
+#### don't try to build here.
+TAO_ORBSVCS := $(shell $(ACE_ROOT)/bin/ace_components --orbsvcs)
+ifeq (AV,$(findstring AV,$(TAO_ORBSVCS)))
+  BIN = $(BIN2)
+endif # AV
+
+#----------------------------------------------------------------------------
+#	Include macros and targets
+#----------------------------------------------------------------------------
+
+include $(ACE_ROOT)/include/makeinclude/wrapper_macros.GNU
+include $(ACE_ROOT)/include/makeinclude/macros.GNU
+include $(TAO_ROOT)/rules.tao.GNU
+include $(ACE_ROOT)/include/makeinclude/rules.common.GNU
+include $(ACE_ROOT)/include/makeinclude/rules.nonested.GNU
+include $(ACE_ROOT)/include/makeinclude/rules.bin.GNU
+include $(ACE_ROOT)/include/makeinclude/rules.local.GNU
+
+DCFLAGS = -g
+LDFLAGS += -L$(TAO_ROOT)/orbsvcs/orbsvcs -L$(TAO_ROOT)/tao
+CPPFLAGS += -I$(TAO_ROOT)/orbsvcs -I$(TAO_ROOT) $(TSS_ORB_FLAG)
+
+# DO NOT DELETE THIS LINE -- g++dep uses it.
+# DO NOT PUT ANYTHING AFTER THIS LINE, IT WILL GO AWAY.
+
+
+
+# IF YOU PUT ANYTHING HERE IT WILL GO AWAY
diff --git a/TAO/orbsvcs/tests/AVStreams/Multicast/ftp.cpp b/TAO/orbsvcs/tests/AVStreams/Multicast/ftp.cpp
new file mode 100644
index 00000000000..b8578ee8997
--- /dev/null
+++ b/TAO/orbsvcs/tests/AVStreams/Multicast/ftp.cpp
@@ -0,0 +1,418 @@
+// $Id$
+
+#include "ftp.h"
+
+FTP_Client_Callback::FTP_Client_Callback (FTP_Client_Flow_Handler *handler)
+  :handler_ (handler)
+{
+}
+
+int
+FTP_Client_Callback::handle_start (void)
+{
+  return this->handler_->start ();
+}
+
+int
+FTP_Client_Callback::handle_stop (void)
+{
+  return this->handler_->stop ();
+}
+
+FTP_Client_StreamEndPoint::FTP_Client_StreamEndPoint (TAO_ORB_Manager *orb_manager)
+  :orb_manager_ (orb_manager)
+{
+
+}
+
+int
+FTP_Client_StreamEndPoint::get_callback (const char *flowname,
+                                         TAO_AV_Callback *&callback)
+{
+  ACE_Time_Value timeout (2);
+  ACE_NEW_RETURN (this->handler_,
+                  FTP_Client_Flow_Handler (this->orb_manager_,
+                                           timeout),
+                  -1);
+  ACE_NEW_RETURN (this->callback_,
+                  FTP_Client_Callback (this->handler_),
+                  -1);
+  callback = this->callback_;
+  return 0;
+}
+
+int
+FTP_Client_StreamEndPoint::set_protocol_object (const char *flowname,
+                                                TAO_AV_Protocol_Object *object)
+{
+  int result = this->handler_->set_protocol_object (object);
+  ACE_CString flow_string (flowname);
+  if (flow_string.find ("RTP") != flow_string.npos)
+    {
+      // Set the policies.
+      TAO_AV_Policy_Manager policy_manager;
+      CORBA::ULong ssrc = 25;
+      int payload_type = 1;
+      PolicyList list (2);
+      list.length (2);
+      list [0] = policy_manager.create_policy (TAO_AV_Policy::TAO_AV_SSRC_POLICY,
+                                               &ssrc);
+      list [1] = policy_manager.create_policy (TAO_AV_Policy::TAO_AV_PAYLOAD_TYPE_POLICY,
+                                               &payload_type);
+      object->set_policies (list);
+    }
+  return 0;
+}
+
+FTP_Client_Flow_Handler::FTP_Client_Flow_Handler (TAO_ORB_Manager *orb_manager,
+                                                  ACE_Time_Value &timeout)
+  :orb_manager_ (orb_manager),
+   count_ (0),
+   protocol_object_ (0),
+   timeout_ (timeout)
+{
+}
+
+int
+FTP_Client_Flow_Handler::start (void)
+{
+  ACE_DEBUG ((LM_DEBUG,"FTP_Client_Flow_Handler::start"));
+  ACE_Time_Value delta = ACE_Time_Value::zero;
+  this->timer_id_ =  
+    TAO_AV_CORE::instance ()->reactor ()->schedule_timer (this,
+                                                          0,
+                                                          delta,
+                                                          this->timeout_);
+  return 0;
+}
+
+int
+FTP_Client_Flow_Handler::stop (void)
+{
+  ACE_DEBUG ((LM_DEBUG,"FTP_Client_Flow_Handler::stop"));
+  int result = TAO_AV_CORE::instance ()->reactor ()->cancel_timer (this->timer_id_);
+  if (result < 0)
+    ACE_ERROR_RETURN ((LM_ERROR,"FTP_Client_Flow_Handler::stop cancel timer failed\n"),-1);
+  return 0;
+}
+
+int
+FTP_Client_Flow_Handler::set_protocol_object (TAO_AV_Protocol_Object *object)
+{
+  this->protocol_object_ = object;
+  return 0;
+}
+int
+FTP_Client_Flow_Handler::handle_timeout (const ACE_Time_Value &tv,
+                                         const void *arg)
+{
+  ACE_DEBUG ((LM_DEBUG,"FTP_Client_StreamEndPoint::handle_timeout"));
+  ACE_Message_Block mb (BUFSIZ);
+  char *buf = mb.rd_ptr ();
+  cerr << "message block size" << mb.size () << endl;
+  int n = ACE_OS::fread(buf,mb.size (),1,CLIENT::instance ()->file ());
+  if (n < 0)
+    {
+      TAO_AV_CORE::instance ()->reactor ()->cancel_timer (this->timer_id_);
+      ACE_ERROR_RETURN ((LM_ERROR,"FTP_Client_Flow_Handler::fread end of file\n"),-1);
+    }
+  if (n == 0)
+    {
+      if (::feof (CLIENT::instance ()->file ()))
+        {
+          // wait for sometime for the data to be flushed to the other side.
+          this->count_++;
+          if (this->count_ == 10)
+            {
+              ACE_DEBUG ((LM_DEBUG,"handle_timeout:End of file\n"));
+              AVStreams::flowSpec stop_spec (1);
+              stop_spec.length (1);
+              ACE_DECLARE_NEW_CORBA_ENV;
+              stop_spec [0] = CORBA::string_dup (CLIENT::instance ()->flowname ());
+              CLIENT::instance ()->streamctrl ()->stop (stop_spec,ACE_TRY_ENV);
+              ACE_CHECK_RETURN (-1);
+            }
+          else
+            return 0;
+        }
+      else
+        ACE_ERROR_RETURN ((LM_ERROR,"FTP_Client_Flow_Handler::fread error\n"),-1);
+    }
+  cerr << "read bytes = " << n << endl;
+  mb.wr_ptr (n*BUFSIZ);
+  int result = this->protocol_object_->send_frame (&mb);
+  if (result < 0)
+    ACE_ERROR_RETURN ((LM_ERROR,"send failed:%p","FTP_Client_Flow_Handler::send \n"),-1);
+  ACE_DEBUG ((LM_DEBUG,"handle_timeout::buffer sent succesfully\n"));
+}
+
+
+Endpoint_Reactive_Strategy::Endpoint_Reactive_Strategy (TAO_ORB_Manager *orb_manager,
+                                                        Client *client)
+  :ENDPOINT_STRATEGY (orb_manager),
+   client_ (client),
+   orb_manager_ (orb_manager)
+{
+}
+
+int
+Endpoint_Reactive_Strategy::make_stream_endpoint (FTP_Client_StreamEndPoint *&endpoint)
+{
+  ACE_DEBUG ((LM_DEBUG,"Endpoint_Reactive_Strategy::make_stream_endpoint"));
+  ACE_NEW_RETURN (endpoint,
+                  FTP_Client_StreamEndPoint (this->orb_manager_),
+                  -1);
+  return 0;
+}
+
+Client::parse_args (int argc,
+                    char **argv)
+{
+  ACE_Get_Opt opts (argc,argv,"f:a:p:s");
+  
+  this->use_sfp_ = 0;
+  char c;
+  while ((c= opts ()) != -1)
+    {
+      switch (c)
+        {
+        case 'f':
+          this->filename_ = ACE_OS::strdup (opts.optarg);
+          break;
+        case 'a':
+          this->address_ = ACE_OS::strdup (opts.optarg);
+          break;
+        case 'p':
+          this->protocol_ = ACE_OS::strdup (opts.optarg);
+          break;
+        case 's':
+          this->use_sfp_ = 1;
+          break;
+        default:
+          ACE_DEBUG ((LM_DEBUG,"Unknown option\n"));
+          return -1;
+          break;
+        }
+    }
+  return 0;
+}
+
+FILE *
+Client::file (void)
+{
+  return this->fp_;
+}
+
+char*
+Client::flowname (void)
+{
+  return this->flowname_;
+}
+
+TAO_StreamCtrl*
+Client::streamctrl (void)
+{
+  return &this->streamctrl_;
+}
+
+Client::Client (void)
+  :orb_manager_ (TAO_AV_CORE::instance ()->orb_manager ()),
+   endpoint_strategy_ (orb_manager_,this),
+   client_mmdevice_ (&endpoint_strategy_),
+   fp_ (0),
+   protocol_ (ACE_OS::strdup ("UDP"))
+{
+}
+
+
+int
+Client::bind_to_server (const char *name)
+{
+  ACE_DECLARE_NEW_CORBA_ENV;
+
+  ACE_TRY
+    {
+      // Initialize the naming services
+      CosNaming::Name server_mmdevice_name (1);
+      server_mmdevice_name.length (1);
+      server_mmdevice_name [0].id = name;
+      CORBA::Object_var server_mmdevice_obj =
+        my_naming_client_->resolve (server_mmdevice_name,
+                                    ACE_TRY_ENV);
+      ACE_TRY_CHECK;
+
+      this->server_mmdevice_ =
+        AVStreams::MMDevice::_narrow (server_mmdevice_obj.in (),
+                                      ACE_TRY_ENV);
+      ACE_TRY_CHECK;
+
+      if (CORBA::is_nil (this->server_mmdevice_.in ()))
+        ACE_ERROR_RETURN ((LM_ERROR,
+                           " could not resolve Server_Mmdevice in Naming service <%s>\n"),
+                          -1);
+    }
+  ACE_CATCHANY
+    {
+      ACE_PRINT_EXCEPTION (ACE_ANY_EXCEPTION,"Command_Handler::resolve_reference");
+      return -1;
+    }
+  ACE_ENDTRY;
+  ACE_CHECK_RETURN (-1);
+  return 0;
+}
+
+int
+Client::init (int argc,char **argv)
+{
+  this->argc_ = argc;
+  this->argv_ = argv;
+
+  // Increase the debug_level so that we can see the output
+  //  TAO_debug_level++;
+  CORBA::String_var ior;
+  ACE_DECLARE_NEW_CORBA_ENV;
+  ACE_TRY
+    {
+      TAO_AV_CORE::instance ()->init (argc,
+                                      argv,
+                                      ACE_TRY_ENV);
+      ACE_TRY_CHECK;
+      this->orb_manager_ = TAO_AV_CORE::instance ()->orb_manager ();
+      this->orb_manager_->init_child_poa (this->argc_,
+                                         this->argv_,
+                                         "child_poa",
+                                         ACE_TRY_ENV);
+      ACE_TRY_CHECK;
+      this->parse_args (this->argc_, this->argv_);
+      // activate the client video mmdevice under the child poa.
+      ior = this->orb_manager_->activate (&this->client_mmdevice_,
+                                         ACE_TRY_ENV);
+      ACE_TRY_CHECK;
+
+      // Initialize the naming services
+      CORBA::ORB_var orb = orb_manager_->orb ();
+      if (this->my_naming_client_.init (orb.in ()) != 0)
+        ACE_ERROR_RETURN ((LM_ERROR,
+                           " (%P|%t) Unable to initialize "
+                           "the TAO_Naming_Client. \n"),
+                          -1);
+
+      this->fp_ = ACE_OS::fopen (this->filename_,"r");
+      if (this->fp_ != 0)
+        {
+          ACE_DEBUG ((LM_DEBUG,"file opened successfully\n"));
+        }
+    }
+  ACE_CATCHANY
+    {
+      ACE_PRINT_EXCEPTION (ACE_ANY_EXCEPTION,"Client::init");
+      return -1;
+    }
+  ACE_ENDTRY;
+  ACE_CHECK_RETURN (-1);
+  return 0;
+}
+
+int
+Client::run (void)
+{
+  ACE_DECLARE_NEW_CORBA_ENV;
+  ACE_TRY
+    {
+      char flow_protocol_str [BUFSIZ];
+      if (this->use_sfp_)
+        ACE_OS::strcpy (flow_protocol_str,"sfp:1.0");
+      else
+        ACE_OS::strcpy (flow_protocol_str,"");
+      AVStreams::streamQoS_var the_qos (new AVStreams::streamQoS);
+      AVStreams::flowSpec flow_spec (1);
+      // Bind the client and server mmdevices.
+
+      this->orb_manager_->activate_poa_manager (ACE_TRY_ENV);
+      ACE_TRY_CHECK;
+      ACE_INET_Addr addr (this->address_);
+      ACE_NEW_RETURN (this->flowname_,
+                      char [BUFSIZ],
+                      0);
+      ACE_OS::sprintf (this->flowname_,
+                       "Data_%s",
+                       this->protocol_);
+      TAO_Forward_FlowSpec_Entry entry (this->flowname_,
+                                        "IN",
+                                        "USER_DEFINED",
+                                        flow_protocol_str,
+                                        this->protocol_,
+                                        &addr);
+      flow_spec [0] = entry.entry_to_string ();
+      flow_spec.length (1);
+      CORBA::Boolean result = 
+        this->streamctrl_.bind_devs (this->client_mmdevice_._this (ACE_TRY_ENV),
+                                     AVStreams::MMDevice::_nil (),
+                                     the_qos.inout (),
+                                     flow_spec,
+                                     ACE_TRY_ENV);
+      ACE_TRY_CHECK;
+      if (this->bind_to_server ("Server_MMDevice1") == -1)
+        ACE_ERROR_RETURN ((LM_ERROR, 
+                           "(%P|%t) Error binding to the naming service\n"),
+                          -1);
+      result = this->streamctrl_.bind_devs (AVStreams::MMDevice::_nil (),
+                                            this->server_mmdevice_.in (),
+                                            the_qos.inout (),
+                                            flow_spec,
+                                            ACE_TRY_ENV);
+      ACE_TRY_CHECK;
+      if (this->bind_to_server ("Server_MMDevice2") == -1)
+        ACE_ERROR_RETURN ((LM_ERROR, 
+                           "(%P|%t) Error binding to the naming service\n"),
+                          -1);
+      result = this->streamctrl_.bind_devs (AVStreams::MMDevice::_nil (),
+                                            this->server_mmdevice_.in (),
+                                            the_qos.inout (),
+                                            flow_spec,
+                                            ACE_TRY_ENV);
+      ACE_TRY_CHECK;
+
+      if (result == 0)
+        ACE_ERROR_RETURN ((LM_ERROR,"streamctrl::bind_devs failed\n"),-1);
+      AVStreams::flowSpec start_spec (1);
+      start_spec.length (1);
+      start_spec [0] = CORBA::string_dup (this->flowname_);
+      this->streamctrl_.start (start_spec,ACE_TRY_ENV);
+      ACE_TRY_CHECK;
+      // Schedule a timer for the for the flow handler.
+      TAO_AV_CORE::instance ()->run ();
+    }
+  ACE_CATCHANY
+    {
+      ACE_PRINT_EXCEPTION (ACE_ANY_EXCEPTION,"Client::run");
+      return -1;
+    }
+  ACE_ENDTRY;
+  ACE_CHECK_RETURN (-1);
+  return 0;
+}
+
+int
+main (int argc,
+      char **argv)
+{
+  int result = 0;
+  result = CLIENT::instance ()->init (argc,argv);
+  if (result < 0)
+    ACE_ERROR_RETURN ((LM_ERROR,"client::init failed\n"),1);
+  result = CLIENT::instance ()->run ();
+  if (result < 0)
+    ACE_ERROR_RETURN ((LM_ERROR,"client::run failed\n"),1);
+
+}
+
+#if defined (ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION)
+template class ACE_Singleton <Client,ACE_Null_Mutex>;
+template class TAO_AV_Endpoint_Reactive_Strategy_A<FTP_Client_StreamEndPoint,TAO_VDev,AV_Null_MediaCtrl>;
+template class TAO_AV_Endpoint_Reactive_Strategy<FTP_Client_StreamEndPoint,TAO_VDev,AV_Null_MediaCtrl>;
+#elif defined (ACE_HAS_TEMPLATE_INSTANTIATION_PRAGMA)
+#pragma instantiate ACE_Singleton <Client,ACE_Null_Mutex>
+#pragma instantiate TAO_AV_Endpoint_Reactive_Strategy_A<FTP_Client_StreamEndPoint,TAO_VDev,AV_Null_MediaCtrl> 
+#pragma instantiate TAO_AV_Endpoint_Reactive_Strategy<FTP_Client_StreamEndPoint,TAO_VDev,AV_Null_MediaCtrl> 
+#endif /* ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION */
diff --git a/TAO/orbsvcs/tests/AVStreams/Multicast/ftp.h b/TAO/orbsvcs/tests/AVStreams/Multicast/ftp.h
new file mode 100644
index 00000000000..0b36782eb5a
--- /dev/null
+++ b/TAO/orbsvcs/tests/AVStreams/Multicast/ftp.h
@@ -0,0 +1,116 @@
+/* -*- C++ -*- */
+// $Id$
+
+#ifndef TAO_AV_FTP_H
+#define TAO_AV_FTP_H
+
+#include "ace/Get_Opt.h"
+#include "tao/TAO.h"
+#include "orbsvcs/Naming/Naming_Utils.h"
+#include "orbsvcs/AV/AVStreams_i.h"
+#include "orbsvcs/AV/Endpoint_Strategy.h"
+#include "orbsvcs/AV/Transport.h"
+#include "orbsvcs/AV/sfp.h"
+#include "orbsvcs/AV/MCast.h"
+
+class FTP_Client_Flow_Handler
+  :public virtual ACE_Event_Handler
+{
+public:
+  FTP_Client_Flow_Handler (TAO_ORB_Manager *orb_manager,
+                           ACE_Time_Value &timeout);
+  virtual int handle_timeout (const ACE_Time_Value &tv,
+                              const void *arg = 0);
+  virtual int start (void);
+  virtual int stop (void);
+  virtual int set_protocol_object (TAO_AV_Protocol_Object *object);
+protected:
+  TAO_ORB_Manager *orb_manager_;
+  long timer_id_;
+  int count_;
+  TAO_AV_Protocol_Object *protocol_object_;
+  ACE_Time_Value timeout_;
+};
+
+class FTP_Client_Callback
+  :public TAO_AV_Callback
+{
+public:
+  FTP_Client_Callback (FTP_Client_Flow_Handler *handler);
+  virtual int handle_start (void);
+  virtual int handle_stop (void);
+protected:
+  FTP_Client_Flow_Handler *handler_;
+};
+
+class FTP_Client_StreamEndPoint
+  :public TAO_Client_StreamEndPoint
+{
+public:
+  FTP_Client_StreamEndPoint (TAO_ORB_Manager *orb_manager = 0);
+
+  virtual int get_callback (const char *flowname,
+                            TAO_AV_Callback *&callback);
+
+  virtual int set_protocol_object (const char *flowname,
+                                   TAO_AV_Protocol_Object *object);
+protected:
+  TAO_ORB_Manager *orb_manager_;
+  FTP_Client_Flow_Handler *handler_;
+  FTP_Client_Callback *callback_;
+};
+
+typedef TAO_AV_Endpoint_Reactive_Strategy_A<FTP_Client_StreamEndPoint,TAO_VDev,AV_Null_MediaCtrl> ENDPOINT_STRATEGY;
+
+class Client;
+class Endpoint_Reactive_Strategy
+  : public ENDPOINT_STRATEGY
+{
+public:
+  Endpoint_Reactive_Strategy (TAO_ORB_Manager *orb_manager,
+                              Client *client_ptr);
+  // constructor . The orb manager is needed for the TAO_AV_Endpoint_Reactive_Strategy_A.
+
+  virtual int make_stream_endpoint (FTP_Client_StreamEndPoint *& endpoint);
+  // hook to make our streamendpoint taking a Client pointer
+private:
+  Client *client_;
+  // pointer to command handler object
+  TAO_ORB_Manager *orb_manager_;
+};
+
+
+class Client
+{
+public:
+  Client (void);
+  int init (int argc, char **argv);
+  int run (void);
+  FILE *file (void);
+  char *flowname (void);
+  TAO_StreamCtrl* streamctrl (void);
+private:
+  int parse_args (int argc, char **argv);
+  int bind_to_server (const char *name);
+  TAO_ORB_Manager *orb_manager_;
+  Endpoint_Reactive_Strategy endpoint_strategy_;
+  AVStreams::MMDevice_var server_mmdevice_;
+  TAO_MMDevice client_mmdevice_;
+  TAO_StreamCtrl streamctrl_;
+  // Video stream controller
+
+  int argc_;
+  char **argv_;
+  const char *filename_;
+  const char *address_;
+
+  TAO_Naming_Client my_naming_client_;
+  FILE *fp_;
+  char *protocol_;
+  char *flowname_;
+  int use_sfp_;
+};
+
+typedef ACE_Singleton<Client,ACE_Null_Mutex> CLIENT;
+
+#endif /* TAO_AV_FTP_H */
diff --git a/TAO/orbsvcs/tests/AVStreams/Multicast/server.cpp b/TAO/orbsvcs/tests/AVStreams/Multicast/server.cpp
new file mode 100644
index 00000000000..8609830d663
--- /dev/null
+++ b/TAO/orbsvcs/tests/AVStreams/Multicast/server.cpp
@@ -0,0 +1,217 @@
+// $Id$
+
+#include "server.h"
+
+FTP_Server_StreamEndPoint::FTP_Server_StreamEndPoint (void)
+{
+  ACE_DEBUG ((LM_DEBUG,"FTP_Server_StreamEndPoint::FTP_Server_StreamEndPoint"));
+}
+
+int
+FTP_Server_StreamEndPoint::get_callback (const char *flowname,
+                                         TAO_AV_Callback *&callback)
+{
+  ACE_DEBUG ((LM_DEBUG,"FTP_Server_StreamEndPoint::get_sfp_callback\n"));
+  ACE_NEW_RETURN (callback,
+                  FTP_Server_Callback,
+                  -1);
+  return 0;
+}
+
+int
+FTP_Server_Callback::handle_stop (void)
+{
+  ACE_DEBUG ((LM_DEBUG,"FTP_Server_Callback::stop"));
+  ACE_OS::fclose (SERVER::instance ()->file ());
+  return 0;
+}
+
+int
+FTP_Server_Callback::receive_frame (ACE_Message_Block *frame)
+{
+  ACE_DEBUG ((LM_DEBUG,"FTP_Server_Callback::receive_frame\n"));
+  while (frame != 0)
+    {
+      int result = ACE_OS::fwrite (frame->rd_ptr (),
+                                   frame->length (),
+                                   1,
+                                   SERVER::instance ()->file ());
+      if (result == 0)
+        ACE_ERROR_RETURN ((LM_ERROR,"FTP_Server_Flow_Handler::fwrite failed\n"),-1);
+      frame = frame->cont ();
+    }
+  return 0;
+}
+
+int
+FTP_Server_Callback::handle_end_stream (void)
+{
+  ACE_DEBUG ((LM_DEBUG,"FTP_SFP_Callback::end_stream\n"));
+  CORBA::ORB_var orb = TAO_AV_CORE::instance ()->orb_manager ()->orb ();
+  orb->shutdown ();
+  return 0;
+}
+
+Server::Server (void)
+  :orb_manager_ (TAO_AV_CORE::instance ()->orb_manager ()),
+   reactive_strategy_ (orb_manager_)
+{
+}
+
+int
+Server::init (int argc,
+              char **argv)
+{
+  ACE_DECLARE_NEW_CORBA_ENV;
+  ACE_TRY
+    {
+      TAO_AV_CORE::instance ()->init (argc,
+                                      argv,
+                                      ACE_TRY_ENV);
+      ACE_TRY_CHECK;
+      this->orb_manager_ = 
+        TAO_AV_CORE::instance ()->orb_manager ();
+
+      // Initialize the orb_manager
+      this->orb_manager_->init_child_poa (argc,
+                                         argv,
+                                         "child_poa",
+                                         ACE_TRY_ENV);
+      ACE_TRY_CHECK;
+
+      CORBA::ORB_var orb =
+        this->orb_manager_->orb ();
+
+      PortableServer::POA_var child_poa =
+        this->orb_manager_->child_poa ();
+
+
+      int result = this->parse_args (argc,argv);
+      if (result == -1)
+        ACE_ERROR_RETURN  ((LM_ERROR,"parse args failed\n"),-1);
+      // Initialize the naming services
+
+      if (my_naming_client_.init (orb.in ()) != 0)
+        ACE_ERROR_RETURN ((LM_ERROR,
+                           " (%P|%t) Unable to initialize "
+                           "the TAO_Naming_Client. \n"),
+                          -1);
+
+      // Register the video mmdevice object with the ORB
+      ACE_NEW_RETURN (this->mmdevice_,
+                      TAO_MMDevice (&this->reactive_strategy_),
+                      -1);
+      // create the video server mmdevice with the naming service pointer.
+      this->orb_manager_->activate_under_child_poa ("Server_MMDevice",
+                                                   this->mmdevice_,
+                                                   ACE_TRY_ENV);
+      ACE_TRY_CHECK;
+
+      // Register the mmdevice with the naming service.
+      CosNaming::Name server_mmdevice_name (1);
+      server_mmdevice_name.length (1);
+      server_mmdevice_name [0].id = CORBA::string_dup ("Server_MMDevice1");
+          AVStreams::MMDevice_var mmdevice = this->mmdevice_->_this (ACE_TRY_ENV);
+      ACE_TRY_EX (bind)
+        {
+          // Register the video control object with the naming server.
+          this->my_naming_client_->bind (server_mmdevice_name,
+                                         mmdevice.in (),
+                                         ACE_TRY_ENV);
+          ACE_TRY_CHECK_EX (bind);
+        }
+      ACE_CATCH (CosNaming::NamingContext::AlreadyBound,al_ex)
+        {
+          server_mmdevice_name [0].id = CORBA::string_dup ("Server_MMDevice2");
+          this->my_naming_client_->bind (server_mmdevice_name,
+                                         mmdevice.in (),
+                                         ACE_TRY_ENV);
+          ACE_TRY_CHECK;
+        }
+      ACE_ENDTRY;
+      ACE_CHECK_RETURN (-1);
+    }
+  ACE_CATCHANY
+    {
+      ACE_PRINT_EXCEPTION (ACE_ANY_EXCEPTION,"server::init");
+      return -1;
+    }
+  ACE_ENDTRY;
+  ACE_CHECK_RETURN (-1);
+  return 0;
+}
+
+int
+Server::run (void)
+{
+  ACE_DECLARE_NEW_CORBA_ENV;
+  ACE_TRY
+    {
+      this->orb_manager_->run (ACE_TRY_ENV);
+      ACE_TRY_CHECK;
+    }
+    ACE_CATCHANY
+    {
+      ACE_PRINT_EXCEPTION (ACE_ANY_EXCEPTION,"server::init");
+      return -1;
+    }
+  ACE_ENDTRY;
+  ACE_CHECK_RETURN (-1);
+  return 0;
+}
+
+int
+Server::parse_args (int argc,char **argv)
+{
+  ACE_Get_Opt opts (argc,argv,"f:p:");
+
+  char c;
+  while ((c = opts ()) != -1)
+    {
+      switch (c)
+        {
+        case 'f':
+          this->fp_ = ACE_OS::fopen (opts.optarg,"w");
+          if (this->fp_ != 0)
+            {
+              ACE_DEBUG ((LM_DEBUG,"file opened successfully\n"));
+            }
+          break;
+        case 'p':
+          this->protocol_ = ACE_OS::strdup (opts.optarg);
+        default:
+          ACE_ERROR_RETURN ((LM_ERROR,"Usage: server -f filename"),-1);
+          break;
+        }
+    }
+  return 0;
+}
+
+FILE*
+Server::file (void)
+{
+  return this->fp_;
+}
+
+int
+main (int argc,
+      char **argv)
+{
+  int result = 0;
+  result = SERVER::instance ()->init (argc,argv);
+  if (result < 0)
+    ACE_ERROR_RETURN ((LM_ERROR,"SERVER::init failed\n"),1);
+  result = SERVER::instance ()->run ();
+  if (result < 0)
+    ACE_ERROR_RETURN ((LM_ERROR,"SERVER::run failed\n"),1);
+}
+
+#if defined (ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION)
+template class ACE_Singleton <Server,ACE_Null_Mutex>;
+template class TAO_AV_Endpoint_Reactive_Strategy_B <FTP_Server_StreamEndPoint,TAO_VDev,AV_Null_MediaCtrl>;
+template class TAO_AV_Endpoint_Reactive_Strategy <FTP_Server_StreamEndPoint,TAO_VDev,AV_Null_MediaCtrl>;
+#elif defined (ACE_HAS_TEMPLATE_INSTANTIATION_PRAGMA)
+#pragma instantiate ACE_Singleton <Server,ACE_Null_Mutex>
+#pragma instantiate TAO_AV_Endpoint_Reactive_Strategy_B <FTP_Server_StreamEndPoint,TAO_VDev,AV_Null_MediaCtrl>
+#pragma instantiate TAO_AV_Endpoint_Reactive_Strategy <FTP_Server_StreamEndPoint,TAO_VDev,AV_Null_MediaCtrl>
+#endif /* ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION */
diff --git a/TAO/orbsvcs/tests/AVStreams/Multicast/server.h b/TAO/orbsvcs/tests/AVStreams/Multicast/server.h
new file mode 100644
index 00000000000..04c90ba2d99
--- /dev/null
+++ b/TAO/orbsvcs/tests/AVStreams/Multicast/server.h
@@ -0,0 +1,51 @@
+/* -*- C++ -*- */
+// $Id$
+
+#include "ace/Get_Opt.h"
+#include "tao/TAO.h"
+#include "orbsvcs/orbsvcs/CosNamingC.h"
+#include "orbsvcs/Naming/Naming_Utils.h"
+#include "orbsvcs/AV/AVStreams_i.h"
+#include "orbsvcs/AV/Endpoint_Strategy.h"
+#include "orbsvcs/AV/Transport.h"
+#include "orbsvcs/AV/sfp.h"
+#include "orbsvcs/AV/MCast.h"
+#include "orbsvcs/AV/Policy.h"
+
+class FTP_Server_StreamEndPoint
+  :public TAO_Server_StreamEndPoint
+{
+public:
+  FTP_Server_StreamEndPoint (void);
+  virtual int get_callback (const char *flowname,
+                            TAO_AV_Callback *&callback);
+};
+
+class FTP_Server_Callback
+  :public TAO_AV_Callback
+{
+public:
+  virtual int handle_stop (void);
+  virtual int receive_frame (ACE_Message_Block *frame);
+  virtual int handle_end_stream (void);
+};
+
+class Server
+{
+public:
+  Server (void);
+  int init (int argc,
+            char **argv);
+  int run (void);
+  FILE *file (void);
+protected:
+  int parse_args (int argc,char **argv);
+  TAO_ORB_Manager *orb_manager_;
+  TAO_Naming_Client my_naming_client_;
+  TAO_AV_Endpoint_Reactive_Strategy_B <FTP_Server_StreamEndPoint,TAO_VDev,AV_Null_MediaCtrl> reactive_strategy_;
+  TAO_MMDevice *mmdevice_;
+  FILE *fp_;
+  char *protocol_;
+};
+
+typedef ACE_Singleton<Server,ACE_Null_Mutex> SERVER;