TAO/orbsvcs/tests/Notify/Bug_3688_Regression/consumer.cpp

   1 #include "common.h"
   2
   3 #include "ace/Get_Opt.h"
   4 #include "ace/OS_NS_stdlib.h"
   5
   6 int runtime = 5; //sec
   7
   8
   9 namespace CosNotifyCommImpl{
  10   class StructuredPushConsumer: public virtual POA_CosNotifyComm:: StructuredPushConsumer
  11   {
  12   public:
  13     StructuredPushConsumer () : received_ (false) {};
  14     virtual ~ StructuredPushConsumer () {};
  15
  16     void push_structured_event(
  17       const CosNotification::StructuredEvent &event)
  18     {
  19       received_ = true;
  20       std::cout << "Received an event...," << std::endl;
  21       std::cout << "event.header.fixed_header.event_type.domain_name = "
  22         << event.header.fixed_header.event_type.domain_name
  23         << std::endl;
  24       std::cout << "event.header.fixed_header.event_type.type_name = "
  25         << event.header.fixed_header.event_type.type_name
  26         << std::endl;
  27     };
  28
  29     void disconnect_structured_push_consumer()
  30     {};
  31
  32     void offer_change (
  33       const CosNotification::EventTypeSeq &,
  34       const CosNotification::EventTypeSeq &)
  35     {};
  36
  37     bool received_events ()
  38     {
  39       return this->received_;
  40     };
  41
  42   private:
  43     bool received_;
  44   };
  45 }
  46
  47 int parse_args (int argc, ACE_TCHAR *argv[])
  48 {
  49   ACE_Get_Opt get_opts (argc, argv, "t:");
  50   int x;
  51
  52   while ((x = get_opts ()) != -1)
  53     switch (x)
  54   {
  55     case 't':
  56       runtime = ACE_OS::atoi (get_opts.optarg);
  57       break;
  58
  59     default:
  60       ACE_ERROR_RETURN ((LM_ERROR,
  61                          ACE_TEXT ("usage:  %s -t <runtime>\n"),
  62                          argv [0]),
  63                         -1);
  64   }
  65
  66   return 0; // successful parse
  67 }
  68
  69 int ACE_TMAIN (int argc, ACE_TCHAR *argv[])
  70 {
  71   try
  72   {
  73     PortableServer::POAManager_var poa_manager;
  74
  75     CORBA::ORB_var orb = CORBA::ORB_init(argc, argv);
  76
  77     CORBA::Object_var poa_obj = orb->resolve_initial_references("RootPOA");
  78     PortableServer::POA_var root_poa = PortableServer::POA::_narrow(poa_obj.in());
  79
  80     poa_manager = root_poa->the_POAManager();
  81
  82     if (parse_args (argc, argv) != 0)
  83       return 1;
  84
  85     poa_manager->activate();
  86
  87     /*Get event_channel*/
  88     std::cout << "Get event_channel now"  << std::endl;
  89     CosNotifyChannelAdmin::EventChannel_var ec = get_event_channel(orb.in());
  90
  91     //Instanciating the Consumer
  92     CosNotifyComm::StructuredPushConsumer_var spc =
  93       CosNotifyComm::StructuredPushConsumer::_nil();
  94
  95     CosNotifyCommImpl::StructuredPushConsumer *pImpl_spc = new CosNotifyCommImpl::StructuredPushConsumer;
  96     spc = pImpl_spc->_this();
  97
  98     //Obtain a Consumer Admin
  99     CosNotifyChannelAdmin::AdminID adminid = 0;
 100     CosNotifyChannelAdmin::ConsumerAdmin_var ca =
 101       ec->new_for_consumers (CosNotifyChannelAdmin::AND_OP, adminid);
 102
 103     if( ca.in() == CosNotifyChannelAdmin::ConsumerAdmin::_nil() ){
 104       std::cerr << "ca is nil!" << std::endl;
 105       return 1;
 106     }
 107
 108     //Obtain a Proxy Consumer
 109     CosNotifyChannelAdmin::ProxyID proxy_id;
 110     CosNotifyChannelAdmin::ClientType ctype = CosNotifyChannelAdmin::STRUCTURED_EVENT;
 111
 112     CosNotifyChannelAdmin::ProxySupplier_var proxySupplier_obj;
 113     try
 114     {
 115       proxySupplier_obj = ca->obtain_notification_push_supplier(ctype, proxy_id);
 116     }
 117     catch(CosNotifyChannelAdmin::AdminLimitExceeded& err)
 118     {
 119       std::cerr << "CosNotifyChannelAdmin::AdminLimitExceeded Exception!" << std::endl;
 120       throw;
 121     }
 122
 123     CosNotifyChannelAdmin::StructuredProxyPushSupplier_var pps =
 124       CosNotifyChannelAdmin::StructuredProxyPushSupplier::_narrow(proxySupplier_obj.in());
 125
 126
 127     //Attaching a filter to pps
 128     CosNotifyFilter::FilterFactory_var dff =
 129       ec->default_filter_factory();
 130
 131     ACE_ASSERT(!CORBA::is_nil(dff.in()));
 132
 133     CosNotifyFilter::Filter_var filter = dff->create_filter("EXTENDED_TCL");
 134
 135     CosNotification::EventTypeSeq event_types(1);
 136     event_types.length(2);
 137
 138     event_types[0].domain_name = CORBA::string_dup("Test_domain");
 139     event_types[0].type_name = CORBA::string_dup("Test_type_name");
 140     event_types[1].domain_name = CORBA::string_dup("DomainB");
 141     event_types[1].type_name = CORBA::string_dup("TypeB");
 142
 143     CosNotifyFilter::ConstraintExpSeq constraints(1);
 144     constraints.length(1);
 145
 146     constraints[0].event_types = event_types;
 147     constraints[0].constraint_expr = CORBA::string_dup(
 148       "$data == 1 or $data == 2 or $data == 4");
 149
 150     CosNotifyFilter::ConstraintInfoSeq_var cons_info = filter->add_constraints(constraints);
 151
 152     pps->add_filter(filter.in());
 153
 154     std::cout << "Attached a filter to ProxyPushSupplier" << std::endl;
 155     std::cout << "The filter's event_types[0].domain_name = " <<  event_types[0].domain_name << std::endl;
 156     std::cout << "The filter's event_types[0].type_name = " <<  event_types[0].type_name << std::endl;
 157     std::cout << "The filter's event_types[1].domain_name = " <<  event_types[1].domain_name << std::endl;
 158     std::cout << "The filter's event_types[1].type_name = " <<  event_types[1].type_name << std::endl;
 159     std::cout << "The filter's constraint_expr=" << constraints[0].constraint_expr << std::endl;
 160
 161     //Connecting a Supplier to a Proxy Consumer
 162     try
 163     {
 164       pps->connect_structured_push_consumer(spc.in());
 165     }
 166     catch (CosEventChannelAdmin::AlreadyConnected& ac)
 167     {
 168       std::cerr << "CosEventChannelAdmin::AlreadyConnected" << std::endl;
 169       throw;
 170     }
 171     catch (const CORBA::SystemException& se)
 172     {
 173       std::cerr << "System exception occurred during connect: " <<
 174         se << std::endl;
 175       throw;
 176     }
 177
 178
 179     ACE_Time_Value tv (runtime);
 180     orb->run (tv);
 181     ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("Consumer done.\n")));
 182
 183     if (pImpl_spc->received_events ())
 184     {
 185       //Consumer should not receive any events as the user defined constraint takes
 186       //effect.
 187       std::cerr << "Test failed - should not receive test events." << std::endl;
 188       return 1;
 189     }
 190     else
 191     {
 192       std::cout << "Test passed - did not receive test events as expected." << std::endl;
 193     }
 194   }
 195   catch(...)
 196   {
 197     std::cerr << "Consumer: Some exceptions was caught!" << std::endl;
 198     return 1;
 199   }
 200
 201
 202   return 0;
 203 }
 204