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

   1 #include "ftp.h"
   2
   3
   4 FTP_Client_Callback::FTP_Client_Callback ()
   5   :count_ (0)
   6 {
   7 }
   8
   9 int
  10 FTP_Client_Callback::handle_end_stream ()
  11 {
  12   TAO_AV_CORE::instance ()->stop_run ();
  13   return 0;
  14 }
  15
  16 void
  17 FTP_Client_Callback::get_timeout (ACE_Time_Value *&tv,
  18                                   void *&)
  19 {
  20   ACE_Time_Value *timeout;
  21   ACE_NEW (timeout,
  22            ACE_Time_Value(2));
  23   tv = timeout;
  24 }
  25
  26 int
  27 FTP_Client_Callback::handle_timeout (void *)
  28 {
  29   try
  30     {
  31       ACE_Message_Block mb (BUFSIZ);
  32       ACE_DEBUG ((LM_DEBUG,"FTP_Client_Callback::handle_timeout\n"));
  33       char *buf = mb.rd_ptr ();
  34       int n = ACE_OS::fread(buf,1,mb.size (),CLIENT::instance ()->file ());
  35       if (n < 0)
  36         {
  37           ACE_ERROR_RETURN ((LM_ERROR,"FTP_Client_Flow_Handler::fread end of file\n"),-1);
  38         }
  39
  40       if (n == 0)
  41         {
  42           if (feof (CLIENT::instance ()->file ()))
  43             {
  44               // wait for sometime for the data to be flushed to the other side.
  45               this->count_++;
  46               if (this->count_ == 2)
  47                 {
  48                   ACE_DEBUG ((LM_DEBUG,"handle_timeout:End of file\n"));
  49                   AVStreams::flowSpec stop_spec (1);
  50                   CLIENT::instance ()->streamctrl ()->stop (stop_spec);
  51                   //CLIENT::instance ()->streamctrl ()->destroy (stop_spec);
  52                   TAO_AV_CORE::instance ()->orb ()->shutdown (0);
  53                   return 0;
  54                 }
  55               else
  56                 return 0;
  57             }
  58           else
  59             ACE_ERROR_RETURN ((LM_ERROR,"FTP_Client_Flow_Handler::fread error\n"),-1);
  60         }
  61       mb.wr_ptr (n);
  62       int result = this->protocol_object_->send_frame (&mb);
  63       if (result < 0)
  64         ACE_ERROR_RETURN ((LM_ERROR,"send failed:%p","FTP_Client_Flow_Handler::send\n"),-1);
  65       ACE_DEBUG ((LM_DEBUG,"handle_timeout::buffer sent successfully\n"));
  66     }
  67   catch (const CORBA::Exception& ex)
  68     {
  69       ex._tao_print_exception (
  70         "FTP_Client_Callback::handle_timeout Failed");
  71       return -1;
  72     }
  73   return 0;
  74 }
  75
  76 FTP_Client_Producer::FTP_Client_Producer ()
  77   :TAO_FlowProducer ("Data",CLIENT::instance ()->protocols (),CLIENT::instance ()->format ())
  78 {
  79 }
  80
  81 int
  82 FTP_Client_Producer::set_protocol_object (const char *,
  83                                           TAO_AV_Protocol_Object *object)
  84 {
  85   this->callback_->set_protocol_object (object);
  86   return 0;
  87 }
  88
  89 int
  90 FTP_Client_Producer::get_callback (const char *,
  91                                    TAO_AV_Callback *&callback)
  92 {
  93    ACE_NEW_RETURN (this->callback_,
  94                    FTP_Client_Callback,
  95                    -1);
  96    callback = this->callback_;
  97    return 0;
  98 }
  99
 100 int
 101 Client::parse_args (int argc,
 102                     ACE_TCHAR *argv[])
 103 {
 104   ACE_Get_Opt opts (argc, argv, ACE_TEXT("f:l:a:p:s"));
 105
 106   this->use_sfp_ = 0;
 107   int c;
 108   int p_addr = 0;
 109   int l_addr = 0;
 110   while ((c= opts ()) != -1)
 111     {
 112       switch (c)
 113         {
 114         case 'f':
 115           this->filename_ = ACE_OS::strdup (ACE_TEXT_ALWAYS_CHAR (opts.opt_arg ()));
 116           break;
 117         case 'l':
 118           this->address_ = ACE_OS::strdup (ACE_TEXT_ALWAYS_CHAR (opts.opt_arg ()));
 119           l_addr = 1;
 120           break;
 121         case 'a':
 122           this->peer_addr_ = ACE_OS::strdup (ACE_TEXT_ALWAYS_CHAR (opts.opt_arg ()));
 123           p_addr =1;
 124           break;
 125         case 'p':
 126           this->protocol_ = ACE_OS::strdup (ACE_TEXT_ALWAYS_CHAR (opts.opt_arg ()));
 127           break;
 128         case 's':
 129           this->use_sfp_ = 1;
 130           break;
 131         default:
 132           ACE_DEBUG ((LM_DEBUG,"Unknown option\n"));
 133           return -1;
 134         }
 135     }
 136   if (l_addr == 0)
 137     {
 138       char buf [BUFSIZ];
 139       ACE_OS::hostname (buf,
 140                         BUFSIZ);
 141       address_ = buf;
 142       address_ += ":5000";
 143     }
 144
 145   if (p_addr == 0)
 146     {
 147       char buf [BUFSIZ];
 148       ACE_OS::hostname (buf,
 149                         BUFSIZ);
 150       peer_addr_ = buf;
 151       peer_addr_ += ":5050";
 152     }
 153
 154   return 0;
 155 }
 156
 157 FILE *
 158 Client::file ()
 159 {
 160   return this->fp_;
 161 }
 162
 163 const char*
 164 Client::flowname ()
 165 {
 166   return this->flowname_.c_str();
 167 }
 168
 169 AVStreams::protocolSpec
 170 Client::protocols ()
 171 {
 172   AVStreams::protocolSpec protocols (1);
 173   protocols.length (1);
 174   char buf [BUFSIZ];
 175   ACE_OS::sprintf (buf,"%s=%s",this->protocol_,this->address_.c_str ());
 176   protocols [0] = CORBA::string_dup (buf);
 177   return protocols;
 178 }
 179
 180 const char *
 181 Client::format ()
 182 {
 183   return "UNS:ftp";
 184 }
 185
 186 const char *
 187 Client::address ()
 188
 189 {
 190   return this->address_.c_str ();
 191 }
 192
 193 TAO_StreamCtrl*
 194 Client::streamctrl ()
 195 {
 196   return &this->streamctrl_;
 197 }
 198
 199 Client::Client ()
 200   : fp_ (0),
 201     protocol_ (ACE_OS::strdup ("UDP")),
 202     orb_ (TAO_AV_CORE::instance ()->orb ()),
 203     poa_ (TAO_AV_CORE::instance ()->poa ())
 204 {
 205   endpoint_strategy_.init (TAO_AV_CORE::instance ()->orb (),
 206                            TAO_AV_CORE::instance ()->poa ());
 207 }
 208
 209
 210 int
 211 Client::init (int argc, ACE_TCHAR *argv[])
 212 {
 213   this->argc_ = argc;
 214   this->argv_ = argv;
 215
 216   CORBA::String_var ior;
 217   try
 218     {
 219       PortableServer::POAManager_var mgr
 220         = TAO_AV_CORE::instance ()->poa ()->the_POAManager ();
 221
 222       mgr->activate ();
 223
 224       this->parse_args (this->argc_, this->argv_);
 225
 226       ACE_NEW_RETURN (this->streamendpoint_a_,
 227                       TAO_StreamEndPoint_A, -1);
 228
 229       ACE_NEW_RETURN (this->fep_a_, FTP_Client_Producer, -1);
 230       this->flowname_ = "Data";
 231
 232       sep_a_ = this->streamendpoint_a_->_this();
 233
 234       fep_a_obj_ = this->fep_a_->_this ();
 235
 236       CORBA::String_var s1 = sep_a_->add_fep( fep_a_obj_.in());
 237
 238       ACE_DEBUG ((LM_DEBUG, "(%N,%l) Added flowendpoint named: %C\n", s1.in() ));
 239
 240       this->fp_ = ACE_OS::fopen (this->filename_,"r");
 241
 242       if (this->fp_ != 0)
 243         {
 244           ACE_DEBUG ((LM_DEBUG,"file opened successfully\n"));
 245         }
 246       else
 247         {
 248           ACE_ERROR_RETURN ((LM_ERROR, "ERROR: file %C could not be opened\n",
 249                                        this->filename_), -1);
 250         }
 251     }
 252   catch (const CORBA::Exception& ex)
 253     {
 254       ex._tao_print_exception ("Client::init\n");
 255       return -1;
 256     }
 257   return 0;
 258 }
 259
 260 int
 261 Client::run ()
 262 {
 263   try
 264     {
 265       char flow_protocol_str [BUFSIZ];
 266       if (this->use_sfp_)
 267         ACE_OS::strcpy (flow_protocol_str,"sfp:1.0");
 268       else
 269         ACE_OS::strcpy (flow_protocol_str,"");
 270       AVStreams::streamQoS_var the_qos (new AVStreams::streamQoS);
 271       AVStreams::flowSpec flow_spec (1);
 272       flow_spec.length (1);
 273
 274       ACE_INET_Addr addr (this->address_.c_str ());
 275
 276       TAO_Forward_FlowSpec_Entry entry (this->flowname_.c_str(),
 277                                         "IN",
 278                                         "USER_DEFINED",
 279                                         flow_protocol_str,
 280                                         this->protocol_,
 281                                         &addr);
 282
 283       ACE_INET_Addr peer_addr (this->peer_addr_.c_str ());
 284       entry.set_peer_addr (&peer_addr);
 285
 286       flow_spec [0] = CORBA::string_dup (entry.entry_to_string ());
 287
 288       ACE_High_Res_Timer timer;
 289       ACE_Time_Value elapsed;
 290       timer.start ();
 291
 292
 293       CORBA::Object_var obj_b = this->orb_->string_to_object("corbaname:rir:#Server_StreamEndPoint_b");
 294       AVStreams::StreamEndPoint_B_var sep_b = AVStreams::StreamEndPoint_B::_narrow (obj_b.in());
 295
 296       CORBA::Boolean result =
 297         this->streamctrl_.bind (sep_a_.in(),
 298                                 sep_b.in(),
 299                                 the_qos.inout(),
 300                                 flow_spec);
 301
 302       timer.stop ();
 303       timer.elapsed_time (elapsed);
 304       elapsed.dump ();
 305       if (result == 0)
 306         ACE_ERROR_RETURN ((LM_ERROR,"streamctrl::bind failed\n"),-1);
 307
 308       AVStreams::flowSpec start_spec (1);
 309       this->streamctrl_.start (start_spec);
 310
 311       // Schedule a timer for the for the flow handler.
 312       ACE_Time_Value tv (10000,0);
 313       this->orb_->run (tv);
 314
 315       ACE_DEBUG ((LM_DEBUG, "event loop finished\n"));
 316     }
 317   catch (const CORBA::Exception& ex)
 318     {
 319       ex._tao_print_exception ("Client::run\n");
 320       return -1;
 321     }
 322   return 0;
 323 }
 324
 325 int
 326 ACE_TMAIN (int argc,
 327       ACE_TCHAR *argv[])
 328 {
 329   try
 330     {
 331       CORBA::ORB_var orb = CORBA::ORB_init (argc,
 332                                             argv);
 333       CORBA::Object_var obj
 334         = orb->resolve_initial_references ("RootPOA");
 335
 336       PortableServer::POA_var poa
 337         = PortableServer::POA::_narrow (obj.in ());
 338
 339       TAO_AV_CORE::instance ()->init (orb.in (),
 340                                       poa.in ());
 341
 342
 343       int result = 0;
 344       result = CLIENT::instance ()->init (argc, argv);
 345       if (result < 0)
 346         ACE_ERROR_RETURN ((LM_ERROR,"client::init failed\n"),1);
 347       result = CLIENT::instance ()->run ();
 348
 349       poa->destroy (true, true);
 350
 351       orb->destroy ();
 352
 353       if (result < 0)
 354         ACE_ERROR_RETURN ((LM_ERROR,"client::run failed\n"),1);
 355     }
 356   catch (const CORBA::Exception& ex)
 357
 358   {
 359     ex._tao_print_exception ("Client Failed\n");
 360     return -1;
 361   }
 362
 363   CLIENT::close ();  // Explicitly finalize the Unmanaged_Singleton.
 364
 365   return 0;
 366 }
 367
 368 #if defined (ACE_HAS_EXPLICIT_STATIC_TEMPLATE_MEMBER_INSTANTIATION)
 369 template ACE_Unmanaged_Singleton<Client, ACE_Null_Mutex> *ACE_Unmanaged_Singleton<Client, ACE_Null_Mutex>::singleton_;
 370 #endif /* ACE_HAS_EXPLICIT_STATIC_TEMPLATE_MEMBER_INSTANTIATION */