TAO/examples/Buffered_AMI/client.cpp

   1
   2 //=============================================================================
   3 /**
   4  *  @file     client.cpp
   5  *
   6  *   This is a client that uses buffered AMI calls.
   7  *
   8  *  @author  Irfan Pyarali
   9  */
  10 //=============================================================================
  11
  12
  13 #include "testS.h"
  14 #include "tao/Messaging/Messaging.h"
  15 #include "tao/AnyTypeCode/Any.h"
  16 #include "tao/AnyTypeCode/TAOA.h"
  17 #include "ace/Get_Opt.h"
  18 #include "ace/Read_Buffer.h"
  19
  20 // Name of file contains ior.
  21 static const ACE_TCHAR *IOR = ACE_TEXT ("file://ior");
  22
  23 // Default iterations.
  24 static CORBA::ULong iterations = 20;
  25
  26 // Default number of invocations to buffer before flushing.
  27 static CORBA::Long message_count = iterations / 4;
  28
  29 // Time interval between invocation (in milli seconds).
  30 static long interval = 1000;
  31
  32 // Flag indicates whether to shutdown remote server or not upon client
  33 // shutdown.
  34 static int shutdown_server = 0;
  35
  36 // AMI call or regular call.
  37 static int invoke_ami_style = 1;
  38
  39 // Setup buffering or not.
  40 static int setup_buffering = 1;
  41
  42 // Flag indicates that all replies have been received
  43 static int received_all_replies = 0;
  44
  45 class Reply_Handler : public POA_AMI_testHandler
  46 {
  47 public:
  48   void method (CORBA::ULong reply_number)
  49     {
  50       ACE_DEBUG ((LM_DEBUG,
  51                   "client: AMI Reply %d @ %T\n",
  52                   reply_number));
  53
  54       // Last reply flips the flag.
  55       if (reply_number == iterations)
  56         received_all_replies = 1;
  57     }
  58
  59   void method_excep (::Messaging::ExceptionHolder *holder)
  60   {
  61     try
  62       {
  63         holder->raise_exception ();
  64       }
  65     catch (const CORBA::SystemException& ex)
  66       {
  67         ex._tao_print_exception ("Reply_Handler::method_excep: ");
  68       }
  69   }
  70
  71   //FUZZ: disable check_for_lack_ACE_OS
  72   void shutdown ()
  73   {
  74   }
  75   //FUZZ: enable check_for_lack_ACE_OS
  76
  77   void shutdown_excep (::Messaging::ExceptionHolder *holder)
  78   {
  79     try
  80       {
  81         holder->raise_exception ();
  82       }
  83     catch (const CORBA::SystemException& ex)
  84       {
  85         ex._tao_print_exception ("Reply_Handler::shutdown_excep: ");
  86       }
  87   }
  88 };
  89
  90 static int
  91 parse_args (int argc, ACE_TCHAR **argv)
  92 {
  93   ACE_Get_Opt get_opts (argc, argv, ACE_TEXT("a:b:k:m:i:t:x"));
  94   int c;
  95
  96   while ((c = get_opts ()) != -1)
  97     switch (c)
  98       {
  99       case 'k':
 100         IOR = get_opts.opt_arg ();
 101         break;
 102
 103       case 'm':
 104         message_count = ACE_OS::atoi (get_opts.opt_arg ());
 105         break;
 106
 107       case 'a':
 108         invoke_ami_style = ACE_OS::atoi (get_opts.opt_arg ());
 109         break;
 110
 111       case 'b':
 112         setup_buffering = ACE_OS::atoi (get_opts.opt_arg ());
 113         break;
 114
 115       case 'i':
 116         iterations = ACE_OS::atoi (get_opts.opt_arg ());
 117         break;
 118
 119       case 't':
 120         interval = ACE_OS::atoi (get_opts.opt_arg ());
 121         break;
 122
 123       case 'x':
 124         shutdown_server = 1;
 125         break;
 126
 127       case '?':
 128       default:
 129         ACE_ERROR_RETURN ((LM_ERROR,
 130                            "usage:  %s "
 131                            "-k IOR "
 132                            "-m message count "
 133                            "-a invoke AMI style [0/1] "
 134                            "-b setup buffering [0/1] "
 135                            "-i iterations "
 136                            "-t interval between calls "
 137                            "-x shutdown server "
 138                            "\n",
 139                            argv [0]),
 140                           -1);
 141       }
 142
 143   if (IOR == 0)
 144     ACE_ERROR_RETURN ((LM_ERROR,
 145                        "Please specify the IOR for the servant\n"), -1);
 146
 147   // Without AMI, replies are immediate.
 148   if (!invoke_ami_style)
 149     received_all_replies = 1;
 150
 151   // Message count must be a multiple of iterations; otherwise we'll
 152   // have some unsent messages left in the buffered queue.  Even
 153   // though we can explicitly flush the queue, I am being lazy and
 154   // forcing the user to give the right numbers.
 155   if ((iterations % message_count) != 0)
 156     {
 157       ACE_ERROR_RETURN ((LM_ERROR,
 158                          "<message_count> must be a multiple <iterations> "
 159                          "or the program should be changed to flush explicitly\n"),
 160                         -1);
 161     }
 162
 163   // Indicates successful parsing of command line.
 164   return 0;
 165 }
 166
 167 void
 168 setup_buffering_constraints (CORBA::ORB_ptr orb)
 169 {
 170   // Obtain PolicyCurrent.
 171   CORBA::Object_var object = orb->resolve_initial_references ("PolicyCurrent");
 172
 173   // Narrow down to correct type.
 174   CORBA::PolicyCurrent_var policy_current =
 175     CORBA::PolicyCurrent::_narrow (object.in ());
 176
 177   // Start off with no constraints.
 178   TAO::BufferingConstraint buffering_constraint;
 179   buffering_constraint.mode = TAO::BUFFER_MESSAGE_COUNT;
 180   buffering_constraint.message_count = message_count;
 181   buffering_constraint.message_bytes = 0;
 182   buffering_constraint.timeout = 0;
 183
 184   // Setup the buffering constraint any.
 185   CORBA::Any buffering_constraint_any;
 186   buffering_constraint_any <<= buffering_constraint;
 187
 188   // Setup the buffering constraint policy list.
 189   CORBA::PolicyList buffering_constraint_policy_list (1);
 190   buffering_constraint_policy_list.length (1);
 191
 192   // Setup the buffering constraint policy.
 193   buffering_constraint_policy_list[0] =
 194     orb->create_policy (TAO::BUFFERING_CONSTRAINT_POLICY_TYPE,
 195                         buffering_constraint_any);
 196
 197   // Setup the constraints (at the ORB level).
 198   policy_current->set_policy_overrides (buffering_constraint_policy_list,
 199                                         CORBA::ADD_OVERRIDE);
 200
 201   // We are done with the policy.
 202   buffering_constraint_policy_list[0]->destroy ();
 203
 204   // Setup the none sync scope policy, i.e., the ORB will buffer AMI
 205   // calls.
 206   Messaging::SyncScope sync_none = Messaging::SYNC_NONE;
 207
 208   // Setup the none sync scope any.
 209   CORBA::Any sync_none_any;
 210   sync_none_any <<= sync_none;
 211
 212   // Setup the none sync scope policy list.
 213   CORBA::PolicyList sync_none_policy_list (1);
 214   sync_none_policy_list.length (1);
 215
 216   // Setup the none sync scope policy.
 217   sync_none_policy_list[0] =
 218     orb->create_policy (Messaging::SYNC_SCOPE_POLICY_TYPE,
 219                         sync_none_any);
 220
 221   // Setup the none sync scope (at the ORB level).
 222   policy_current->set_policy_overrides (sync_none_policy_list,
 223                                         CORBA::ADD_OVERRIDE);
 224
 225   // We are now done with these policies.
 226   sync_none_policy_list[0]->destroy ();
 227 }
 228
 229 int
 230 ACE_TMAIN (int argc, ACE_TCHAR *argv[])
 231 {
 232   try
 233     {
 234       // Initialize the ORB.
 235       CORBA::ORB_var orb =
 236         CORBA::ORB_init (argc, argv);
 237
 238       // Initialize options based on command-line arguments.
 239       int parse_args_result = parse_args (argc, argv);
 240       if (parse_args_result != 0)
 241         return parse_args_result;
 242
 243       CORBA::Object_var base =
 244         orb->resolve_initial_references ("RootPOA");
 245
 246       PortableServer::POA_var root_poa =
 247         PortableServer::POA::_narrow (base.in ());
 248
 249       // Get an object reference from the argument string.
 250       base = orb->string_to_object (IOR);
 251
 252       PortableServer::POAManager_var poa_manager =
 253         root_poa->the_POAManager ();
 254
 255       poa_manager->activate ();
 256
 257       // Try to narrow the object reference to a <test> reference.
 258       test_var test_object = test::_narrow (base.in ());
 259
 260       Reply_Handler reply_handler_servant;
 261       AMI_testHandler_var reply_handler_object = reply_handler_servant._this ();
 262
 263       if (setup_buffering)
 264         {
 265           setup_buffering_constraints (orb.in ());
 266         }
 267
 268       for (CORBA::ULong i = 1; i <= iterations; ++i)
 269         {
 270           ACE_DEBUG ((LM_DEBUG,
 271                       "client: Iteration %d @ %T\n",
 272                       i));
 273
 274           if (invoke_ami_style)
 275             {
 276               // Invoke the AMI method.
 277               test_object->sendc_method (reply_handler_object.in (),
 278                                          i);
 279             }
 280           else
 281             {
 282               CORBA::ULong reply_number = 0;
 283
 284               // Invoke the regular method.
 285               test_object->method (i,
 286                                    reply_number);
 287
 288               ACE_DEBUG ((LM_DEBUG,
 289                           "client: Regular Reply %d @ %T\n",
 290                           reply_number));
 291             }
 292
 293           // Interval between successive calls.
 294           ACE_Time_Value sleep_interval (0,
 295                                          interval * 1000);
 296
 297           orb->run (sleep_interval);
 298         }
 299
 300       // Loop until all replies have been received.
 301       while (!received_all_replies)
 302         {
 303           orb->perform_work ();
 304         }
 305
 306       // Shutdown server.
 307       if (shutdown_server)
 308         {
 309           test_object->shutdown ();
 310         }
 311
 312       root_poa->destroy (true, true);
 313
 314       // Destroy the ORB.  On some platforms, e.g., Win32, the socket
 315       // library is closed at the end of main().  This means that any
 316       // socket calls made after main() fail. Hence if we wait for
 317       // static destructors to flush the queues, it will be too late.
 318       // Therefore, we use explicit destruction here and flush the
 319       // queues before main() ends.
 320       orb->destroy ();
 321     }
 322   catch (const CORBA::Exception& ex)
 323     {
 324       ex._tao_print_exception ("Exception caught:");
 325       return -1;
 326     }
 327
 328   return 0;
 329 }