TAO/orbsvcs/tests/Notify/Discarding/Structured_Consumer.cpp

   1 #include "Notify_Structured_Push_Consumer.h"
   2 #include "goC.h"
   3 #include "Notify_Test_Client.h"
   4
   5 #include "orbsvcs/CosNotifyCommC.h"
   6 #include "orbsvcs/CosNamingC.h"
   7
   8 #include "ace/Get_Opt.h"
   9 #include "ace/OS_NS_unistd.h"
  10 #include "ace/OS_NS_strings.h"
  11
  12 static const ACE_TCHAR *ior = ACE_TEXT ("file://supplier.ior");
  13 static CORBA::Short discard_policy = CosNotification::FifoOrder;
  14 CORBA::Long max_events_per_consumer = 4;
  15 static Notify_Structured_Push_Consumer* consumer_1;
  16
  17 class Consumer_Client : public Notify_Test_Client
  18 {
  19 public:
  20   virtual int parse_args (int argc, ACE_TCHAR *argv[]);
  21 };
  22
  23
  24 int
  25 Consumer_Client::parse_args (int argc, ACE_TCHAR *argv[])
  26 {
  27   ACE_Get_Opt get_opts (argc, argv, ACE_TEXT("k:e:d:"));
  28   int c;
  29
  30   while ((c = get_opts ()) != -1)
  31     switch (c)
  32       {
  33       case 'k':
  34         ior = get_opts.optarg;
  35         break;
  36
  37       case 'd':
  38       {
  39         const ACE_TCHAR *discard = get_opts.optarg;
  40         if (ACE_OS::strcasecmp (discard, ACE_TEXT ("fifo")) == 0)
  41           {
  42             discard_policy = CosNotification::FifoOrder;
  43           }
  44         else if (ACE_OS::strcasecmp (discard, ACE_TEXT ("priority")) == 0)
  45           {
  46             discard_policy = CosNotification::PriorityOrder;
  47           }
  48         else if (ACE_OS::strcasecmp (discard, ACE_TEXT ("lifo")) == 0)
  49           {
  50             discard_policy = CosNotification::LifoOrder;
  51           }
  52         else if (ACE_OS::strcasecmp (discard, ACE_TEXT ("deadline")) == 0)
  53           {
  54             discard_policy = CosNotification::DeadlineOrder;
  55 #if !defined (ACE_HAS_TIMED_MESSAGE_BLOCKS)
  56             ACE_ERROR_RETURN ((LM_ERROR,
  57                                "This order policy requires timed message "
  58                                "blocks.\nPlease #define "
  59                                "ACE_HAS_TIMED_MESSAGE_BLOCKS in your "
  60                                "config.h\n"), -1);
  61 #endif
  62           }
  63         else
  64           {
  65             ACE_ERROR_RETURN ((LM_ERROR,
  66                                "Unknown discard policy: %s\n",
  67                                discard_policy),
  68                               -1);
  69           }
  70         break;
  71       }
  72       default:
  73         ACE_ERROR_RETURN ((LM_ERROR,
  74                            "usage:  %s "
  75                            "-k <ior> "
  76                            "-d <fifo|priority|lifo|deadline> "
  77                            "\n",
  78                            argv [0]),
  79                           -1);
  80       }
  81   // Indicates successful parsing of the command line
  82   return 0;
  83 }
  84
  85
  86 static CosNotifyChannelAdmin::ConsumerAdmin_ptr
  87 create_consumeradmin (CosNotifyChannelAdmin::EventChannel_ptr ec)
  88 {
  89   CosNotifyChannelAdmin::AdminID adminid = 0;
  90   CosNotifyChannelAdmin::ConsumerAdmin_var admin =
  91     ec->new_for_consumers (CosNotifyChannelAdmin::OR_OP,
  92                            adminid);
  93
  94
  95   return CosNotifyChannelAdmin::ConsumerAdmin::_duplicate (admin.in ());
  96 }
  97
  98
  99 static void
 100 create_consumers (CosNotifyChannelAdmin::ConsumerAdmin_ptr admin,
 101                   Notify_Test_Client* client)
 102 {
 103   ACE_DEBUG ((LM_DEBUG, "Max Events per Consumer = %d...\n", max_events_per_consumer));
 104
 105   // startup the consumer
 106   ACE_NEW_THROW_EX (consumer_1,
 107                     Notify_Structured_Push_Consumer (
 108                                           "consumer1",
 109                                           discard_policy,
 110                                           max_events_per_consumer,
 111                                           *client),
 112                     CORBA::NO_MEMORY ());
 113   consumer_1->init (client->root_poa ());
 114
 115   consumer_1->_connect (admin);
 116 }
 117
 118 // ******************************************************************
 119 // Main Section
 120 // ******************************************************************
 121
 122 int ACE_TMAIN (int argc, ACE_TCHAR *argv[])
 123 {
 124   int status = 0;
 125   try
 126     {
 127       Consumer_Client client;
 128
 129       status = client.init (argc, argv);
 130
 131       if (status == 0)
 132         {
 133           CosNotifyChannelAdmin::EventChannel_var ec =
 134             client.create_event_channel ("MyEventChannel", 1);
 135
 136           CORBA::ORB_ptr orb = client.orb ();
 137           CORBA::Object_var object =
 138                               orb->string_to_object (ior);
 139
 140           sig_var sig = sig::_narrow (object.in ());
 141
 142           if (CORBA::is_nil (sig.in ()))
 143             {
 144               ACE_ERROR_RETURN ((LM_ERROR,
 145                                  "Object reference <%s> is nil\n",
 146                                  ior),
 147                                 1);
 148             }
 149
 150           CosNotifyChannelAdmin::ConsumerAdmin_var admin =
 151             create_consumeradmin (ec.in ());
 152
 153           if (!CORBA::is_nil (admin.in ()))
 154             {
 155               create_consumers (admin.in (), &client);
 156               // Tell the supplier to go
 157               sig->go ();
 158
 159               ACE_DEBUG((LM_DEBUG, "Consumer waiting for events...\n"));
 160
 161               client.ORB_run();
 162
 163               ACE_DEBUG((LM_DEBUG, "Consumer done.\n"));
 164               consumer_1->disconnect();
 165
 166               sig->done ();
 167             }
 168         }
 169     }
 170   catch (const CORBA::Exception& e)
 171     {
 172       e._tao_print_exception ("Error: ");
 173       status = 1;
 174     }
 175
 176   return status;
 177 }