Added a new test similar to the file transfer program under Pluggable directory.

This test make use of the full profile classes. A FDev create flow consumers
and flow producers and is added to the MMDevice. During streamconnection
this fdev creates the consumer/producer depending on whether its a client
or a server. The flow endpoint is also a base_streamendpoint ensuring that
the pluggable data protocols can set the protocol object.

5 files changed, 912 insertions, 0 deletions

diff --git a/TAO/orbsvcs/tests/AVStreams/Full_Profile/Makefile b/TAO/orbsvcs/tests/AVStreams/Full_Profile/Makefile
new file mode 100644
index 00000000000..d2f38f81303
--- /dev/null
+++ b/TAO/orbsvcs/tests/AVStreams/Full_Profile/Makefile
@@ -0,0 +1,57 @@
+#----------------------------------------------------------------------------
+#	$Id$
+#
+#	Top-level Makefile for Full Profile 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/Full_Profile/ftp.cpp b/TAO/orbsvcs/tests/AVStreams/Full_Profile/ftp.cpp
new file mode 100644
index 00000000000..af9e19167b0
--- /dev/null
+++ b/TAO/orbsvcs/tests/AVStreams/Full_Profile/ftp.cpp
@@ -0,0 +1,417 @@
+// $Id$
+
+#include "ftp.h"
+
+FTP_Client_Callback::FTP_Client_Callback (FTP_Client_Flow_Handler *handler)
+  :handler_ (handler)
+{
+}
+
+int
+FTP_Client_Callback::handle_start (void)
+{
+  ACE_DEBUG ((LM_DEBUG,"FTP_Client_Callback::handle_start"));
+  return this->handler_->start ();
+}
+
+int
+FTP_Client_Callback::handle_stop (void)
+{
+  ACE_DEBUG ((LM_DEBUG,"FTP_Client_Callback::handle_stop"));
+  return this->handler_->stop ();
+}
+
+
+FTP_Client_Flow_Handler::FTP_Client_Flow_Handler (TAO_ORB_Manager *orb_manager,
+                                                  ACE_Time_Value &timeout)
+  :TAO_FlowProducer ("Data",CLIENT::instance ()->protocols (),CLIENT::instance ()->format ()),
+   orb_manager_ (orb_manager),
+   count_ (0),
+   timeout_ (timeout)
+{
+}
+
+int
+FTP_Client_Flow_Handler::get_callback (const char *flowname,
+                                       TAO_AV_Callback *&callback)
+{
+  ACE_DEBUG ((LM_DEBUG,"FTP_Client_Flow_Handler::get_callback\n"));
+  ACE_NEW_RETURN (callback,
+                  FTP_Client_Callback (this),
+                  -1);
+  return 0;
+}
+
+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::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);
+              return 0;
+            }
+          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"));
+}
+
+//------------------------------------------------------------
+// FTP_Client_FDev
+//------------------------------------------------------------
+
+FTP_Client_FDev::FTP_Client_FDev (TAO_ORB_Manager *orb_manager)
+  :TAO_FDev (CORBA::string_dup ("Data")),
+   orb_manager_ (orb_manager)
+{
+}
+
+AVStreams::FlowProducer_ptr
+FTP_Client_FDev::make_producer (AVStreams::FlowConnection_ptr the_requester,
+                                AVStreams::QoS & the_qos,
+                                CORBA::Boolean_out met_qos,
+                                char *& named_fdev,
+                                CORBA::Environment &ACE_TRY_ENV)
+{
+  ACE_DEBUG ((LM_DEBUG,"FTP_Client_FDev::make_producer\n"));
+  FTP_Client_Flow_Handler *handler;
+  ACE_Time_Value timeout (2);
+  ACE_NEW_RETURN (handler,
+                  FTP_Client_Flow_Handler (this->orb_manager_,
+                                           timeout),
+                  0);
+  AVStreams::FlowProducer_ptr producer = handler->_this (ACE_TRY_ENV);
+  ACE_CHECK_RETURN (0);
+  return producer;
+}
+
+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_;
+}
+
+AVStreams::protocolSpec
+Client::protocols (void)
+{
+  AVStreams::protocolSpec protocols (1);
+  protocols.length (1);
+  char buf [BUFSIZ];
+  ACE_OS::sprintf (buf,"%s=%s",this->protocol_,this->address_);
+  protocols [0] = CORBA::string_dup (buf);
+  return protocols;
+}
+
+const char *
+Client::format (void)
+{
+  return "UNS:ftp";
+}
+
+const char *
+Client::address (void)
+
+{
+  return this->address_;
+}
+
+TAO_StreamCtrl*
+Client::streamctrl (void)
+{
+  return &this->streamctrl_;
+}
+
+Client::Client (void)
+  :orb_manager_ (TAO_AV_CORE::instance ()->orb_manager ()),
+   endpoint_strategy_ (orb_manager_),
+   client_mmdevice_ (&endpoint_strategy_),
+   fdev_ (0),
+   fp_ (0),
+   protocol_ (ACE_OS::strdup ("UDP"))
+{
+}
+
+
+int
+Client::bind_to_server (void)
+{
+  ACE_DECLARE_NEW_CORBA_ENV;
+
+  ACE_TRY
+    {
+      // Initialize the naming services
+      if (my_naming_client_.init (this->orb_manager_->orb ()) != 0)
+        ACE_ERROR_RETURN ((LM_ERROR,
+                           " (%P|%t) Unable to initialize "
+                           "the TAO_Naming_Client. \n"),
+                          -1);
+      
+      CosNaming::Name server_mmdevice_name (1);
+      server_mmdevice_name.length (1);
+      server_mmdevice_name [0].id = CORBA::string_dup ("Server_MMDevice");
+      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;
+
+      this->orb_manager_->activate_poa_manager (ACE_TRY_ENV);
+      ACE_TRY_CHECK;
+      ACE_NEW_RETURN (this->fdev_,
+                      FTP_Client_FDev (this->orb_manager_),
+                      -1);
+
+      AVStreams::MMDevice_var mmdevice = this->client_mmdevice_._this (ACE_TRY_ENV);
+      ACE_TRY_CHECK;
+      AVStreams::FDev_var fdev = this->fdev_->_this (ACE_TRY_ENV);
+      ACE_TRY_CHECK;
+      mmdevice->add_fdev (fdev.in (),
+                          ACE_TRY_ENV);
+
+      // 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"));
+        }
+      if (this->bind_to_server () == -1)
+        ACE_ERROR_RETURN ((LM_ERROR, 
+                           "(%P|%t) Error binding to the naming service\n"),
+                          -1);
+    }
+  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.
+
+      ACE_INET_Addr addr (this->address_);
+      ACE_NEW_RETURN (this->flowname_,
+                      char [BUFSIZ],
+                      0);
+      ACE_OS::sprintf (this->flowname_,
+                       "Data",
+                       this->protocol_);
+      TAO_Forward_FlowSpec_Entry entry (this->flowname_,
+                                        "IN",
+                                        "USER_DEFINED",
+                                        flow_protocol_str,
+                                        this->protocol_,
+                                        &addr);
+      flow_spec [0] = CORBA::string_dup (entry.entry_to_string ());
+      flow_spec.length (1);
+      CORBA::Boolean result = 
+        this->streamctrl_.bind_devs (this->client_mmdevice_._this (ACE_TRY_ENV),
+                                     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/Full_Profile/ftp.h b/TAO/orbsvcs/tests/AVStreams/Full_Profile/ftp.h
new file mode 100644
index 00000000000..a82b1b30a08
--- /dev/null
+++ b/TAO/orbsvcs/tests/AVStreams/Full_Profile/ftp.h
@@ -0,0 +1,100 @@
+/* -*- 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 virtual TAO_FlowProducer
+{
+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 get_callback (const char *flowname,
+                            TAO_AV_Callback *&callback);
+protected:
+  TAO_ORB_Manager *orb_manager_;
+  long timer_id_;
+  int count_;
+  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_FDev
+  :public TAO_FDev
+{
+public:
+  FTP_Client_FDev (TAO_ORB_Manager *orb_manger);
+  virtual AVStreams::FlowProducer_ptr make_producer (AVStreams::FlowConnection_ptr the_requester,
+                                                     AVStreams::QoS & the_qos,
+                                                     CORBA::Boolean_out met_qos,
+                                                     char *& named_fdev,
+                                                     CORBA::Environment &env = CORBA::Environment::default_environment ());
+protected:
+  TAO_ORB_Manager *orb_manager_;
+};
+
+typedef TAO_AV_Endpoint_Reactive_Strategy_A<TAO_StreamEndPoint_A,TAO_VDev,AV_Null_MediaCtrl> ENDPOINT_STRATEGY;
+
+class Client
+{
+public:
+  Client (void);
+  int init (int argc, char **argv);
+  int run (void);
+  FILE *file (void);
+  char *flowname (void);
+  TAO_StreamCtrl* streamctrl (void);
+  AVStreams::protocolSpec protocols (void);
+  const char *format (void);
+  const char *address (void);
+private:
+  int parse_args (int argc, char **argv);
+  int bind_to_server (void);
+  TAO_ORB_Manager *orb_manager_;
+  ENDPOINT_STRATEGY endpoint_strategy_;
+  AVStreams::MMDevice_var server_mmdevice_;
+  TAO_MMDevice client_mmdevice_;
+  FTP_Client_FDev *fdev_;
+  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/Full_Profile/server.cpp b/TAO/orbsvcs/tests/AVStreams/Full_Profile/server.cpp
new file mode 100644
index 00000000000..aa80fd6923a
--- /dev/null
+++ b/TAO/orbsvcs/tests/AVStreams/Full_Profile/server.cpp
@@ -0,0 +1,270 @@
+// $Id$
+
+#include "server.h"
+
+FTP_Server_FlowEndPoint::FTP_Server_FlowEndPoint (void)
+  :TAO_FlowConsumer ("Data",SERVER::instance ()->protocols (),SERVER::instance ()->format ())
+{
+  AVStreams::protocolSpec protocols (2);
+  protocols.length (2);
+  protocols [0] = CORBA::string_dup ("TCP");
+  protocols [1] = CORBA::string_dup ("UDP");
+  ACE_DECLARE_NEW_CORBA_ENV;
+  ACE_TRY
+    {
+      this->set_protocol_restriction (protocols,
+                                      ACE_TRY_ENV);
+      ACE_TRY_CHECK;
+    }
+  ACE_CATCHANY
+    {
+      ACE_PRINT_EXCEPTION (ACE_ANY_EXCEPTION,"FTP_Server_FlowEndPoint::FTP_Server_FlowEndPoint");
+    }
+  ACE_ENDTRY;
+  ACE_CHECK;
+}
+
+int
+FTP_Server_FlowEndPoint::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;
+}
+
+FTP_Server_FDev::FTP_Server_FDev (void)
+  :TAO_FDev ("Data")
+{
+}
+
+AVStreams::FlowConsumer_ptr 
+FTP_Server_FDev::make_consumer (AVStreams::FlowConnection_ptr the_requester,
+                                AVStreams::QoS & the_qos,
+                                CORBA::Boolean_out met_qos,
+                                char *& named_fdev,
+                                CORBA::Environment &ACE_TRY_ENV)
+{
+  ACE_DEBUG ((LM_DEBUG,"FTP_Server_FDev::make_consumer"));
+  FTP_Server_FlowEndPoint *endpoint;
+  ACE_NEW_RETURN (endpoint,
+                  FTP_Server_FlowEndPoint,
+                  0);
+  return endpoint->_this (ACE_TRY_ENV);
+  ACE_CHECK_RETURN (0);
+}
+
+Server::Server (void)
+  :orb_manager_ (TAO_AV_CORE::instance ()->orb_manager ()),
+   reactive_strategy_ (orb_manager_)
+{
+}
+
+AVStreams::protocolSpec
+Server::protocols (void)
+{
+  AVStreams::protocolSpec protocols (2);
+  protocols.length (2);
+  protocols [0] = CORBA::string_dup ("TCP");
+  protocols [1] = CORBA::string_dup ("UDP");
+  return protocols;
+}
+
+const char*
+Server::format (void)
+{
+  return "UNS:ftp";
+}
+
+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);
+
+      this->orb_manager_->activate_poa_manager (ACE_TRY_ENV);
+      ACE_TRY_CHECK;
+      // Register the video mmdevice object with the ORB
+      ACE_NEW_RETURN (this->mmdevice_,
+                      TAO_MMDevice (&this->reactive_strategy_),
+                      -1);
+
+      ACE_NEW_RETURN (this->fdev_,
+                      FTP_Server_FDev,
+                      -1);
+      AVStreams::MMDevice_var mmdevice = this->mmdevice_->_this (ACE_TRY_ENV);
+      ACE_TRY_CHECK;
+      AVStreams::FDev_var fdev = this->fdev_->_this (ACE_TRY_ENV);
+      ACE_TRY_CHECK;
+      mmdevice->add_fdev (fdev.in (),
+                          ACE_TRY_ENV);
+      ACE_TRY_CHECK;
+      // 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_MMDevice");
+
+      // Register the video control object with the naming server.
+      this->my_naming_client_->rebind (server_mmdevice_name,
+                                       mmdevice.in (),
+                                       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::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/Full_Profile/server.h b/TAO/orbsvcs/tests/AVStreams/Full_Profile/server.h
new file mode 100644
index 00000000000..524b6653029
--- /dev/null
+++ b/TAO/orbsvcs/tests/AVStreams/Full_Profile/server.h
@@ -0,0 +1,68 @@
+/* -*- C++ -*- */
+// $Id$
+
+#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"
+#include "orbsvcs/AV/Policy.h"
+
+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 FTP_Server_FlowEndPoint
+  :public TAO_FlowConsumer
+{
+public:
+  FTP_Server_FlowEndPoint (void);
+  int get_callback (const char *flowname,
+                    TAO_AV_Callback *&callback);
+};
+
+class FTP_Server_FDev
+  :public TAO_FDev
+{
+public:
+  FTP_Server_FDev (void);
+  virtual AVStreams::FlowConsumer_ptr make_consumer (AVStreams::FlowConnection_ptr the_requester,
+                                                       AVStreams::QoS & the_qos,
+                                                       CORBA::Boolean_out met_qos,
+                                                       char *& named_fdev,
+                                                       CORBA::Environment &env = CORBA::Environment::default_environment ());
+  // bridge method for the application to override the consumer object
+  // creation. Default implementation creates a TAO_FlowConsumer.
+
+};
+
+class Server
+{
+public:
+  Server (void);
+  int init (int argc,
+            char **argv);
+  int run (void);
+  FILE *file (void);
+  AVStreams::protocolSpec protocols (void);
+  const char *format (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 <TAO_StreamEndPoint_B,TAO_VDev,AV_Null_MediaCtrl> reactive_strategy_;
+  TAO_MMDevice *mmdevice_;
+  FTP_Server_FDev *fdev_;
+  FILE *fp_;
+  char *protocol_;
+};
+
+typedef ACE_Singleton<Server,ACE_Null_Mutex> SERVER;