147 # include <iostream.h>
150 #ifdef HAVE_STD_IOSTREAM
164 #include "CosEventComm.hh"
165 #include "CosEventChannelAdmin.hh"
170 static void usage(
int argc,
char **argv);
172 class Consumer_i :
virtual public POA_CosEventComm::PushConsumer {
176 void push(
const CORBA::Any& data);
190 cout<<
"Push Consumer: push() called. Data : "<< l <<endl;
193 if (i == _disconnect)
201 omni_mutex_lock condition_lock(
mutex);
207 cerr<<
"Push Consumer: push() called. UNEXPECTED TYPE"<<endl;
212 cout <<
"Push Consumer: disconnected." << endl;
220 CORBA::ORB_ptr
orb = CORBA::ORB_init(argc,argv);
224 int sleepInterval =0;
225 const char* channelName =
"EventChannel";
228 while ((c =
getopt(argc,argv,
"hd:s:n:")) != EOF)
232 case 'd': discnum = atoi(
optarg);
235 case 's': sleepInterval = atoi(
optarg);
238 case 'n': channelName =
optarg;
241 case 'h':
usage(argc,argv);
243 default :
usage(argc,argv);
248 #if defined(HAVE_SIGNAL_H) && defined(SIGPIPE)
250 signal(SIGPIPE, SIG_IGN);
254 CosEventChannelAdmin::EventChannel_var channel;
256 const char* action=
"";
258 CORBA::Object_var obj;
260 action=
"resolve initial reference 'RootPOA'";
261 obj=
orb->resolve_initial_references(
"RootPOA");
262 PortableServer::POA_var rootPoa =PortableServer::POA::_narrow(obj);
263 if(CORBA::is_nil(rootPoa))
264 throw CORBA::OBJECT_NOT_EXIST();
266 action=
"activate the RootPOA's POAManager";
267 PortableServer::POAManager_var pman =rootPoa->the_POAManager();
275 action=
"convert URI from command line into object reference";
280 action=
"resolve initial reference 'NameService'";
281 obj=
orb->resolve_initial_references(
"NameService");
282 CosNaming::NamingContext_var rootContext=
283 CosNaming::NamingContext::_narrow(obj);
284 if(CORBA::is_nil(rootContext))
285 throw CORBA::OBJECT_NOT_EXIST();
287 action=
"find EventChannel in NameService";
288 cout << action << endl;
289 obj=rootContext->resolve(
str2name(channelName));
292 action=
"narrow object reference to event channel";
293 channel=CosEventChannelAdmin::EventChannel::_narrow(obj);
294 if(CORBA::is_nil(channel))
296 cerr <<
"Failed to narrow Event Channel reference." << endl;
301 catch(CORBA::ORB::InvalidName& ex) {
302 cerr<<
"Failed to "<<action<<
". ORB::InvalidName"<<endl;
305 catch(CosNaming::NamingContext::InvalidName& ex) {
306 cerr<<
"Failed to "<<action<<
". NamingContext::InvalidName"<<endl;
309 catch(CosNaming::NamingContext::NotFound& ex) {
310 cerr<<
"Failed to "<<action<<
". NamingContext::NotFound"<<endl;
313 catch(CosNaming::NamingContext::CannotProceed& ex) {
314 cerr<<
"Failed to "<<action<<
". NamingContext::CannotProceed"<<endl;
317 catch(CORBA::TRANSIENT& ex) {
318 cerr<<
"Failed to "<<action<<
". TRANSIENT"<<endl;
321 catch(CORBA::OBJECT_NOT_EXIST& ex) {
322 cerr<<
"Failed to "<<action<<
". OBJECT_NOT_EXIST"<<endl;
325 catch(CORBA::SystemException& ex) {
326 cerr<<
"Failed to "<<action<<
".";
327 #if defined(HAVE_OMNIORB4)
328 cerr<<
" "<<ex._name();
329 if(ex.NP_minorString())
330 cerr<<
" ("<<ex.NP_minorString()<<
")";
335 catch(CORBA::Exception& ex) {
336 cerr<<
"Failed to "<<action<<
"."
337 #if defined(HAVE_OMNIORB4)
346 CosEventChannelAdmin::ConsumerAdmin_var consumer_admin;
350 consumer_admin = channel->for_consumers ();
351 if (CORBA::is_nil (consumer_admin))
353 cerr <<
"Event Channel returned nil Consumer Admin!" << endl;
358 catch (CORBA::COMM_FAILURE& ex) {
359 cerr <<
"Caught COMM_FAILURE exception "
360 <<
"obtaining Consumer Admin! Retrying..."
365 cout <<
"Obtained ConsumerAdmin." << endl;
367 omni_mutex_lock condition_lock(
mutex);
371 CosEventChannelAdmin::ProxyPushSupplier_var proxy_supplier;
375 proxy_supplier = consumer_admin->obtain_push_supplier ();
376 if (CORBA::is_nil (proxy_supplier))
378 cerr <<
"Consumer Admin returned nil proxy_supplier!"
384 catch (CORBA::COMM_FAILURE& ex) {
385 cerr <<
"Caught COMM_FAILURE Exception "
386 <<
"obtaining Push Supplier! Retrying..."
391 cout <<
"Obtained ProxyPushSupplier." << endl;
398 proxy_supplier->connect_push_consumer(consumer->_this());
401 catch (CORBA::BAD_PARAM& ex) {
402 cerr <<
"Caught BAD_PARAM Exception connecting Push Consumer!"
406 catch (CosEventChannelAdmin::AlreadyConnected& ex) {
407 cerr <<
"Proxy Push Supplier already connected!"
411 catch (CORBA::COMM_FAILURE& ex) {
412 cerr <<
"Caught COMM_FAILURE exception "
413 <<
"connecting Push Consumer! Retrying..."
418 cout <<
"Connected Push Consumer." << endl;
427 proxy_supplier->disconnect_push_supplier();
430 catch (CORBA::COMM_FAILURE& ex) {
431 cerr <<
"Caught COMM_FAILURE Exception "
432 <<
"disconnecting Push Consumer! Retrying..."
437 cout <<
"Disconnected Push Consumer." << endl;
440 cout <<
"Sleeping " << sleepInterval <<
" seconds." << endl;
441 omni_thread::sleep(sleepInterval);
452 "\nCreate a PushConsumer to receive events from a channel.\n"
453 "syntax: "<<(argc?argv[0]:
"pushcons")<<
" OPTIONS [CHANNEL_URI]\n"
455 "CHANNEL_URI: The event channel may be specified as a URI.\n"
456 " This may be an IOR, or a corbaloc::: or corbaname::: URI.\n"
458 "OPTIONS: DEFAULT:\n"
459 " -d NUM disconnect after receiving NUM events [0 - never disconnect]\n"
460 " -s SECS sleep SECS seconds after disconnecting [0]\n"
461 " -n NAME channel name (if URI is not specified) [\"EventChannel\"]\n"
462 " -h display this help text\n" << endl;
int getopt(int argc, char *argv[], const char *optionS)
CosNaming::Name str2name(const char *namestr)
Converts stringified name to naming service name.
static omni_semaphore connect_cond(0)
int main(int argc, char **argv)
The main process entry point.
static void usage(int argc, char **argv)
void push(const CORBA::Any &data)
void disconnect_push_consumer()
Consumer_i(long disconnect=0)