TAO/orbsvcs/tests/AVStreams/Pluggable_Flow_Protocol/receiver.cpp

   1 #include "receiver.h"
   2 #include "ace/Get_Opt.h"
   3 #include "ace/High_Res_Timer.h"
   4
   5 static FILE *output_file = 0;
   6 // File handle of the file into which received data is written.
   7
   8 static const ACE_TCHAR *output_file_name = ACE_TEXT ("output");
   9 // File name of the file into which received data is written.
  10
  11 int endstream = 0;
  12
  13 int
  14 Receiver_StreamEndPoint::get_callback (const char *,
  15                                        TAO_AV_Callback *&callback)
  16 {
  17   // Return the receiver application callback to the AVStreams for
  18   // future upcalls.
  19   callback = &this->callback_;
  20   return 0;
  21 }
  22
  23 Receiver_Callback::Receiver_Callback ()
  24   : frame_count_ (1)
  25 {
  26 }
  27
  28 int
  29 Receiver_Callback::receive_frame (ACE_Message_Block *frame,
  30                                   TAO_AV_frame_info *,
  31                                   const ACE_Addr &)
  32 {
  33   //
  34   // Upcall from the AVStreams when there is data to be received from
  35   // the sender.
  36   //
  37   ACE_DEBUG ((LM_DEBUG,
  38               "Receiver_Callback::receive_frame for frame %d\n",
  39               this->frame_count_++));
  40
  41   int frame_size = BUFSIZ;
  42
  43   // Write the received data to the file.
  44   size_t result =
  45     ACE_OS::fwrite (frame->rd_ptr (),
  46                     frame_size,
  47                     1,
  48                     output_file);
  49
  50   if (result == frame->length ())
  51     ACE_ERROR_RETURN ((LM_ERROR,
  52                        "Receiver_Callback::fwrite failed\n"),
  53                       -1);
  54   frame->rd_ptr (frame_size);
  55
  56   ACE_hrtime_t stamp;
  57   ACE_OS::memcpy (&stamp, frame->rd_ptr (), sizeof (stamp));
  58
  59   ACE_UINT64 usec = stamp;
  60   ACE_UINT32 val_1 = ACE_CU64_TO_CU32 (usec);
  61
  62   ACE_DEBUG ((LM_DEBUG,
  63               "Time Stamp %u\n",
  64               val_1));
  65
  66   frame->reset ();
  67   return 0;
  68 }
  69
  70 int
  71 Receiver_Callback::handle_destroy ()
  72 {
  73   // Called when the distributer requests the stream to be shutdown.
  74   ACE_DEBUG ((LM_DEBUG,
  75               "Receiver_Callback::end_stream\n"));
  76
  77   endstream = 1;
  78   return 0;
  79 }
  80
  81 Receiver::Receiver ()
  82   : mmdevice_ (0)
  83 {
  84 }
  85
  86 Receiver::~Receiver ()
  87 {
  88 }
  89
  90 int
  91 Receiver::init (int,
  92                 ACE_TCHAR *[])
  93 {
  94   // Initialize the endpoint strategy with the orb and poa.
  95   int result =
  96     this->reactive_strategy_.init (TAO_AV_CORE::instance ()->orb (),
  97                                    TAO_AV_CORE::instance ()->poa ());
  98   if (result != 0)
  99     return result;
 100
 101   // Register the receiver mmdevice object with the ORB
 102   ACE_NEW_RETURN (this->mmdevice_,
 103                   TAO_MMDevice (&this->reactive_strategy_),
 104                   -1);
 105
 106   // Servant Reference Counting to manage lifetime
 107   PortableServer::ServantBase_var safe_mmdevice =
 108     this->mmdevice_;
 109
 110   CORBA::Object_var mmdevice =
 111     this->mmdevice_->_this ();
 112
 113   // Register the mmdevice with the naming service.
 114   CosNaming::Name name (1);
 115   name.length (1);
 116   name [0].id =
 117     CORBA::string_dup ("Receiver");
 118
 119   // Initialize the naming services
 120   if (this->naming_client_.init (TAO_AV_CORE::instance ()->orb ()) != 0)
 121     ACE_ERROR_RETURN ((LM_ERROR,
 122                        "Unable to initialize "
 123                        "the TAO_Naming_Client\n"),
 124                       -1);
 125
 126   // Register the receiver object with the naming server.
 127   this->naming_client_->rebind (name,
 128                                 mmdevice.in ());
 129
 130   return 0;
 131 }
 132
 133 int
 134 parse_args (int argc,
 135             ACE_TCHAR *argv[])
 136 {
 137   // Parse the command line arguments
 138   ACE_Get_Opt opts (argc,
 139                     argv,
 140                     "f:");
 141
 142   int c;
 143   while ((c = opts ()) != -1)
 144     {
 145       switch (c)
 146         {
 147         case 'f':
 148           output_file_name = opts.opt_arg ();
 149           break;
 150         default:
 151           ACE_ERROR_RETURN ((LM_ERROR,
 152                              "Usage: receiver -f filename"),
 153                             -1);
 154         }
 155     }
 156
 157   return 0;
 158 }
 159
 160 int
 161 ACE_TMAIN (int argc,
 162       ACE_TCHAR *argv[])
 163 {
 164   try
 165     {
 166       ACE_High_Res_Timer::global_scale_factor ();
 167
 168       // Initialize the ORB first.
 169       CORBA::ORB_var orb =
 170         CORBA::ORB_init (argc, argv);
 171
 172       CORBA::Object_var obj
 173         = orb->resolve_initial_references ("RootPOA");
 174
 175       // Get the POA_var object from Object_var.
 176       PortableServer::POA_var root_poa =
 177         PortableServer::POA::_narrow (obj.in ());
 178
 179       PortableServer::POAManager_var mgr
 180         = root_poa->the_POAManager ();
 181
 182       mgr->activate ();
 183
 184       // Initialize the AVStreams components.
 185       TAO_AV_CORE::instance ()->init (orb.in (),
 186                                       root_poa.in ());
 187
 188       int result =
 189         parse_args (argc,
 190                     argv);
 191
 192       if (result == -1)
 193         return -1;
 194
 195       // Make sure we have a valid <output_file>
 196       output_file = ACE_OS::fopen (output_file_name,
 197                                    "w");
 198       if (output_file == 0)
 199         ACE_ERROR_RETURN ((LM_DEBUG,
 200                            "Cannot open output file %s\n",
 201                            output_file_name),
 202                           -1);
 203
 204       else
 205         ACE_DEBUG ((LM_DEBUG,
 206                     "File Opened Successfully\n"));
 207
 208       Receiver receiver;
 209       result =
 210         receiver.init (argc,
 211                        argv);
 212
 213       if (result != 0)
 214         return result;
 215
 216       while (!endstream)
 217         {
 218           orb->perform_work ();
 219         }
 220
 221       // Hack for now....
 222       ACE_OS::sleep (1);
 223
 224       orb->destroy ();
 225     }
 226   catch (const CORBA::Exception& ex)
 227     {
 228       ex._tao_print_exception ("receiver::init");
 229       return -1;
 230     }
 231
 232   ACE_OS::fclose (output_file);
 233
 234   return 0;
 235 }