TAO/orbsvcs/tests/AVStreams/Pluggable/ftp.cpp

   1 #include "ftp.h"
   2 #include "tao/debug.h"
   3 #include "ace/Get_Opt.h"
   4 #include "ace/High_Res_Timer.h"
   5
   6 ACE_High_Res_Timer last_frame_sent_time;
   7 // The time taken for sending a frmae and preparing for the next frame
   8
   9 ACE_Time_Value inter_frame_time;
  10 // The time that should lapse between two consecutive frames sent.
  11
  12 FTP_Client_Callback::FTP_Client_Callback ()
  13 {
  14 }
  15
  16 FTP_Client_StreamEndPoint::FTP_Client_StreamEndPoint ()
  17 {
  18 }
  19
  20 int
  21 FTP_Client_StreamEndPoint::get_callback (const char *,
  22                                          TAO_AV_Callback *&callback)
  23 {
  24   // Create and return the clienmt application callback and return to the AVStreams
  25   // for further upcalls.
  26   callback = &this->callback_;
  27   return 0;
  28 }
  29
  30 int
  31 FTP_Client_StreamEndPoint::set_protocol_object (const char *,
  32                                                 TAO_AV_Protocol_Object *object)
  33 {
  34   // Set the client protocol object corresponding to the transport protocol selected.
  35   CLIENT::instance ()->set_protocol_object (object);
  36   return 0;
  37 }
  38
  39 Client::Client ()
  40   :client_mmdevice_ (&endpoint_strategy_),
  41    count_ (0),
  42    address_ (0),
  43    peer_addr_str_ (0),
  44    fp_ (0),
  45    protocol_ (ACE_OS::strdup ("UDP")),
  46    frame_rate_ (30)
  47 {
  48   this->mb.size (BUFSIZ);
  49 }
  50
  51 void
  52 Client::set_protocol_object (TAO_AV_Protocol_Object *object)
  53 {
  54   // Set the client protocol object corresponding to the transport protocol selected.
  55   this->protocol_object_ = object;
  56 }
  57
  58 int
  59 Client::parse_args (int argc,
  60                     ACE_TCHAR *argv[])
  61 {
  62   // Parse command line arguments
  63   ACE_Get_Opt opts (argc, argv, ACE_TEXT("f:l:a:p:r:sd"));
  64
  65   this->use_sfp_ = 0;
  66
  67   int c;
  68   while ((c= opts ()) != -1)
  69     {
  70       switch (c)
  71         {
  72         case 'f':
  73           this->filename_ = ACE_OS::strdup (ACE_TEXT_ALWAYS_CHAR (opts.opt_arg ()));
  74           break;
  75         case 'l':
  76           this->address_ = ACE_OS::strdup (ACE_TEXT_ALWAYS_CHAR (opts.opt_arg ()));
  77           break;
  78         case 'a':
  79           this->peer_addr_str_ = ACE_OS::strdup (ACE_TEXT_ALWAYS_CHAR (opts.opt_arg ()));
  80           break;
  81         case 'p':
  82           this->protocol_ = ACE_OS::strdup (ACE_TEXT_ALWAYS_CHAR (opts.opt_arg ()));
  83           break;
  84         case 'r':
  85           this->frame_rate_ = ACE_OS::atoi (opts.opt_arg ());
  86           break;
  87         case 's':
  88           this->use_sfp_ = 1;
  89           break;
  90         case 'd':
  91           TAO_debug_level++;
  92           break;
  93         default:
  94           ACE_DEBUG ((LM_DEBUG,"Unknown Option\n"));
  95           return -1;
  96         }
  97     }
  98   return 0;
  99 }
 100
 101 FILE *
 102 Client::file ()
 103 {
 104   return this->fp_;
 105 }
 106
 107 char*
 108 Client::flowname ()
 109 {
 110   return this->flowname_;
 111 }
 112
 113 TAO_StreamCtrl*
 114 Client::streamctrl ()
 115 {
 116   return &this->streamctrl_;
 117 }
 118
 119
 120 int
 121 Client::frame_rate ()
 122 {
 123   return this->frame_rate_;
 124 }
 125
 126
 127 // Method to get the object reference of the server
 128 int
 129 Client::bind_to_server ()
 130 {
 131   // Initialize the naming services
 132   if (my_naming_client_.init (TAO_AV_CORE::instance ()->orb ()) != 0)
 133     ACE_ERROR_RETURN ((LM_ERROR,
 134                        " (%P|%t) Unable to initialize "
 135                        "the TAO_Naming_Client.\n"),
 136                       -1);
 137
 138   CosNaming::Name server_mmdevice_name (1);
 139   server_mmdevice_name.length (1);
 140   server_mmdevice_name [0].id = CORBA::string_dup ("Server_MMDevice");
 141
 142   // Resolve the server object reference from the Naming Service
 143   CORBA::Object_var server_mmdevice_obj =
 144     my_naming_client_->resolve (server_mmdevice_name);
 145
 146   this->server_mmdevice_ =
 147     AVStreams::MMDevice::_narrow (server_mmdevice_obj.in ());
 148
 149   if (CORBA::is_nil (this->server_mmdevice_.in ()))
 150     ACE_ERROR_RETURN ((LM_ERROR,
 151                        "Could not resolve Server_MMdevice in Naming service <%s>\n"),
 152                       -1);
 153
 154   return 0;
 155 }
 156
 157 int
 158 Client::init (int argc,
 159               ACE_TCHAR *argv[])
 160 {
 161   this->argc_ = argc;
 162   this->argv_ = argv;
 163
 164   CORBA::String_var ior;
 165
 166   // Initialize the endpoint strategy with the orb and poa.
 167   this->endpoint_strategy_.init(TAO_AV_CORE::instance ()->orb (),
 168                                 TAO_AV_CORE::instance ()->poa ());
 169
 170   // Parse the command line arguments
 171   int result = this->parse_args (argc,
 172                                  argv);
 173
 174   if (result != 0)
 175     return result;
 176
 177   // Open file to read.
 178   this->fp_ = ACE_OS::fopen (this->filename_,
 179                              "r");
 180   if (this->fp_ == 0)
 181     ACE_ERROR_RETURN ((LM_DEBUG,
 182                        "Cannot open input file %s\n",
 183                        this->filename_),
 184                       -1);
 185
 186   result
 187     = this->bind_to_server ();
 188
 189   // Resolve the object reference of the server from the Naming Service.
 190   if (result != 0)
 191     ACE_ERROR_RETURN ((LM_ERROR,
 192                        "(%P|%t) Error binding to the naming service\n"),
 193                       -1);
 194
 195   // Create the Flow protocol name
 196   char flow_protocol_str [BUFSIZ];
 197   if (this->use_sfp_)
 198     ACE_OS::strcpy (flow_protocol_str,
 199                     "sfp:1.0");
 200   else
 201     ACE_OS::strcpy (flow_protocol_str,
 202                     "");
 203
 204   // Initialize the  QoS
 205   AVStreams::streamQoS_var the_qos (new AVStreams::streamQoS);
 206
 207   // Set the address of the ftp client.
 208   ACE_INET_Addr* addr;
 209   if (this->address_ != 0)
 210     ACE_NEW_RETURN (addr,
 211                     ACE_INET_Addr (this->address_),
 212                     -1);
 213   else
 214     {
 215       char buf [BUFSIZ];
 216       ACE_OS::hostname (buf,
 217                         BUFSIZ);
 218       ACE_NEW_RETURN (addr,
 219                       ACE_INET_Addr ("5000",
 220                                      buf),
 221                       -1);
 222     }
 223
 224   // Initialize the flowname
 225   ACE_NEW_RETURN (this->flowname_,
 226                   char [BUFSIZ],
 227                   0);
 228
 229   ACE_OS::sprintf (this->flowname_,
 230                    "Data_%s",
 231                    this->protocol_);
 232
 233   // Create the forward flow specification to describe the flow.
 234   TAO_Forward_FlowSpec_Entry entry (this->flowname_,
 235                                     "IN",
 236                                     "USER_DEFINED",
 237                                     flow_protocol_str,
 238                                     this->protocol_,
 239                                     addr);
 240
 241   ACE_INET_Addr* peer_addr;
 242   if (this->peer_addr_str_ != 0)
 243     ACE_NEW_RETURN (peer_addr,
 244                     ACE_INET_Addr (this->peer_addr_str_),
 245                     -1);
 246   else
 247     {
 248       char buf [BUFSIZ];
 249       ACE_OS::hostname (buf,
 250                         BUFSIZ);
 251       ACE_NEW_RETURN (peer_addr,
 252                       ACE_INET_Addr ("5050",
 253                                      buf),
 254                       -1);
 255     }
 256
 257   entry.set_peer_addr (peer_addr);
 258
 259   AVStreams::flowSpec flow_spec (1);
 260   flow_spec.length (1);
 261   flow_spec [0] = CORBA::string_dup (entry.entry_to_string ());
 262
 263   AVStreams::MMDevice_var client_mmdevice =
 264     this->client_mmdevice_._this ();
 265
 266   // Bind/Connect  the client and server MMDevices.
 267   CORBA::Boolean bind_result =
 268     this->streamctrl_.bind_devs (client_mmdevice.in (),
 269                                  this->server_mmdevice_.in (),
 270                                  the_qos.inout (),
 271                                  flow_spec);
 272
 273   if (bind_result == 0)
 274     ACE_ERROR_RETURN ((LM_ERROR,"streamctrl::bind_devs failed\n"),-1);
 275
 276   return 0;
 277 }
 278
 279 // Method to send data at the specified rate
 280 int
 281 Client::pace_data ()
 282 {
 283   // Rate at which frames of data need to be sent.
 284   this->frame_rate_ = CLIENT::instance ()->frame_rate ();
 285
 286   // Time within which a frame should be sent.
 287   double frame_time = 1/ (double) this->frame_rate_;
 288
 289   if (TAO_debug_level > 0)
 290     ACE_DEBUG ((LM_DEBUG,
 291                 "Frame Time ONE = %f\n Frame Rate = %d\n",
 292                 frame_time,
 293                 this->frame_rate_));
 294
 295   // The time between two consecutive frames.
 296   inter_frame_time.set (frame_time);
 297
 298   if (TAO_debug_level > 0)
 299     ACE_DEBUG ((LM_DEBUG,
 300                 "Inter Frame Time = %d\n",
 301                 inter_frame_time.msec ()));
 302
 303   try
 304     {
 305       // Continue to send data till the file is read to the end.
 306       while (1)
 307         {
 308           // Count the frames sent.
 309           count_++;
 310
 311           // Read from the file into a message block.
 312           size_t n = ACE_OS::fread(this->mb.rd_ptr (),
 313                                    1,
 314                                    this->mb.size (),
 315                                    CLIENT::instance ()->file ());
 316
 317           if (n == 0)
 318             {
 319               if (feof (CLIENT::instance ()->file ()))
 320                 {
 321                   // At end of file break the loop and end the client.
 322                   if (TAO_debug_level > 0)
 323                     ACE_DEBUG ((LM_DEBUG,"Handle_Start:End of file\n"));
 324                   break;
 325                 }
 326
 327             }
 328
 329           this->mb.wr_ptr (n);
 330
 331           if ( this->count_ > 1)
 332             {
 333               // Greater than the first frame.
 334
 335               // Stop the timer that was started just before the previous frame was sent.
 336               last_frame_sent_time.stop ();
 337
 338               // Get the time elapsed after sending the previous frame.
 339               ACE_Time_Value tv;
 340               last_frame_sent_time.elapsed_time (tv);
 341
 342               if (TAO_debug_level > 0)
 343                 ACE_DEBUG ((LM_DEBUG,
 344                             "Elapsed Time = %d\n",
 345                             tv.msec ()));
 346
 347               // Check to see if the inter frame time has elapsed.
 348               if (tv < inter_frame_time)
 349                 {
 350                   // Inter frame time has not elapsed.
 351
 352                   // Claculate the time to wait before the next frame needs to be sent.
 353                   ACE_Time_Value wait_time (inter_frame_time - tv);
 354
 355                   if (TAO_debug_level > 0)
 356                     ACE_DEBUG ((LM_DEBUG,
 357                                 "Wait Time = %d\n",
 358                                 wait_time.msec ()));
 359
 360                   // run the orb for the wait time so the client can continue other orb requests.
 361                   TAO_AV_CORE::instance ()->orb ()->run (wait_time);
 362                 }
 363             }
 364
 365           // Start timer before sending the frame.
 366           last_frame_sent_time.start ();
 367
 368           // Send frame.
 369           int result = this->protocol_object_->send_frame (&this->mb);
 370           if (result < 0)
 371             ACE_ERROR_RETURN ((LM_ERROR,
 372                                "send failed:%p","FTP_Client_Flow_Handler::send\n"),
 373                               -1);
 374           ACE_DEBUG ((LM_DEBUG,"Client::pace_data buffer sent successfully\n"));
 375
 376           // Reset the mb.
 377           this->mb.reset ();
 378         } // end while
 379
 380       // Since the file is read stop the stream.
 381       AVStreams::flowSpec stop_spec (1);
 382       CLIENT::instance ()->streamctrl ()->destroy (stop_spec);
 383
 384       // Shut the orb down.
 385       TAO_AV_CORE::instance ()->orb ()->shutdown (0);
 386     }
 387   catch (const CORBA::Exception& ex)
 388     {
 389       ex._tao_print_exception ("Client::pace_data Failed");
 390       return -1;
 391     }
 392   return 0;
 393 }
 394
 395 int
 396 ACE_TMAIN (int argc, ACE_TCHAR *argv[])
 397 {
 398   try
 399     {
 400       CORBA::ORB_var orb = CORBA::ORB_init (argc, argv);
 401
 402       CORBA::Object_var obj
 403         = orb->resolve_initial_references ("RootPOA");
 404
 405
 406       //Get the POA_var object from Object_var
 407       PortableServer::POA_var root_poa
 408         = PortableServer::POA::_narrow (obj.in ());
 409
 410       PortableServer::POAManager_var mgr
 411         = root_poa->the_POAManager ();
 412
 413       mgr->activate ();
 414
 415       // Initialize the AV STream components.
 416       TAO_AV_CORE::instance ()->init (orb.in (),
 417                                       root_poa.in ());
 418
 419       // INitialize the Client.
 420       int result = 0;
 421       result = CLIENT::instance ()->init (argc,
 422                                           argv);
 423
 424       if (result < 0)
 425         ACE_ERROR_RETURN ((LM_ERROR,
 426                            "client::init failed\n"),1);
 427
 428       // Start sending data.
 429       result = CLIENT::instance ()->pace_data ();
 430
 431       orb->destroy ();
 432     }
 433   catch (const CORBA::Exception& ex)
 434
 435     {
 436     ex._tao_print_exception ("Client Failed\n");
 437     return -1;
 438     }
 439
 440   CLIENT::close ();
 441
 442   return 0;
 443 }
 444
 445 #if defined (ACE_HAS_EXPLICIT_STATIC_TEMPLATE_MEMBER_INSTANTIATION)
 446 template ACE_Unmanaged_Singleton<Client, ACE_Null_Mutex> *ACE_Unmanaged_Singleton<Client, ACE_Null_Mutex>::singleton_;
 447 #endif /* ACE_HAS_EXPLICIT_STATIC_TEMPLATE_MEMBER_INSTANTIATION */