diff options
Diffstat (limited to 'TAO')
-rw-r--r-- | TAO/examples/Event_Comm/Consumer_Handler.cpp | 127 | ||||
-rw-r--r-- | TAO/examples/Event_Comm/Consumer_Handler.h | 58 | ||||
-rw-r--r-- | TAO/examples/Event_Comm/Consumer_Input_Handler.cpp | 145 | ||||
-rw-r--r-- | TAO/examples/Event_Comm/Consumer_Input_Handler.h | 67 | ||||
-rw-r--r-- | TAO/examples/Event_Comm/Event_Comm.idl | 93 | ||||
-rw-r--r-- | TAO/examples/Event_Comm/Event_Comm_i.cpp | 374 | ||||
-rw-r--r-- | TAO/examples/Event_Comm/Event_Comm_i.h | 99 | ||||
-rw-r--r-- | TAO/examples/Event_Comm/Makefile | 643 | ||||
-rw-r--r-- | TAO/examples/Event_Comm/Notifier_Handler.cpp | 57 | ||||
-rw-r--r-- | TAO/examples/Event_Comm/Notifier_Handler.h | 47 | ||||
-rw-r--r-- | TAO/examples/Event_Comm/Supplier_Input_Handler.cpp | 136 | ||||
-rw-r--r-- | TAO/examples/Event_Comm/Supplier_Input_Handler.h | 68 | ||||
-rw-r--r-- | TAO/examples/Event_Comm/consumer.cpp | 101 | ||||
-rw-r--r-- | TAO/examples/Event_Comm/notifier.cpp | 94 | ||||
-rw-r--r-- | TAO/examples/Event_Comm/supplier.cpp | 90 |
15 files changed, 2199 insertions, 0 deletions
diff --git a/TAO/examples/Event_Comm/Consumer_Handler.cpp b/TAO/examples/Event_Comm/Consumer_Handler.cpp new file mode 100644 index 00000000000..9b9b3bb043d --- /dev/null +++ b/TAO/examples/Event_Comm/Consumer_Handler.cpp @@ -0,0 +1,127 @@ +// $Id$ + +#include "Consumer_Handler.h" + +ACE_RCSID(Consumer, Consumer_Handler, "$Id$") + +#if defined (ACE_HAS_ORBIX) && (ACE_HAS_ORBIX != 0) + +#if defined (ACE_HAS_MT_ORBIX) && (ACE_HAS_MT_ORBIX != 0) +typedef ACE_MT_CORBA_Handler CORBA_HANDLER; +#else +typedef ACE_ST_CORBA_Handler CORBA_HANDLER; +#endif /* ACE_HAS_MT_ORBIX */ + +int +Consumer_Handler::handle_close (ACE_HANDLE, ACE_Reactor_Mask) +{ +// ACE_ST_CORBA_Handler::remove_service (Event_Comm_Consumer_IMPL); + + if (this->receiver_ != 0) + { + ACE_DEBUG ((LM_DEBUG, + "closing down Consumer_Handler\n")); + CORBA_HANDLER::instance ()->deactivate_service (Event_Comm_Consumer_IMPL, + this->receiver_->_marker ()); + CORBA::release (this->receiver_); + this->receiver_ = 0; + CORBA::release (this->notifier_); + this->notifier_ = 0; + // *Must* be allocated dynamically in order to delete this! + delete this; + } + return 0; +} + +Consumer_Handler::Consumer_Handler (int argc, char *argv[]) + : notifier_ (0), + receiver_ (0) +{ + const char *server_name = + Event_Comm_Consumer_IMPL; + char buf[BUFSIZ]; + char *receiver_marker = buf; + char *filtering_criteria; + char *host; + char *notifier_marker; + char *service_location = argv[0]; + + // First see if we have any environment variables. + filtering_criteria = ACE_OS::getenv ("FILTERING_CRITERIA"); + host = ACE_OS::getenv ("HOST"); + notifier_marker = ACE_OS::getenv ("NOTIFIER_MARKER"); + + // Then override these variables with command-line arguments. + filtering_criteria = argc > 1 ? argv[1] : ""; + host = argc > 2 ? argv[2] : "tango.cs"; + notifier_marker = argc > 3 ? argv[3] : "notifier:" Event_Comm_Notifier_IR; + + CORBA::Orbix.setDiagnostics (0); + + utsname name; + + // Make the marker name be the "/hostname/processid" + ACE_OS::uname (&name); + sprintf (buf, "/%s/%d", name.nodename, ACE_OS::getpid ()); + + CORBA_HANDLER::instance ()->activate_service (Event_Comm_Consumer_IMPL, + receiver_marker, + service_location); + + // Create the receiver object. + ACE_NEW (this->receiver_, + TIE_Event_Comm_Consumer (Consumer_i) + (new Consumer_i)); + + this->receiver_->_marker (receiver_marker); + + ACE_ASSERT (this->receiver_); + + TRY + { + // Get a binding to the notifier. + this->notifier_ = Event_Comm::Notifier::_bind (notifier_marker, host, IT_X); + + if (this->notifier_ != CORBA::OBJECT_NIL) + // Subscribe ourselves with the notifier's broker. + this->notifier_->subscribe (this->receiver_, + filtering_criteria, IT_X); + } + CATCHANY + { + cerr << "Unexpected exception " << IT_X << endl; + ACE_OS::exit (1); + } + ENDTRY; + // Print out context. + + receiver_marker = (char *) this->receiver_->_marker (); + CORBA::BOA::activationMode mode = CORBA::Orbix.myActivationMode (); + ACE_DEBUG ((LM_DEBUG, + "starting up a %spersistent server in mode %d with marker name %s\n", + mode == CORBA::BOA::persistentActivationMode ? "" : "non-", + mode, + receiver_marker)); +} + +Event_Comm::Consumer * +Consumer_Handler::receiver (void) +{ + return this->receiver_; +} + +Event_Comm::Notifier * +Consumer_Handler::notifier (void) +{ + return this->notifier_; +} + +// Destroy a Receiver target object. + +Consumer_Handler::~Consumer_Handler (void) +{ + this->handle_close (-1, + ACE_Event_Handler::ALL_EVENTS_MASK); +} + +#endif /* ACE_HAS_ORBIX */ diff --git a/TAO/examples/Event_Comm/Consumer_Handler.h b/TAO/examples/Event_Comm/Consumer_Handler.h new file mode 100644 index 00000000000..352c8231bf1 --- /dev/null +++ b/TAO/examples/Event_Comm/Consumer_Handler.h @@ -0,0 +1,58 @@ +/* -*- C++ -*- */ +// $Id$ + +// ============================================================================ +// +// = LIBRARY +// EventComm +// +// = FILENAME +// Consumer_Handler.h +// +// = DESCRIPTION +// Subclass of Corba_Handler that sets up the Consumer handler +// for use with the ACE ACE_Reactor. +// +// = AUTHOR +// Douglas C. Schmidt (schmidt@cs.wustl.edu) +// +// ============================================================================ + +#if !defined (_CONSUMER_HANDLER_H) +#define _CONSUMER_HANDLER_H + +#include "ace/CORBA_Handler.h" +#include "Event_Comm_i.h" + +class Consumer_Handler +{ + // = TITLE + // Subclass of Corba_Handler that sets up the Notification + // Receiver handler for use with the ACE_Reactor. + // + // = DESCRIPTION + // Note that this class doesn't inherit from ACE_ST_CORBA_Handler + // (unlike the Supplier's Notifier_Handler class). Instead, it + // uses an alternative interface that can be called directly. +public: + Consumer_Handler (int argc, char *argv[]); + + Event_Comm::Consumer *receiver (void); + Event_Comm::Notifier *notifier (void); + + virtual int handle_close (ACE_HANDLE = ACE_INVALID_HANDLE, + ACE_Reactor_Mask = ACE_Event_Handler::NULL_MASK); + // Close down the handler. + +private: + ~Consumer_Handler (void); + // Ensure dynamic allocation. + + Event_Comm::Consumer *receiver_; + // Pointer to an IDL <Consumer> proxy object. + + Event_Comm::Notifier *notifier_; + // Pointer to an IDL <Notifier> proxy object. +}; + +#endif /* _CONSUMER_HANDLER_H */ diff --git a/TAO/examples/Event_Comm/Consumer_Input_Handler.cpp b/TAO/examples/Event_Comm/Consumer_Input_Handler.cpp new file mode 100644 index 00000000000..3b55cc9c13d --- /dev/null +++ b/TAO/examples/Event_Comm/Consumer_Input_Handler.cpp @@ -0,0 +1,145 @@ +// $Id$ + +#include "Consumer_Input_Handler.h" +#include "Consumer_Handler.h" + +ACE_RCSID(Consumer, Consumer_Input_Handler, "$Id$") + +Consumer_Input_Handler::~Consumer_Input_Handler (void) +{ + ACE_DEBUG ((LM_DEBUG, + "closing down Input_Handler::~Input_Handler\n")); + this->handle_close (); +} + +int +Consumer_Input_Handler::consumer_initiated_shutdown (void) +{ + return this->consumer_initiated_shutdown_; +} + +void +Consumer_Input_Handler::consumer_initiated_shutdown (int c) +{ + this->consumer_initiated_shutdown_ = c; +} + +ACE_HANDLE +Consumer_Input_Handler::get_handle (void) const +{ + return this->handle_; +} + +int +Consumer_Input_Handler::handle_close (ACE_HANDLE, ACE_Reactor_Mask) +{ + ACE_DEBUG ((LM_DEBUG, + "closing down Consumer::Input_Handler\n")); + + Event_Comm::Consumer *receiver = this->receiver_handler_->receiver (); + Event_Comm::Notifier *notifier = this->receiver_handler_->notifier (); + + if (this->consumer_initiated_shutdown ()) + { + // Only try to unsubscribe if the Consumer initiated the + // shutdown. Otherwise, the Supplier initiated it and it has + // probably gone away by now! + TRY + { + // Gracefully shutdown the Receiver by removing it from the + // Notifier's internal map. + + notifier->unsubscribe (receiver, "", IT_X); + } + CATCHANY + { + cerr << IT_X << endl; + } + ENDTRY; + } + + // Don't execute a callback here otherwise we'll recurse + // indefinitely! + if (ACE_Reactor::instance ()->remove_handler + (this, + ACE_Event_Handler::READ_MASK | ACE_Event_Handler::DONT_CALL) == -1) + ACE_ERROR ((LM_ERROR, + "%p\n", + "remove_handler")); + + // *Must* be allocated dyanmically! + ::operator delete (this); + return 0; +} + +Consumer_Input_Handler::Input_Handler (Consumer_Handler *ch, + ACE_HANDLE handle) + : receiver_handler_ (ch), + handle_ (handle), + consumer_initiated_shutdown_ (0) +{ + if (ACE_Reactor::instance ()->register_handler + (this, + ACE_Event_Handler::READ_MASK) == -1) + ACE_ERROR ((LM_ERROR, + "Consumer_Input_Handler::Input_Handler\n")); +} + +int +Consumer_Input_Handler::handle_input (ACE_HANDLE h) +{ + char buf[BUFSIZ]; + + // Read up to BUFSIZ worth of data from ACE_HANDLE h. + ssize_t n = ACE_OS::read (h, buf, sizeof buf - 1); + + if (n > 0) + { + // Null terminate the buffer, replacing the '\n' with '\0'. + if (buf[n - 1] == '\n' || buf[n - 1] == EOF) + buf[n - 1] = '\0'; + else + buf[n] = '\0'; + ACE_DEBUG ((LM_DEBUG, + "notifying for event %s\n", + buf)); + } + else + { + ACE_OS::strcpy (buf, "quit"); + ACE_DEBUG ((LM_DEBUG, + "shutting down Input_Handler\n")); + } + + Event_Comm::Notifier *notifier = + this->receiver_handler_->notifier (); + + ACE_ASSERT (notifier != 0); + + if (ACE_OS::strcmp (buf, "quit") == 0) + { + // Consumer wants to shutdown. + this->consumer_initiated_shutdown (1); + + // Tell the main event loop to shutdown. + ACE_Reactor::end_event_loop(); + } + else + { + TRY + { + Event_Comm::Event event; + + event.tag_ = ACE_OS::strdup (buf); + + notifier->push (event, IT_X); + } + CATCHANY + { + cerr << "Unexpected exception " << IT_X << endl; + } ENDTRY; + } + + /* NOTREACHED */ + return 0; +} diff --git a/TAO/examples/Event_Comm/Consumer_Input_Handler.h b/TAO/examples/Event_Comm/Consumer_Input_Handler.h new file mode 100644 index 00000000000..2ae7069dfe1 --- /dev/null +++ b/TAO/examples/Event_Comm/Consumer_Input_Handler.h @@ -0,0 +1,67 @@ +/* -*- C++ -*- */ +// $Id$ + +// ============================================================================ +// +// = LIBRARY +// EventComm +// +// = FILENAME +// Consumer_Input_Handler.h +// +// = AUTHOR +// Douglas C. Schmidt (schmidt@cs.wustl.edu) +// +// ============================================================================ + +#if !defined (CONSUMER_INPUT_HANDLER_H) +#define CONSUMER_INPUT_HANDLER_ + +#include "ace/Service_Config.h" + +// Forward declaration. +class Consumer_Handler; + +class Input_Handler : public ACE_Service_Object +{ + // = TITLE + // Handles input events generated from a keyboard. + // + // = DESCRIPTION + // This subclass <ACE_Service_Object> receives "unsubscribes" + // from the <Notifier> when input is received from the keyboard. +public: + Consumer_Input_Handler (Consumer_Handler *, + ACE_HANDLE h = 0); + + virtual int handle_input (ACE_HANDLE); + // Dispatch the callback when events occur. + + virtual int handle_close (ACE_HANDLE = ACE_INVALID_HANDLE, + ACE_Reactor_Mask = ACE_Event_Handler::NULL_MASK); + // Close down the handler. + + int consumer_initiated_shutdown (void); + // Report whether the Consumer initiated the shutdown. + + void consumer_initiated_shutdown (int); + // Indicate that the Consumer initiated the shutdown. + +private: + ~Consumer_Input_Handler (void); + // Ensure dynamic allocation. + + virtual ACE_HANDLE get_handle (void) const; + + ACE_HANDLE handle_; + // ACE_HANDLE where the input comes from. + + Consumer_Handler *receiver_handler_; + // Pointer to the <Consumer_Handler> that receives notifications + // from the <Event_Comm::Notifier>. + + int consumer_initiated_shutdown_; + // Keep track of whether the Consumer initiated the shutdown. +}; + +#endif /* CONSUMER_INPUT_HANDLER_H */ diff --git a/TAO/examples/Event_Comm/Event_Comm.idl b/TAO/examples/Event_Comm/Event_Comm.idl new file mode 100644 index 00000000000..1f7bdcc6441 --- /dev/null +++ b/TAO/examples/Event_Comm/Event_Comm.idl @@ -0,0 +1,93 @@ +/* -*- C++ -*- */ +// $Id$ + +// ============================================================================ +// +// = LIBRARY +// EventComm +// +// = FILENAME +// Event_Comm.idl +// +// = DESCRIPTION +// The CORBA IDL module for distributed event notification. +// +// = AUTHOR +// Douglas C. Schmidt (schmidt@cs.wustl.edu) +// +// ============================================================================ + +#if !defined (_EVENT_COMM_IDL) +#define _EVENT_COMM_IDL + +module Event_Comm +{ + // = TITLE + // The CORBA IDL module for distributed event notification. + + struct Event + { + // = TITLE + // Defines the interface for an event <Event>. + // + // = DESCRIPTION + // This is the type passed by the Notifier to the Consumer. + // Since it contains an <any>, it can hold any type. Naturally, + // the consumer must understand how to interpret this! + + string tag_; + // Tag for the event. This is used by the <Notifier> to compare + // with the <Consumer>s' filtering criteria. + + any value_; + // An event can contain anything. + + Object object_ref_; + // Object reference for callbacks. + }; + + interface Consumer + // = TITLE + // Defines the interface for a <Consumer> of events. + // Note that all operations are <oneway> to avoid blocking. + { + void push (in Event event); + // Inform the <Consumer> that <event> has occurred. + + void disconnect (in string reason); + // Disconnect the <Consumer> from the <Notifier>, + // giving it the <reason>. + }; + + interface Notifier + { + // = TITLE + // Defines the interface for a <Notifier> of events. + + // = The following operations are intended for Suppliers. + + void disconnect (in string reason); + // Disconnect all the receivers, giving them the <reason>. + + void push (in Event event); + // Send the <event> to all the consumers who have subscribed and + // who match the filtering criteria. + + // = The following operations are intended for Consumers. + + void subscribe (in Consumer Consumer, + in string filtering_criteria); + // Subscribe the <Consumer> to receive events that match the + // regular expresssion <filtering_criteria> applied by the + // <Notifier>. If <filtering_criteria> is "" then all events are + // matched. + + void unsubscribe (in Consumer Consumer, + in string filtering_criteria); + // Unsubscribe the <Consumer> that matches the filtering criteria. + // If <filtering_criteria> is "" then all <Consumers> with the + // matching object reference are removed. + }; +}; + +#endif /* _EVENT_COMM_IDL */ diff --git a/TAO/examples/Event_Comm/Event_Comm_i.cpp b/TAO/examples/Event_Comm/Event_Comm_i.cpp new file mode 100644 index 00000000000..f181a15d460 --- /dev/null +++ b/TAO/examples/Event_Comm/Event_Comm_i.cpp @@ -0,0 +1,374 @@ +// $Id$ + +#include "Event_Comm_i.h" + +ACE_RCSID(Event_Comm, Event_Comm_i, "$Id$") + +class Consumer_Entry +{ + // = TITLE + // Keeps track of context information associated with + // a <Event_Comm::Consumer> entry. +public: + Consumer_Entry (Event_Comm::Consumer *notification_receiver, + const char *filtering_criteria); + ~Consumer_Entry (void); + + // = Set/get filtering criteria. + void criteria (const char *criteria); + const char *criteria (void); + + // = Set/get Event_Comm::Consumer object reference. + Event_Comm::Consumer *receiver (void); + void receiver (Event_Comm::Consumer *); + + // = Set/get the compiled regular expression buffer. + const char *regexp (void); + void regexp (char *); + +private: + const char *filtering_criteria_; + // String containing the filtering criteria. + + char *compiled_regexp_; + // Compiled representation of the regular expression (see + // regexpr(3g)). + + Event_Comm::Consumer *receiver_; + // Object reference for the Event_Comm::Consumer. +}; + +// = Set/get filtering criteria. + +void +Consumer_Entry::criteria (const char *criteria) +{ + ACE_OS::free (ACE_MALLOC_T (this->filtering_criteria_)); + this->filtering_criteria_ = ACE_OS::strdup (criteria); +} + +const char * +Consumer_Entry::criteria (void) +{ + return this->filtering_criteria_; +} + +// = Set/get Event_Comm::Consumer object reference. + +Event_Comm::Consumer * +Consumer_Entry::receiver (void) +{ + return this->receiver_; +} + +void +Consumer_Entry::receiver (Event_Comm::Consumer *receiver) +{ + this->receiver_ = receiver; +} + +const char * +Consumer_Entry::regexp (void) +{ + return this->compiled_regexp_; +} + +void +Consumer_Entry::regexp (char *regexp) +{ + ACE_OS::free (ACE_MALLOC_T (this->compiled_regexp_)); + this->compiled_regexp_ = regexp; +} + +Consumer_Entry::Consumer_Entry (Event_Comm::Consumer *receiver, + const char *filtering_criteria) + : receiver_ (receiver), + filtering_criteria_ (0), + compiled_regexp_ (0) +{ + char *compile_buffer = 0; + + this->criteria (filtering_criteria); + ACE_ASSERT (this->criteria ()); + + // Check for wildcard case first. + if (ACE_OS::strcmp (filtering_criteria, "") == 0) + compile_buffer = ACE_OS::strdup (""); + else + // Compile the regular expression (the 0's cause ACE_OS::compile + // to allocate space). + compile_buffer = ACE_OS::compile (filtering_criteria, 0, 0); + + // Should throw an exception here! + ACE_ASSERT (compile_buffer != 0); + + this->regexp (compile_buffer); + ACE_ASSERT (this->regexp ()); + + // Increment the reference count since we are keeping a copy of + // this... + this->receiver_->_duplicate (this->receiver_); +} + +Consumer_Entry::~Consumer_Entry (void) +{ + ACE_OS::free (this->filtering_criteria_); + ACE_OS::free (this->compiled_regexp_); + // Decrement the object reference count. + CORBA::release (this->receiver_); +} + +Notifier_i::Notifier_i (size_t size) + : map_ (size) +{ +} + +// Add a new receiver to the table, being careful to check for +// duplicate entries. A receiver is considered a duplicate under the +// following circumstances: +// 1. It has the same marker name and the same filtering criteria +// 2. It has the same marker name and its filtering criteria is "" (the wild card). + +void +Notifier_i::subscribe (Event_Comm::Consumer *receiver_ref, + const char *filtering_criteria, + CORBA::Environment &IT_env) +{ + ACE_DEBUG ((LM_DEBUG, + "in Notifier_i::subscribe for %s with filtering criteria \"%s\"\n", + receiver_ref->_marker (), + filtering_criteria)); + ACE_SString key (receiver_ref->_marker ()); + MAP_ITERATOR mi (this->map_); + + // Try to locate an entry using its marker name (which should be + // unique across the system). If we don't find the entry, or if the + // filtering criteria is different that is good news since we + // currently don't allow duplicates... In particular, if @@ Should + // duplicates be allowed? + + for (MAP_ENTRY *me = 0; mi.next (me) != 0; mi.advance ()) + { + Consumer_Entry *nr_entry = me->int_id_; + + // Check for a duplicate entry. + if (key == me->ext_id_ + && (ACE_OS::strcmp (filtering_criteria, "") == 0 + || ACE_OS::strcmp (filtering_criteria, nr_entry->criteria ()) == 0)) + { + // Inform the caller that the + // Event_Comm::Consumer * is already being + // used. + + errno = EADDRINUSE; + ACE_ERROR ((LM_ERROR, + "duplicate entry for receiver %s with criteria \"%s\"", + receiver_ref->_marker (), + filtering_criteria)); + // Raise exception here??? + return; + } + } + + // If we get this far then we didn't find a duplicate, so add the + // new entry! + Consumer_Entry *nr_entry; + ACE_NEW (nr_entry, + Consumer_Entry (receiver_ref, + filtering_criteria)); + // Try to add new <Consumer_Entry> to the map. + else if (this->map_.bind (key, nr_entry) == -1) + { + // Prevent memory leaks. + delete nr_entry; + // Raise exception here... + ACE_ERROR ((LM_ERROR, + "%p\n", + "bind failed")); + } +} + +// Remove a receiver from the table. + +void +Notifier_i::unsubscribe (Event_Comm::Consumer *receiver_ref, + const char *filtering_criteria, + CORBA::Environment &IT_env) +{ + ACE_DEBUG ((LM_DEBUG, + "in Notifier_i::unsubscribe for %s\n", + receiver_ref->_marker ())); + Consumer_Entry *nr_entry = 0; + ACE_SString key; + MAP_ITERATOR mi (this->map_); + int found = 0; + + // Don't make a copy since we are deleting... + key.rep ((char *) receiver_ref->_marker ()); + + // Locate <Consumer_Entry> and free up resources. @@ + // Note, we don't properly handle deallocation of KEYS! + + for (MAP_ENTRY *me = 0; mi.next (me) != 0; mi.advance ()) + { + if (key == me->ext_id_ + && (ACE_OS::strcmp (filtering_criteria, "") == 0 + || ACE_OS::strcmp (filtering_criteria, nr_entry->criteria ()) == 0)) + { + ACE_DEBUG ((LM_DEBUG, + "removed entry %s with criteria \"%s\"\n", + receiver_ref->_marker (), + filtering_criteria)); + found = 1; + // @@ This is a hack, we need a better approach! + if (this->map_.unbind (key, nr_entry) == -1) + ACE_ERROR ((LM_ERROR, + "unbind failed for %s\n", + receiver_ref->_marker ())); + else + delete nr_entry; + } + } + + if (found == 0) + ACE_ERROR ((LM_ERROR, + "entry %s with criteria \"%s\" not found\n", + receiver_ref->_marker (), + filtering_criteria)); +} + +// Disconnect all the receivers, giving them the <reason>. + +void +Notifier_i::disconnect (const char *reason, + CORBA::Environment &IT_env) +{ + ACE_DEBUG ((LM_DEBUG, + "in Notifier_i::send_disconnect = %s\n", + reason)); + MAP_ITERATOR mi (this->map_); + int count = 0; + + // Notify all the receivers, taking into account the filtering criteria. + + for (MAP_ENTRY *me = 0; mi.next (me) != 0; mi.advance ()) + { + Event_Comm::Consumer *receiver_ref = me->int_id_->receiver (); + ACE_ASSERT (receiver_ref->_marker () != 0); + ACE_DEBUG ((LM_DEBUG, + "disconnecting client %s\n", + receiver_ref->_marker ())); + TRY + { + receiver_ref->disconnect (reason, IT_X); + } + CATCHANY + { + cerr << "Unexpected exception " << IT_X << endl; + } + ENDTRY; + delete me->int_id_; + delete me->ext_id_.rep (); + count++; + } + + this->map_.close (); + if (count == 1) + ACE_DEBUG ((LM_DEBUG, + "there was 1 receiver\n")); + else + ACE_DEBUG ((LM_DEBUG, + "there were %d receivers\n", + count)); +} + +// Notify all receivers whose filtering criteria match the event. + +void +Notifier_i::push (const Event_Comm::Event &event, + CORBA::Environment &IT_env) +{ + ACE_DEBUG ((LM_DEBUG, + "in Notifier_i::send_notification = %s\n", + notification.tag_)); + MAP_ITERATOR mi (this->map_); + int count = 0; + + // Notify all the receivers. + // @@ Later on we need to consider the filtering_criteria! + + for (MAP_ENTRY *me = 0; mi.next (me) != 0; mi.advance ()) + { + Event_Comm::Consumer *receiver_ref = me->int_id_->receiver (); + ACE_ASSERT (receiver_ref->_marker () != 0); + const char *regexp = me->int_id_->regexp (); + const char *criteria = me->int_id_->criteria (); + ACE_ASSERT (regexp); + ACE_ASSERT (criteria); + + // Do a regular expression comparison to determine matching. + if (ACE_OS::strcmp ("", criteria) == 0 // Everything matches the wildcard. +// || ACE_OS::strcmp (notification.tag_, regexp) == 0) + || ACE_OS::step (notification.tag_, regexp) != 0) + { + ACE_DEBUG ((LM_DEBUG, + "string %s matched regexp \"%s\" for client %s\n", + notification.tag_, me->int_id_->criteria (), + receiver_ref->_marker ())); + TRY + { + receiver_ref->receive_notification (notification, IT_X); + } + CATCHANY + { + cerr << "Unexpected exception " << IT_X << endl; + continue; + } + ENDTRY; + count++; + } + } + + if (count == 1) + ACE_DEBUG ((LM_DEBUG, + "there was 1 receiver\n")); + else + ACE_DEBUG ((LM_DEBUG, + "there were %d receivers\n", + count)); +} + +Consumer_i::Consumer_i (void) +{ +} + +Consumer_i::~Consumer_i (void) +{ +} + +// Inform the <Event_Comm::Consumer> that <event> has +// occurred. + +void +Consumer_i::push (const Event_Comm::Event &event, + CORBA::Environment &IT_env) +{ + const char *tmpstr = notification.tag_; + + ACE_DEBUG ((LM_DEBUG, + "**** got notification = %s\n", + tmpstr)); +} + +// Disconnect the <Event_Comm::Consumer> from the +// <Event_Comm::Notifier>. + +void +Consumer_i::disconnect (const char *reason, + CORBA::Environment &IT_env) +{ + ACE_DEBUG ((LM_DEBUG, + "**** got disconnected due to %s\n", + reason)); + ACE_Reactor::end_event_loop (); +} diff --git a/TAO/examples/Event_Comm/Event_Comm_i.h b/TAO/examples/Event_Comm/Event_Comm_i.h new file mode 100644 index 00000000000..a68d3fb9c31 --- /dev/null +++ b/TAO/examples/Event_Comm/Event_Comm_i.h @@ -0,0 +1,99 @@ +/* -*- C++ -*- */ +// $Id$ + +// ============================================================================ +// +// = LIBRARY +// EventComm +// +// = FILENAME +// Event_Comm_i.h +// +// = DESCRIPTION +// Class interface for the implementation of the distributed +// event notification mechanism. +// +// = AUTHOR +// Douglas C. Schmidt (schmidt@cs.wustl.edu) +// +// ============================================================================ + +#if !defined (_EVENT_COMM_I_H) +#define _EVENT_COMM_I_H + +#include "ace/Map_Manager.h" +#include "ace/Synch.h" +#include "ace/SString.h" + +class Consumer_i +{ + // = TITLE + // Defines the implementation class for event <Consumers>. +public: + // = Initialization and termination methods. + Consumer_i (void); + // Constructor. + + ~Consumer_i (void); + // Destructor. + + virtual void push (const Event_Comm::Event &event, + CORBA::Environment &IT_env); + // Pass the <event> to the <Consumer>. + + virtual void disconnect (const char *reason, + CORBA::Environment &IT_env); + // Disconnect the <Consumer> from the <Notifier>, + // giving it the <reason>. +}; + +// Forward reference. +class Consumer_Entry; + +class Notifier_i +{ + // = TITLE + // Defines the implementation class for event <Notifiers>. +public: + enum + { + DEFAULT_SIZE = 1024 // Default max number of Event_Comm::Consumers. + }; + + Notifier_i (size_t size_hint = Notifier_i::DEFAULT_SIZE); + // Initialize a Notifier_i object with the specified size hint. + + void disconnect (const char *reason, + CORBA::Environment &IT_env); + // Disconnect all the receivers, giving them the <reason>. + + void push (const Event_Comm::Event &event, + CORBA::Environment &IT_env); + // Send the <event> to all the consumers who have subscribed + // and who match the filtering criteria. + + void subscribe (Event_Comm::Consumer *consumer, + const char *filtering_criteria, + CORBA::Environment &IT_env); + // Subscribe the <Consumer> to receive events that match + // <filtering_criteria> applied by the <Notifier>. + + void unsubscribe (Event_Comm::Consumer *consumer, + const char *filtering_criteria, + CORBA::Environment &IT_env); + // Unsubscribe the <Consumer>. + +private: + typedef ACE_Map_Manager <ACE_SString, Consumer_Entry *, ACE_Null_Mutex> + MAP_MANAGER; + typedef ACE_Map_Iterator <ACE_SString, Consumer_Entry *, ACE_Null_Mutex> + MAP_ITERATOR; + typedef ACE_Map_Entry <ACE_SString, Consumer_Entry *> + MAP_ENTRY; + + MAP_MANAGER map_; + // Table that maps a <Event_Comm::Consumer *> to a <Consumer_Entry + // *>. +}; + +#endif /* _EVENT_COMM_I_H */ diff --git a/TAO/examples/Event_Comm/Makefile b/TAO/examples/Event_Comm/Makefile new file mode 100644 index 00000000000..140721bcc95 --- /dev/null +++ b/TAO/examples/Event_Comm/Makefile @@ -0,0 +1,643 @@ +#---------------------------------------------------------------------------- +# +# $Id$ +# +#---------------------------------------------------------------------------- + +#---------------------------------------------------------------------------- +# Local macros +#---------------------------------------------------------------------------- + +ifndef TAO_ROOT + TAO_ROOT = $(ACE_ROOT)/TAO +endif # ! TAO_ROOT + +LDLIBS = -lorbsvcs -lTAO +PROG_SRCS = Consumer_Handler.cpp \ + Consumer_Input_Handler.cpp \ + Event_Comm_i.cpp \ + Notifier_Handler.cpp \ + Supplier_Input_Handler.cpp \ + consumer.cpp \ + notifier.cpp \ + supplier.cpp + +IDL_SRC = Event_CommS.cpp Event_CommC.cpp + +LSRC = $(IDL_SRC) $(PROG_SRCS) + +CONSUMER_OBJS = Consumer_Handler.o \ + Consumer_Input_Handler.o \ + Event_Comm_i.o + +SUPPLIER_OBJS = Supplier_Input_Handler.o \ + Event_Comm_i.o + +NOTIFIER_OBJS = Notifier_Handler.o \ + +BIN = consumer \ + notifier \ + supplier +BUILD = $(BIN) +VLDLIBS = $(LDLIBS:%=%$(VAR)) +VBIN = $(BIN:%=%$(VAR)) + +#---------------------------------------------------------------------------- +# Include macros and targets +#---------------------------------------------------------------------------- + +include $(ACE_ROOT)/include/makeinclude/wrapper_macros.GNU +include $(ACE_ROOT)/include/makeinclude/macros.GNU +include $(TAO_ROOT)/rules.tao.GNU +include $(ACE_ROOT)/include/makeinclude/rules.common.GNU +include $(ACE_ROOT)/include/makeinclude/rules.nonested.GNU +include $(ACE_ROOT)/include/makeinclude/rules.local.GNU +include $(TAO_ROOT)/taoconfig.mk + +#---------------------------------------------------------------------------- +# Local targets +#---------------------------------------------------------------------------- + +LDFLAGS += -L$(TAO_ROOT)/orbsvcs/orbsvcs -L$(TAO_ROOT)/tao -L$(TAO_ROOT)/orbsvcs/Naming_Service +CPPFLAGS += -I$(TAO_ROOT)/orbsvcs + +.PRECIOUS: cubitC.cpp cubitS.cpp cubitC.h cubitS.h +#$(IDL_SRC): cubit.idl +# $(TAO_ROOT)/TAO_IDL/tao_idl cubit.idl + +supplier: $(addprefix $(VDIR),$(SUPPLIER_OBJS)) + $(LINK.cc) $(LDFLAGS) -o $@ $^ $(VLDLIBS) $(POSTLINK) + +consumer: $(addprefix $(VDIR),$(CONSUMER_OBJS)) + $(LINK.cc) $(LDFLAGS) -o $@ $^ $(VLDLIBS) $(POSTLINK) + +notifier: $(addprefix $(VDIR),$(NOTIFIER_OBJS)) + $(LINK.cc) $(LDFLAGS) -o $@ $^ $(VLDLIBS) $(POSTLINK) + +realclean: clean + -/bin/rm -rf cubitC.* cubitS.* cubitS_T.* + +# DO NOT DELETE THIS LINE -- g++dep uses it. +# DO NOT PUT ANYTHING AFTER THIS LINE, IT WILL GO AWAY. + +.obj/Consumer_Handler.o .obj/Consumer_Handler.so .shobj/Consumer_Handler.o .shobj/Consumer_Handler.so: Consumer_Handler.cpp Consumer_Handler.h \ + $(ACE_ROOT)/ace/CORBA_Handler.h \ + $(ACE_ROOT)/ace/Service_Config.h \ + $(ACE_ROOT)/ace/Service_Object.h \ + $(ACE_ROOT)/ace/Shared_Object.h \ + $(ACE_ROOT)/ace/ACE.h \ + $(ACE_ROOT)/ace/OS.h \ + $(ACE_ROOT)/ace/inc_user_config.h \ + $(ACE_ROOT)/ace/config.h \ + $(ACE_ROOT)/ace/config-g++-common.h \ + $(ACE_ROOT)/ace/streams.h \ + $(ACE_ROOT)/ace/Basic_Types.h \ + $(ACE_ROOT)/ace/Basic_Types.i \ + $(ACE_ROOT)/ace/OS.i \ + $(ACE_ROOT)/ace/Trace.h \ + $(ACE_ROOT)/ace/Log_Msg.h \ + $(ACE_ROOT)/ace/Log_Record.h \ + $(ACE_ROOT)/ace/ACE.i \ + $(ACE_ROOT)/ace/Log_Priority.h \ + $(ACE_ROOT)/ace/Log_Record.i \ + $(ACE_ROOT)/ace/Shared_Object.i \ + $(ACE_ROOT)/ace/Event_Handler.h \ + $(ACE_ROOT)/ace/Event_Handler.i \ + $(ACE_ROOT)/ace/Service_Object.i \ + $(ACE_ROOT)/ace/Signal.h \ + $(ACE_ROOT)/ace/Synch.h \ + $(ACE_ROOT)/ace/SV_Semaphore_Complex.h \ + $(ACE_ROOT)/ace/SV_Semaphore_Simple.h \ + $(ACE_ROOT)/ace/SV_Semaphore_Simple.i \ + $(ACE_ROOT)/ace/SV_Semaphore_Complex.i \ + $(ACE_ROOT)/ace/Synch.i \ + $(ACE_ROOT)/ace/Synch_T.h \ + $(ACE_ROOT)/ace/Synch_T.i \ + $(ACE_ROOT)/ace/Thread.h \ + $(ACE_ROOT)/ace/Thread.i \ + $(ACE_ROOT)/ace/Atomic_Op.i \ + $(ACE_ROOT)/ace/Synch_T.cpp \ + $(ACE_ROOT)/ace/Containers.h \ + $(ACE_ROOT)/ace/Containers.i \ + $(ACE_ROOT)/ace/Containers_T.h \ + $(ACE_ROOT)/ace/Containers_T.i \ + $(ACE_ROOT)/ace/Containers_T.cpp \ + $(ACE_ROOT)/ace/Malloc.h \ + $(ACE_ROOT)/ace/Malloc_Base.h \ + $(ACE_ROOT)/ace/Malloc.i \ + $(ACE_ROOT)/ace/Malloc_T.h \ + $(ACE_ROOT)/ace/Free_List.h \ + $(ACE_ROOT)/ace/Free_List.i \ + $(ACE_ROOT)/ace/Free_List.cpp \ + $(ACE_ROOT)/ace/Malloc_T.i \ + $(ACE_ROOT)/ace/Malloc_T.cpp \ + $(ACE_ROOT)/ace/Memory_Pool.h \ + $(ACE_ROOT)/ace/Mem_Map.h \ + $(ACE_ROOT)/ace/Mem_Map.i \ + $(ACE_ROOT)/ace/Memory_Pool.i \ + $(ACE_ROOT)/ace/Signal.i \ + $(ACE_ROOT)/ace/Object_Manager.h \ + $(ACE_ROOT)/ace/Object_Manager.i \ + $(ACE_ROOT)/ace/Managed_Object.h \ + $(ACE_ROOT)/ace/Managed_Object.i \ + $(ACE_ROOT)/ace/Managed_Object.cpp \ + $(ACE_ROOT)/ace/SString.h \ + $(ACE_ROOT)/ace/SString.i \ + $(ACE_ROOT)/ace/Service_Config.i \ + $(ACE_ROOT)/ace/Reactor.h \ + $(ACE_ROOT)/ace/Handle_Set.h \ + $(ACE_ROOT)/ace/Handle_Set.i \ + $(ACE_ROOT)/ace/Timer_Queue.h \ + $(ACE_ROOT)/ace/Timer_Queue_T.h \ + $(ACE_ROOT)/ace/Timer_Queue_T.i \ + $(ACE_ROOT)/ace/Timer_Queue_T.cpp \ + $(ACE_ROOT)/ace/Reactor.i \ + $(ACE_ROOT)/ace/Reactor_Impl.h \ + $(ACE_ROOT)/ace/Svc_Conf_Tokens.h \ + $(ACE_ROOT)/ace/Pipe.h \ + $(ACE_ROOT)/ace/Pipe.i Event_Comm_i.h \ + $(ACE_ROOT)/ace/Map_Manager.h \ + $(ACE_ROOT)/ace/Map_Manager.i \ + $(ACE_ROOT)/ace/Map_Manager.cpp +.obj/Consumer_Input_Handler.o .obj/Consumer_Input_Handler.so .shobj/Consumer_Input_Handler.o .shobj/Consumer_Input_Handler.so: Consumer_Input_Handler.cpp \ + Consumer_Input_Handler.h \ + $(ACE_ROOT)/ace/Service_Config.h \ + $(ACE_ROOT)/ace/Service_Object.h \ + $(ACE_ROOT)/ace/Shared_Object.h \ + $(ACE_ROOT)/ace/ACE.h \ + $(ACE_ROOT)/ace/OS.h \ + $(ACE_ROOT)/ace/inc_user_config.h \ + $(ACE_ROOT)/ace/config.h \ + $(ACE_ROOT)/ace/config-g++-common.h \ + $(ACE_ROOT)/ace/streams.h \ + $(ACE_ROOT)/ace/Basic_Types.h \ + $(ACE_ROOT)/ace/Basic_Types.i \ + $(ACE_ROOT)/ace/OS.i \ + $(ACE_ROOT)/ace/Trace.h \ + $(ACE_ROOT)/ace/Log_Msg.h \ + $(ACE_ROOT)/ace/Log_Record.h \ + $(ACE_ROOT)/ace/ACE.i \ + $(ACE_ROOT)/ace/Log_Priority.h \ + $(ACE_ROOT)/ace/Log_Record.i \ + $(ACE_ROOT)/ace/Shared_Object.i \ + $(ACE_ROOT)/ace/Event_Handler.h \ + $(ACE_ROOT)/ace/Event_Handler.i \ + $(ACE_ROOT)/ace/Service_Object.i \ + $(ACE_ROOT)/ace/Signal.h \ + $(ACE_ROOT)/ace/Synch.h \ + $(ACE_ROOT)/ace/SV_Semaphore_Complex.h \ + $(ACE_ROOT)/ace/SV_Semaphore_Simple.h \ + $(ACE_ROOT)/ace/SV_Semaphore_Simple.i \ + $(ACE_ROOT)/ace/SV_Semaphore_Complex.i \ + $(ACE_ROOT)/ace/Synch.i \ + $(ACE_ROOT)/ace/Synch_T.h \ + $(ACE_ROOT)/ace/Synch_T.i \ + $(ACE_ROOT)/ace/Thread.h \ + $(ACE_ROOT)/ace/Thread.i \ + $(ACE_ROOT)/ace/Atomic_Op.i \ + $(ACE_ROOT)/ace/Synch_T.cpp \ + $(ACE_ROOT)/ace/Containers.h \ + $(ACE_ROOT)/ace/Containers.i \ + $(ACE_ROOT)/ace/Containers_T.h \ + $(ACE_ROOT)/ace/Containers_T.i \ + $(ACE_ROOT)/ace/Containers_T.cpp \ + $(ACE_ROOT)/ace/Malloc.h \ + $(ACE_ROOT)/ace/Malloc_Base.h \ + $(ACE_ROOT)/ace/Malloc.i \ + $(ACE_ROOT)/ace/Malloc_T.h \ + $(ACE_ROOT)/ace/Free_List.h \ + $(ACE_ROOT)/ace/Free_List.i \ + $(ACE_ROOT)/ace/Free_List.cpp \ + $(ACE_ROOT)/ace/Malloc_T.i \ + $(ACE_ROOT)/ace/Malloc_T.cpp \ + $(ACE_ROOT)/ace/Memory_Pool.h \ + $(ACE_ROOT)/ace/Mem_Map.h \ + $(ACE_ROOT)/ace/Mem_Map.i \ + $(ACE_ROOT)/ace/Memory_Pool.i \ + $(ACE_ROOT)/ace/Signal.i \ + $(ACE_ROOT)/ace/Object_Manager.h \ + $(ACE_ROOT)/ace/Object_Manager.i \ + $(ACE_ROOT)/ace/Managed_Object.h \ + $(ACE_ROOT)/ace/Managed_Object.i \ + $(ACE_ROOT)/ace/Managed_Object.cpp \ + $(ACE_ROOT)/ace/SString.h \ + $(ACE_ROOT)/ace/SString.i \ + $(ACE_ROOT)/ace/Service_Config.i \ + $(ACE_ROOT)/ace/Reactor.h \ + $(ACE_ROOT)/ace/Handle_Set.h \ + $(ACE_ROOT)/ace/Handle_Set.i \ + $(ACE_ROOT)/ace/Timer_Queue.h \ + $(ACE_ROOT)/ace/Timer_Queue_T.h \ + $(ACE_ROOT)/ace/Timer_Queue_T.i \ + $(ACE_ROOT)/ace/Timer_Queue_T.cpp \ + $(ACE_ROOT)/ace/Reactor.i \ + $(ACE_ROOT)/ace/Reactor_Impl.h \ + $(ACE_ROOT)/ace/Svc_Conf_Tokens.h \ + Consumer_Handler.h \ + $(ACE_ROOT)/ace/CORBA_Handler.h \ + $(ACE_ROOT)/ace/Pipe.h \ + $(ACE_ROOT)/ace/Pipe.i Event_Comm_i.h \ + $(ACE_ROOT)/ace/Map_Manager.h \ + $(ACE_ROOT)/ace/Map_Manager.i \ + $(ACE_ROOT)/ace/Map_Manager.cpp +.obj/Event_Comm_i.o .obj/Event_Comm_i.so .shobj/Event_Comm_i.o .shobj/Event_Comm_i.so: Event_Comm_i.cpp +.obj/Notifier_Handler.o .obj/Notifier_Handler.so .shobj/Notifier_Handler.o .shobj/Notifier_Handler.so: Notifier_Handler.cpp Notifier_Handler.h \ + $(ACE_ROOT)/ace/CORBA_Handler.h \ + $(ACE_ROOT)/ace/Service_Config.h \ + $(ACE_ROOT)/ace/Service_Object.h \ + $(ACE_ROOT)/ace/Shared_Object.h \ + $(ACE_ROOT)/ace/ACE.h \ + $(ACE_ROOT)/ace/OS.h \ + $(ACE_ROOT)/ace/inc_user_config.h \ + $(ACE_ROOT)/ace/config.h \ + $(ACE_ROOT)/ace/config-g++-common.h \ + $(ACE_ROOT)/ace/streams.h \ + $(ACE_ROOT)/ace/Basic_Types.h \ + $(ACE_ROOT)/ace/Basic_Types.i \ + $(ACE_ROOT)/ace/OS.i \ + $(ACE_ROOT)/ace/Trace.h \ + $(ACE_ROOT)/ace/Log_Msg.h \ + $(ACE_ROOT)/ace/Log_Record.h \ + $(ACE_ROOT)/ace/ACE.i \ + $(ACE_ROOT)/ace/Log_Priority.h \ + $(ACE_ROOT)/ace/Log_Record.i \ + $(ACE_ROOT)/ace/Shared_Object.i \ + $(ACE_ROOT)/ace/Event_Handler.h \ + $(ACE_ROOT)/ace/Event_Handler.i \ + $(ACE_ROOT)/ace/Service_Object.i \ + $(ACE_ROOT)/ace/Signal.h \ + $(ACE_ROOT)/ace/Synch.h \ + $(ACE_ROOT)/ace/SV_Semaphore_Complex.h \ + $(ACE_ROOT)/ace/SV_Semaphore_Simple.h \ + $(ACE_ROOT)/ace/SV_Semaphore_Simple.i \ + $(ACE_ROOT)/ace/SV_Semaphore_Complex.i \ + $(ACE_ROOT)/ace/Synch.i \ + $(ACE_ROOT)/ace/Synch_T.h \ + $(ACE_ROOT)/ace/Synch_T.i \ + $(ACE_ROOT)/ace/Thread.h \ + $(ACE_ROOT)/ace/Thread.i \ + $(ACE_ROOT)/ace/Atomic_Op.i \ + $(ACE_ROOT)/ace/Synch_T.cpp \ + $(ACE_ROOT)/ace/Containers.h \ + $(ACE_ROOT)/ace/Containers.i \ + $(ACE_ROOT)/ace/Containers_T.h \ + $(ACE_ROOT)/ace/Containers_T.i \ + $(ACE_ROOT)/ace/Containers_T.cpp \ + $(ACE_ROOT)/ace/Malloc.h \ + $(ACE_ROOT)/ace/Malloc_Base.h \ + $(ACE_ROOT)/ace/Malloc.i \ + $(ACE_ROOT)/ace/Malloc_T.h \ + $(ACE_ROOT)/ace/Free_List.h \ + $(ACE_ROOT)/ace/Free_List.i \ + $(ACE_ROOT)/ace/Free_List.cpp \ + $(ACE_ROOT)/ace/Malloc_T.i \ + $(ACE_ROOT)/ace/Malloc_T.cpp \ + $(ACE_ROOT)/ace/Memory_Pool.h \ + $(ACE_ROOT)/ace/Mem_Map.h \ + $(ACE_ROOT)/ace/Mem_Map.i \ + $(ACE_ROOT)/ace/Memory_Pool.i \ + $(ACE_ROOT)/ace/Signal.i \ + $(ACE_ROOT)/ace/Object_Manager.h \ + $(ACE_ROOT)/ace/Object_Manager.i \ + $(ACE_ROOT)/ace/Managed_Object.h \ + $(ACE_ROOT)/ace/Managed_Object.i \ + $(ACE_ROOT)/ace/Managed_Object.cpp \ + $(ACE_ROOT)/ace/SString.h \ + $(ACE_ROOT)/ace/SString.i \ + $(ACE_ROOT)/ace/Service_Config.i \ + $(ACE_ROOT)/ace/Reactor.h \ + $(ACE_ROOT)/ace/Handle_Set.h \ + $(ACE_ROOT)/ace/Handle_Set.i \ + $(ACE_ROOT)/ace/Timer_Queue.h \ + $(ACE_ROOT)/ace/Timer_Queue_T.h \ + $(ACE_ROOT)/ace/Timer_Queue_T.i \ + $(ACE_ROOT)/ace/Timer_Queue_T.cpp \ + $(ACE_ROOT)/ace/Reactor.i \ + $(ACE_ROOT)/ace/Reactor_Impl.h \ + $(ACE_ROOT)/ace/Svc_Conf_Tokens.h \ + $(ACE_ROOT)/ace/Pipe.h \ + $(ACE_ROOT)/ace/Pipe.i Event_Comm_i.h \ + $(ACE_ROOT)/ace/Map_Manager.h \ + $(ACE_ROOT)/ace/Map_Manager.i \ + $(ACE_ROOT)/ace/Map_Manager.cpp +.obj/Supplier_Input_Handler.o .obj/Supplier_Input_Handler.so .shobj/Supplier_Input_Handler.o .shobj/Supplier_Input_Handler.so: Supplier_Input_Handler.cpp Event_Comm_i.h \ + $(ACE_ROOT)/ace/Map_Manager.h \ + $(ACE_ROOT)/ace/ACE.h \ + $(ACE_ROOT)/ace/OS.h \ + $(ACE_ROOT)/ace/inc_user_config.h \ + $(ACE_ROOT)/ace/config.h \ + $(ACE_ROOT)/ace/config-g++-common.h \ + $(ACE_ROOT)/ace/streams.h \ + $(ACE_ROOT)/ace/Basic_Types.h \ + $(ACE_ROOT)/ace/Basic_Types.i \ + $(ACE_ROOT)/ace/OS.i \ + $(ACE_ROOT)/ace/Trace.h \ + $(ACE_ROOT)/ace/Log_Msg.h \ + $(ACE_ROOT)/ace/Log_Record.h \ + $(ACE_ROOT)/ace/ACE.i \ + $(ACE_ROOT)/ace/Log_Priority.h \ + $(ACE_ROOT)/ace/Log_Record.i \ + $(ACE_ROOT)/ace/Map_Manager.i \ + $(ACE_ROOT)/ace/Map_Manager.cpp \ + $(ACE_ROOT)/ace/Synch.h \ + $(ACE_ROOT)/ace/SV_Semaphore_Complex.h \ + $(ACE_ROOT)/ace/SV_Semaphore_Simple.h \ + $(ACE_ROOT)/ace/SV_Semaphore_Simple.i \ + $(ACE_ROOT)/ace/SV_Semaphore_Complex.i \ + $(ACE_ROOT)/ace/Synch.i \ + $(ACE_ROOT)/ace/Synch_T.h \ + $(ACE_ROOT)/ace/Event_Handler.h \ + $(ACE_ROOT)/ace/Event_Handler.i \ + $(ACE_ROOT)/ace/Synch_T.i \ + $(ACE_ROOT)/ace/Thread.h \ + $(ACE_ROOT)/ace/Thread.i \ + $(ACE_ROOT)/ace/Atomic_Op.i \ + $(ACE_ROOT)/ace/Synch_T.cpp \ + $(ACE_ROOT)/ace/Malloc.h \ + $(ACE_ROOT)/ace/Malloc_Base.h \ + $(ACE_ROOT)/ace/Malloc.i \ + $(ACE_ROOT)/ace/Malloc_T.h \ + $(ACE_ROOT)/ace/Free_List.h \ + $(ACE_ROOT)/ace/Free_List.i \ + $(ACE_ROOT)/ace/Free_List.cpp \ + $(ACE_ROOT)/ace/Malloc_T.i \ + $(ACE_ROOT)/ace/Malloc_T.cpp \ + $(ACE_ROOT)/ace/Memory_Pool.h \ + $(ACE_ROOT)/ace/Signal.h \ + $(ACE_ROOT)/ace/Containers.h \ + $(ACE_ROOT)/ace/Containers.i \ + $(ACE_ROOT)/ace/Containers_T.h \ + $(ACE_ROOT)/ace/Containers_T.i \ + $(ACE_ROOT)/ace/Containers_T.cpp \ + $(ACE_ROOT)/ace/Signal.i \ + $(ACE_ROOT)/ace/Object_Manager.h \ + $(ACE_ROOT)/ace/Object_Manager.i \ + $(ACE_ROOT)/ace/Managed_Object.h \ + $(ACE_ROOT)/ace/Managed_Object.i \ + $(ACE_ROOT)/ace/Managed_Object.cpp \ + $(ACE_ROOT)/ace/Mem_Map.h \ + $(ACE_ROOT)/ace/Mem_Map.i \ + $(ACE_ROOT)/ace/Memory_Pool.i \ + $(ACE_ROOT)/ace/Service_Config.h \ + $(ACE_ROOT)/ace/Service_Object.h \ + $(ACE_ROOT)/ace/Shared_Object.h \ + $(ACE_ROOT)/ace/Shared_Object.i \ + $(ACE_ROOT)/ace/Service_Object.i \ + $(ACE_ROOT)/ace/SString.h \ + $(ACE_ROOT)/ace/SString.i \ + $(ACE_ROOT)/ace/Service_Config.i \ + $(ACE_ROOT)/ace/Reactor.h \ + $(ACE_ROOT)/ace/Handle_Set.h \ + $(ACE_ROOT)/ace/Handle_Set.i \ + $(ACE_ROOT)/ace/Timer_Queue.h \ + $(ACE_ROOT)/ace/Timer_Queue_T.h \ + $(ACE_ROOT)/ace/Timer_Queue_T.i \ + $(ACE_ROOT)/ace/Timer_Queue_T.cpp \ + $(ACE_ROOT)/ace/Reactor.i \ + $(ACE_ROOT)/ace/Reactor_Impl.h \ + $(ACE_ROOT)/ace/Svc_Conf_Tokens.h \ + Notifier_Handler.h \ + $(ACE_ROOT)/ace/CORBA_Handler.h \ + $(ACE_ROOT)/ace/Pipe.h \ + $(ACE_ROOT)/ace/Pipe.i Supplier_Input_Handler.h +.obj/consumer.o .obj/consumer.so .shobj/consumer.o .shobj/consumer.so: consumer.cpp Consumer_Handler.h \ + $(ACE_ROOT)/ace/CORBA_Handler.h \ + $(ACE_ROOT)/ace/Service_Config.h \ + $(ACE_ROOT)/ace/Service_Object.h \ + $(ACE_ROOT)/ace/Shared_Object.h \ + $(ACE_ROOT)/ace/ACE.h \ + $(ACE_ROOT)/ace/OS.h \ + $(ACE_ROOT)/ace/inc_user_config.h \ + $(ACE_ROOT)/ace/config.h \ + $(ACE_ROOT)/ace/config-g++-common.h \ + $(ACE_ROOT)/ace/streams.h \ + $(ACE_ROOT)/ace/Basic_Types.h \ + $(ACE_ROOT)/ace/Basic_Types.i \ + $(ACE_ROOT)/ace/OS.i \ + $(ACE_ROOT)/ace/Trace.h \ + $(ACE_ROOT)/ace/Log_Msg.h \ + $(ACE_ROOT)/ace/Log_Record.h \ + $(ACE_ROOT)/ace/ACE.i \ + $(ACE_ROOT)/ace/Log_Priority.h \ + $(ACE_ROOT)/ace/Log_Record.i \ + $(ACE_ROOT)/ace/Shared_Object.i \ + $(ACE_ROOT)/ace/Event_Handler.h \ + $(ACE_ROOT)/ace/Event_Handler.i \ + $(ACE_ROOT)/ace/Service_Object.i \ + $(ACE_ROOT)/ace/Signal.h \ + $(ACE_ROOT)/ace/Synch.h \ + $(ACE_ROOT)/ace/SV_Semaphore_Complex.h \ + $(ACE_ROOT)/ace/SV_Semaphore_Simple.h \ + $(ACE_ROOT)/ace/SV_Semaphore_Simple.i \ + $(ACE_ROOT)/ace/SV_Semaphore_Complex.i \ + $(ACE_ROOT)/ace/Synch.i \ + $(ACE_ROOT)/ace/Synch_T.h \ + $(ACE_ROOT)/ace/Synch_T.i \ + $(ACE_ROOT)/ace/Thread.h \ + $(ACE_ROOT)/ace/Thread.i \ + $(ACE_ROOT)/ace/Atomic_Op.i \ + $(ACE_ROOT)/ace/Synch_T.cpp \ + $(ACE_ROOT)/ace/Containers.h \ + $(ACE_ROOT)/ace/Containers.i \ + $(ACE_ROOT)/ace/Containers_T.h \ + $(ACE_ROOT)/ace/Containers_T.i \ + $(ACE_ROOT)/ace/Containers_T.cpp \ + $(ACE_ROOT)/ace/Malloc.h \ + $(ACE_ROOT)/ace/Malloc_Base.h \ + $(ACE_ROOT)/ace/Malloc.i \ + $(ACE_ROOT)/ace/Malloc_T.h \ + $(ACE_ROOT)/ace/Free_List.h \ + $(ACE_ROOT)/ace/Free_List.i \ + $(ACE_ROOT)/ace/Free_List.cpp \ + $(ACE_ROOT)/ace/Malloc_T.i \ + $(ACE_ROOT)/ace/Malloc_T.cpp \ + $(ACE_ROOT)/ace/Memory_Pool.h \ + $(ACE_ROOT)/ace/Mem_Map.h \ + $(ACE_ROOT)/ace/Mem_Map.i \ + $(ACE_ROOT)/ace/Memory_Pool.i \ + $(ACE_ROOT)/ace/Signal.i \ + $(ACE_ROOT)/ace/Object_Manager.h \ + $(ACE_ROOT)/ace/Object_Manager.i \ + $(ACE_ROOT)/ace/Managed_Object.h \ + $(ACE_ROOT)/ace/Managed_Object.i \ + $(ACE_ROOT)/ace/Managed_Object.cpp \ + $(ACE_ROOT)/ace/SString.h \ + $(ACE_ROOT)/ace/SString.i \ + $(ACE_ROOT)/ace/Service_Config.i \ + $(ACE_ROOT)/ace/Reactor.h \ + $(ACE_ROOT)/ace/Handle_Set.h \ + $(ACE_ROOT)/ace/Handle_Set.i \ + $(ACE_ROOT)/ace/Timer_Queue.h \ + $(ACE_ROOT)/ace/Timer_Queue_T.h \ + $(ACE_ROOT)/ace/Timer_Queue_T.i \ + $(ACE_ROOT)/ace/Timer_Queue_T.cpp \ + $(ACE_ROOT)/ace/Reactor.i \ + $(ACE_ROOT)/ace/Reactor_Impl.h \ + $(ACE_ROOT)/ace/Svc_Conf_Tokens.h \ + $(ACE_ROOT)/ace/Pipe.h \ + $(ACE_ROOT)/ace/Pipe.i Event_Comm_i.h \ + $(ACE_ROOT)/ace/Map_Manager.h \ + $(ACE_ROOT)/ace/Map_Manager.i \ + $(ACE_ROOT)/ace/Map_Manager.cpp \ + Consumer_Input_Handler.h +.obj/notifier.o .obj/notifier.so .shobj/notifier.o .shobj/notifier.so: notifier.cpp Notifier_Handler.h \ + $(ACE_ROOT)/ace/CORBA_Handler.h \ + $(ACE_ROOT)/ace/Service_Config.h \ + $(ACE_ROOT)/ace/Service_Object.h \ + $(ACE_ROOT)/ace/Shared_Object.h \ + $(ACE_ROOT)/ace/ACE.h \ + $(ACE_ROOT)/ace/OS.h \ + $(ACE_ROOT)/ace/inc_user_config.h \ + $(ACE_ROOT)/ace/config.h \ + $(ACE_ROOT)/ace/config-g++-common.h \ + $(ACE_ROOT)/ace/streams.h \ + $(ACE_ROOT)/ace/Basic_Types.h \ + $(ACE_ROOT)/ace/Basic_Types.i \ + $(ACE_ROOT)/ace/OS.i \ + $(ACE_ROOT)/ace/Trace.h \ + $(ACE_ROOT)/ace/Log_Msg.h \ + $(ACE_ROOT)/ace/Log_Record.h \ + $(ACE_ROOT)/ace/ACE.i \ + $(ACE_ROOT)/ace/Log_Priority.h \ + $(ACE_ROOT)/ace/Log_Record.i \ + $(ACE_ROOT)/ace/Shared_Object.i \ + $(ACE_ROOT)/ace/Event_Handler.h \ + $(ACE_ROOT)/ace/Event_Handler.i \ + $(ACE_ROOT)/ace/Service_Object.i \ + $(ACE_ROOT)/ace/Signal.h \ + $(ACE_ROOT)/ace/Synch.h \ + $(ACE_ROOT)/ace/SV_Semaphore_Complex.h \ + $(ACE_ROOT)/ace/SV_Semaphore_Simple.h \ + $(ACE_ROOT)/ace/SV_Semaphore_Simple.i \ + $(ACE_ROOT)/ace/SV_Semaphore_Complex.i \ + $(ACE_ROOT)/ace/Synch.i \ + $(ACE_ROOT)/ace/Synch_T.h \ + $(ACE_ROOT)/ace/Synch_T.i \ + $(ACE_ROOT)/ace/Thread.h \ + $(ACE_ROOT)/ace/Thread.i \ + $(ACE_ROOT)/ace/Atomic_Op.i \ + $(ACE_ROOT)/ace/Synch_T.cpp \ + $(ACE_ROOT)/ace/Containers.h \ + $(ACE_ROOT)/ace/Containers.i \ + $(ACE_ROOT)/ace/Containers_T.h \ + $(ACE_ROOT)/ace/Containers_T.i \ + $(ACE_ROOT)/ace/Containers_T.cpp \ + $(ACE_ROOT)/ace/Malloc.h \ + $(ACE_ROOT)/ace/Malloc_Base.h \ + $(ACE_ROOT)/ace/Malloc.i \ + $(ACE_ROOT)/ace/Malloc_T.h \ + $(ACE_ROOT)/ace/Free_List.h \ + $(ACE_ROOT)/ace/Free_List.i \ + $(ACE_ROOT)/ace/Free_List.cpp \ + $(ACE_ROOT)/ace/Malloc_T.i \ + $(ACE_ROOT)/ace/Malloc_T.cpp \ + $(ACE_ROOT)/ace/Memory_Pool.h \ + $(ACE_ROOT)/ace/Mem_Map.h \ + $(ACE_ROOT)/ace/Mem_Map.i \ + $(ACE_ROOT)/ace/Memory_Pool.i \ + $(ACE_ROOT)/ace/Signal.i \ + $(ACE_ROOT)/ace/Object_Manager.h \ + $(ACE_ROOT)/ace/Object_Manager.i \ + $(ACE_ROOT)/ace/Managed_Object.h \ + $(ACE_ROOT)/ace/Managed_Object.i \ + $(ACE_ROOT)/ace/Managed_Object.cpp \ + $(ACE_ROOT)/ace/SString.h \ + $(ACE_ROOT)/ace/SString.i \ + $(ACE_ROOT)/ace/Service_Config.i \ + $(ACE_ROOT)/ace/Reactor.h \ + $(ACE_ROOT)/ace/Handle_Set.h \ + $(ACE_ROOT)/ace/Handle_Set.i \ + $(ACE_ROOT)/ace/Timer_Queue.h \ + $(ACE_ROOT)/ace/Timer_Queue_T.h \ + $(ACE_ROOT)/ace/Timer_Queue_T.i \ + $(ACE_ROOT)/ace/Timer_Queue_T.cpp \ + $(ACE_ROOT)/ace/Reactor.i \ + $(ACE_ROOT)/ace/Reactor_Impl.h \ + $(ACE_ROOT)/ace/Svc_Conf_Tokens.h \ + $(ACE_ROOT)/ace/Pipe.h \ + $(ACE_ROOT)/ace/Pipe.i Event_Comm_i.h \ + $(ACE_ROOT)/ace/Map_Manager.h \ + $(ACE_ROOT)/ace/Map_Manager.i \ + $(ACE_ROOT)/ace/Map_Manager.cpp +.obj/supplier.o .obj/supplier.so .shobj/supplier.o .shobj/supplier.so: supplier.cpp Notifier_Handler.h \ + $(ACE_ROOT)/ace/CORBA_Handler.h \ + $(ACE_ROOT)/ace/Service_Config.h \ + $(ACE_ROOT)/ace/Service_Object.h \ + $(ACE_ROOT)/ace/Shared_Object.h \ + $(ACE_ROOT)/ace/ACE.h \ + $(ACE_ROOT)/ace/OS.h \ + $(ACE_ROOT)/ace/inc_user_config.h \ + $(ACE_ROOT)/ace/config.h \ + $(ACE_ROOT)/ace/config-g++-common.h \ + $(ACE_ROOT)/ace/streams.h \ + $(ACE_ROOT)/ace/Basic_Types.h \ + $(ACE_ROOT)/ace/Basic_Types.i \ + $(ACE_ROOT)/ace/OS.i \ + $(ACE_ROOT)/ace/Trace.h \ + $(ACE_ROOT)/ace/Log_Msg.h \ + $(ACE_ROOT)/ace/Log_Record.h \ + $(ACE_ROOT)/ace/ACE.i \ + $(ACE_ROOT)/ace/Log_Priority.h \ + $(ACE_ROOT)/ace/Log_Record.i \ + $(ACE_ROOT)/ace/Shared_Object.i \ + $(ACE_ROOT)/ace/Event_Handler.h \ + $(ACE_ROOT)/ace/Event_Handler.i \ + $(ACE_ROOT)/ace/Service_Object.i \ + $(ACE_ROOT)/ace/Signal.h \ + $(ACE_ROOT)/ace/Synch.h \ + $(ACE_ROOT)/ace/SV_Semaphore_Complex.h \ + $(ACE_ROOT)/ace/SV_Semaphore_Simple.h \ + $(ACE_ROOT)/ace/SV_Semaphore_Simple.i \ + $(ACE_ROOT)/ace/SV_Semaphore_Complex.i \ + $(ACE_ROOT)/ace/Synch.i \ + $(ACE_ROOT)/ace/Synch_T.h \ + $(ACE_ROOT)/ace/Synch_T.i \ + $(ACE_ROOT)/ace/Thread.h \ + $(ACE_ROOT)/ace/Thread.i \ + $(ACE_ROOT)/ace/Atomic_Op.i \ + $(ACE_ROOT)/ace/Synch_T.cpp \ + $(ACE_ROOT)/ace/Containers.h \ + $(ACE_ROOT)/ace/Containers.i \ + $(ACE_ROOT)/ace/Containers_T.h \ + $(ACE_ROOT)/ace/Containers_T.i \ + $(ACE_ROOT)/ace/Containers_T.cpp \ + $(ACE_ROOT)/ace/Malloc.h \ + $(ACE_ROOT)/ace/Malloc_Base.h \ + $(ACE_ROOT)/ace/Malloc.i \ + $(ACE_ROOT)/ace/Malloc_T.h \ + $(ACE_ROOT)/ace/Free_List.h \ + $(ACE_ROOT)/ace/Free_List.i \ + $(ACE_ROOT)/ace/Free_List.cpp \ + $(ACE_ROOT)/ace/Malloc_T.i \ + $(ACE_ROOT)/ace/Malloc_T.cpp \ + $(ACE_ROOT)/ace/Memory_Pool.h \ + $(ACE_ROOT)/ace/Mem_Map.h \ + $(ACE_ROOT)/ace/Mem_Map.i \ + $(ACE_ROOT)/ace/Memory_Pool.i \ + $(ACE_ROOT)/ace/Signal.i \ + $(ACE_ROOT)/ace/Object_Manager.h \ + $(ACE_ROOT)/ace/Object_Manager.i \ + $(ACE_ROOT)/ace/Managed_Object.h \ + $(ACE_ROOT)/ace/Managed_Object.i \ + $(ACE_ROOT)/ace/Managed_Object.cpp \ + $(ACE_ROOT)/ace/SString.h \ + $(ACE_ROOT)/ace/SString.i \ + $(ACE_ROOT)/ace/Service_Config.i \ + $(ACE_ROOT)/ace/Reactor.h \ + $(ACE_ROOT)/ace/Handle_Set.h \ + $(ACE_ROOT)/ace/Handle_Set.i \ + $(ACE_ROOT)/ace/Timer_Queue.h \ + $(ACE_ROOT)/ace/Timer_Queue_T.h \ + $(ACE_ROOT)/ace/Timer_Queue_T.i \ + $(ACE_ROOT)/ace/Timer_Queue_T.cpp \ + $(ACE_ROOT)/ace/Reactor.i \ + $(ACE_ROOT)/ace/Reactor_Impl.h \ + $(ACE_ROOT)/ace/Svc_Conf_Tokens.h \ + $(ACE_ROOT)/ace/Pipe.h \ + $(ACE_ROOT)/ace/Pipe.i Event_Comm_i.h \ + $(ACE_ROOT)/ace/Map_Manager.h \ + $(ACE_ROOT)/ace/Map_Manager.i \ + $(ACE_ROOT)/ace/Map_Manager.cpp \ + Supplier_Input_Handler.h + +# IF YOU PUT ANYTHING HERE IT WILL GO AWAY diff --git a/TAO/examples/Event_Comm/Notifier_Handler.cpp b/TAO/examples/Event_Comm/Notifier_Handler.cpp new file mode 100644 index 00000000000..5fedf385a0e --- /dev/null +++ b/TAO/examples/Event_Comm/Notifier_Handler.cpp @@ -0,0 +1,57 @@ +// $Id$ + +#include "Notifier_Handler.h" + +ACE_RCSID(Supplier, Notifier_Handler, "$Id$") + +int +Notifier_Handler::handle_close (ACE_HANDLE, ACE_Reactor_Mask) +{ + if (this->notifier_ != 0) + { + ACE_DEBUG ((LM_DEBUG, + "closing down Notifier_Handler\n")); + CORBA::release (this->notifier_); + this->notifier_ = 0; + // *Must* be allocated dyanmically! + ::operator delete ((void *) this); + } + + return 0; +} + +Event_Comm::Notifier * +Notifier_Handler::notifier (void) +{ + return this->notifier_; +} + +void +Notifier_Handler::notifier (Event_Comm::Notifier *notifier) +{ + if (this->notifier_ != notifier) + { + CORBA::release (this->notifier_); + this->notifier_ = notifier; + } +} + +// Create and initialize a Notifier target object. + +Notifier_Handler::Notifier_Handler (const char *service_location, + const char *marker, + int putit) +{ + // Create a notifier object using the implementation class + // Notifier_i. + ACE_NEW (this->notifier_, + TIE_Event_Comm_Notifier (Notifier_i) (new Notifier_i, + marker)); +} + +// Destroy a Notifier target object. + +Notifier_Handler::~Notifier_Handler (void) +{ + this->handle_close (); +} diff --git a/TAO/examples/Event_Comm/Notifier_Handler.h b/TAO/examples/Event_Comm/Notifier_Handler.h new file mode 100644 index 00000000000..1a7571a4e1e --- /dev/null +++ b/TAO/examples/Event_Comm/Notifier_Handler.h @@ -0,0 +1,47 @@ +/* -*- C++ -*- */ +// $Id$ + +// ============================================================================ +// +// = LIBRARY +// EventComm +// +// = FILENAME +// Notifier_Handler.h +// +// = AUTHOR +// Douglas C. Schmidt (schmidt@cs.wustl.edu) +// +// ============================================================================ + +#if !defined (_NOTIFIER_HANDLER_H) +#define _NOTIFIER_HANDLER_H + +#include "ace/CORBA_Handler.h" +#include "Event_Comm_i.h" + +class Notifier_Handler + // = TITLE + // Integrate CORBA with the ACE ACE_Reactor. +{ +public: + Notifier_Handler (const char *service_location, + const char *marker = "notifier", + int putit = 1); // Default marker name. + + Event_Comm::Notifier *notifier (void); + void notifier (Event_Comm::Notifier *); + + virtual int handle_close (ACE_HANDLE = ACE_INVALID_HANDLE, + ACE_Reactor_Mask = ACE_Event_Handler::NULL_MASK); + // Close down the handler. + +private: + ~Notifier_Handler (void); + // Ensure dynamic allocation. + + Event_Comm::Notifier *notifier_; + // Pointer to an <Event_Comm::Notifier> object. +}; + +#endif /* _NOTIFIER_HANDLER_H */ diff --git a/TAO/examples/Event_Comm/Supplier_Input_Handler.cpp b/TAO/examples/Event_Comm/Supplier_Input_Handler.cpp new file mode 100644 index 00000000000..3b8db05c62b --- /dev/null +++ b/TAO/examples/Event_Comm/Supplier_Input_Handler.cpp @@ -0,0 +1,136 @@ +// $Id$ + +#include "Event_Comm_i.h" +#include "Notifier_Handler.h" +#include "Supplier_Input_Handler.h" + +ACE_RCSID(Supplier, Supplier_Input_Handler, "$Id$") + +int +Supplier_Input_Handler::handle_close (ACE_HANDLE, ACE_Reactor_Mask) +{ + ACE_DEBUG ((LM_DEBUG, + "closing down Supplier::Supplier_Input_Handler\n")); + + Event_Comm::Notifier *notifier = this->notifier_->notifier (); + ACE_ASSERT (notifier != 0); + + ACE_OS::fclose (this->fp_); + + TRY + { + // Disconnect all the consumers gracefully. + notifier->disconnect ("quit", IT_X); + } + CATCHANY + { + cerr << IT_X << endl; + } + ENDTRY; + + // Don't execute a callback here otherwise we'll recurse + // indefinitely! + if (ACE_Reactor::instance ()->remove_handler + (this, + ACE_Event_Handler::READ_MASK | ACE_Event_Handler::DONT_CALL) == -1) + ACE_ERROR ((LM_ERROR, + "%p\n", + "remove_handler")); + + // *Must* be allocated dyanmically! + ::operator delete ((void *) this); + return 0; +} + +Supplier_Input_Handler::Supplier_Input_Handler (Notifier_Handler *notifier, + ACE_HANDLE handle) // Use stdin by default. + : notifier_ (notifier), + handle_ (handle) +{ + // Register ourselves with the ACE_Reactor so that input events + // cause our handle_input() method to be dispatched automatically. + + if (ACE_Reactor::instance ()->register_handler + (this, ACE_Event_Handler::READ_MASK) == -1) + ACE_ERROR ((LM_ERROR, + "%p\n", + "register_handler")); + + this->fp_ = ACE_OS::fdopen (handle, "r"); + + if (this->fp_ == 0) + ACE_ERROR ((LM_ERROR, + "%p\n", + "fdopen")); +} + +Supplier_Input_Handler::~Supplier_Input_Handler (void) +{ + ACE_DEBUG ((LM_DEBUG, + "closing down Supplier_Input_Handler::~Supplier_Input_Handler\n")); + this->handle_close (); +} + +ACE_HANDLE +Supplier_Input_Handler::get_handle (void) const +{ + return this->handle_; +} + +// Frame input events and notify <Consumers>. + +int +Supplier_Input_Handler::handle_input (ACE_HANDLE h) +{ + char buf[BUFSIZ]; + + // Read up to BUFSIZ worth of data from ACE_HANDLE h. + + if (ACE_OS::fgets (buf, sizeof buf - 1, this->fp_) == 0) + { + ACE_OS::strcpy (buf, "quit"); + ACE_DEBUG ((LM_DEBUG, + "shutting down Supplier_Input_Handler\n")); + } + else + { + size_t n = ACE_OS::strlen (buf); + + // Null terminate the buffer, replacing the '\n' with '\0'. + if (buf[n - 1] == '\n' || buf[n - 1] == EOF) + buf[n - 1] = '\0'; + else + buf[n] = '\0'; + ACE_DEBUG ((LM_DEBUG, "notifying for event %s\n", buf)); + } + + Event_Comm::Notifier *notifier = this->notifier_->notifier (); + ACE_ASSERT (notifier != 0); + + if (ACE_OS::strcmp (buf, "quit") == 0) + // Tell the main event loop to shutdown. + ACE_Reactor::end_event_loop(); + else + { + // Use the notifier to notify Consumers. + TRY + { + Event_Comm::Event event; + + // Pass the buf over in the tag field. + event.tag_ = ACE_OS::strdup (buf); + + // This is where the "any" value goes or the object + // reference... event.value_ = ... + + // Forward <Event> to all <Consumers>. + notifier->push (event, IT_X); + } + CATCHANY + { + cerr << "unexpected exception " << IT_X << endl; + } + ENDTRY; + } + return 0; +} diff --git a/TAO/examples/Event_Comm/Supplier_Input_Handler.h b/TAO/examples/Event_Comm/Supplier_Input_Handler.h new file mode 100644 index 00000000000..8f06be607f8 --- /dev/null +++ b/TAO/examples/Event_Comm/Supplier_Input_Handler.h @@ -0,0 +1,68 @@ +/* -*- C++ -*- */ +// $Id$ + +// ============================================================================ +// +// = LIBRARY +// EventComm +// +// = FILENAME +// Supplier_Input_Handler.h +// +// = DESCRIPTION +// Handle input from the keyboard. +// +// = AUTHOR +// Douglas C. Schmidt (schmidt@cs.wustl.edu) +// +// ============================================================================ + +#if !defined (SUPPLIER_INPUT_HANDLER_H) +#define SUPPLIER_INPUT_HANDLER_H + +#include "ace/Service_Config.h" + +// Forward declaration. +class Notifier_Handler; + +class Input_Handler : public ACE_Service_Object +{ + // = TITLE + // Handles input events generated from a keyboard. + // + // = DESCRIPTION + // The events are currently framed and forwarded to + // all Consumers. In the future, we will need to + // be more selective and only send to those Consumers + // whose filtering criteria matches! +public: + Supplier_Input_Handler (Notifier_Handler *, + ACE_HANDLE = ACE_STDIN); + // Use stdin by default. + + virtual int handle_input (ACE_HANDLE); + // Frame input events and notify <Consumers>. + + virtual int handle_close (ACE_HANDLE = ACE_INVALID_HANDLE, + ACE_Reactor_Mask = ACE_Event_Handler::NULL_MASK); + // Close down the handler. + +protected: + virtual ACE_HANDLE get_handle (void) const; + + ACE_HANDLE handle_; + // ACE_HANDLE where the input comes from. + + Notifier_Handler *notifier_; + // Pointer to a <Notifier_Handler> that's used to inform Consumers + // that events of interest have occurred. + + FILE *fp_; + // Pointer to an input ACE_FILE. + +private: + ~Supplier_Input_Handler (void); + // Ensure dynamic allocation. +}; + +#endif /* SUPPLIER_INPUT_HANDLER_H */ diff --git a/TAO/examples/Event_Comm/consumer.cpp b/TAO/examples/Event_Comm/consumer.cpp new file mode 100644 index 00000000000..2056af45f4a --- /dev/null +++ b/TAO/examples/Event_Comm/consumer.cpp @@ -0,0 +1,101 @@ +// $Id$ + +#include "Consumer_Handler.h" +#include "Consumer_Input_Handler.h" + +ACE_RCSID(Consumer, consumer, "$Id$") + +class Consumer : public ACE_Event_Handler +{ + // = TITLE + // Consumer driver for the Publish/Subscribe example. +public: + Consumer (int argc, char *argv[]); + ~Consumer (void); + + void run (void); + // Execute the consumer; + +private: + virtual int handle_signal (int signum, siginfo_t *, ucontext_t *); + + virtual int handle_close (ACE_HANDLE, ACE_Reactor_Mask); + + Consumer_Input_Handler *ih_; + // Handler for keyboard input. + + Consumer_Handler *ch_; + // Handler for CORBA Consumer. +}; + +int +Consumer::handle_close (ACE_HANDLE, ACE_Reactor_Mask) +{ + ACE_DEBUG ((LM_DEBUG, + "closing down Consumer\n")); + return 0; +} + +int +Consumer::handle_signal (int signum, siginfo_t *, ucontext_t *) +{ + ACE_DEBUG ((LM_DEBUG, + "%S\n", + signum)); + + // Indicate that the consumer initiated the shutdown. + this->ih_->consumer_initiated_shutdown (1); + + // Shut down the event loop. + ACE_Reactor::end_event_loop(); + return 0; +} + +// Run the event loop until someone calls +// calls ACE_Reactor::end_event_loop(). + +void +Consumer::run (void) +{ + if (ACE_Reactor::run_event_loop () == -1) + ACE_ERROR ((LM_ERROR, + "%p\n", + "run_reactor_event_loop")); +} + +Consumer::Consumer (int argc, char *argv[]) + : ih_ (0), + ch_ (0) +{ + ACE_DEBUG ((LM_DEBUG, + "no config file, using static binding\n")); + // The constructor registers the handlers... + ACE_NEW (this->ch_, + Consumer_Handler (argc, argv)); + ACE_NEW (this->ih_, + Consumer_Input_Handler (this->ch_)); + + if (ACE_Reactor::instance ()->register_handler (SIGINT, this) == -1) + ACE_ERROR ((LM_ERROR, + "%p\n", + "register_handler")); +} + +Consumer::~Consumer (void) +{ + // Free up the handlers if they were statically bound. + this->ih_->handle_close (); + this->ch_->handle_close (); +} + +int +main (int argc, char *argv[]) +{ + // Initialize the supplier and consumer object references. + Consumer consumer (argc, argv); + + // Loop forever handling events. + consumer.run (); + + return 0; +} diff --git a/TAO/examples/Event_Comm/notifier.cpp b/TAO/examples/Event_Comm/notifier.cpp new file mode 100644 index 00000000000..0e7fa7803f0 --- /dev/null +++ b/TAO/examples/Event_Comm/notifier.cpp @@ -0,0 +1,94 @@ +// $Id$ + +#include "Notifier_Handler.h" + +ACE_RCSID(Notifier, notifier, "$Id$") + +class Notifier : public ACE_Event_Handler +{ + // = TITLE + // Notifier driver for the TAO Publish/Subscribe example. + // + // = DESCRIPTION + // The executable file generated from this code should be + // registered (under the name 'logger'). +public: + Notifier (int argc, char *argv[]); + ~Notifier (void); + + void run (void); + // Execute the notifier. + +private: + virtual int handle_signal (int signum, siginfo_t *, ucontext_t *); + + virtual int handle_close (ACE_HANDLE, ACE_Reactor_Mask); + + Notifier_Handler *nh_; + // Handler for CORBA Notifier. +}; + +int +Notifier::handle_close (ACE_HANDLE, ACE_Reactor_Mask) +{ + ACE_DEBUG ((LM_DEBUG, + "closing down Notifier\n")); + return 0; +} + +int +Notifier::handle_signal (int signum, siginfo_t *, ucontext_t *) +{ + ACE_DEBUG ((LM_DEBUG, + "%S\n", + signum)); + ACE_Reactor::end_event_loop (); + return 0; +} + +void +Notifier::run (void) +{ + if (ACE_Reactor::run_event_loop () == -1) + ACE_ERROR ((LM_ERROR, + "%p\n", + "run_reactor_event_loop")); +} + +Notifier::Notifier (int argc, char *argv[]) + : ih_ (0), + nh_ (0) +{ + ACE_DEBUG ((LM_DEBUG, + "no config file, using static binding\n")); + + // Pass in program exec name to use a service_location! + ACE_NEW (this->nh_, + Notifier_Handler (argv[0], + "notifier")); + ACE_NEW (this->ih_, + Notifier_Input_Handler (this->nh_)); + + if (ACE_Reactor::instance ()->register_handler (SIGINT, this) == -1) + ACE_ERROR ((LM_ERROR, + "%p\n", + "register_handler")); +} + +Notifier::~Notifier (void) +{ + // Free up the handlers if they were statically bound. + this->nh_->handle_close (); +} + +int +main (int argc, char *argv[]) +{ + // Initialize server daemon. + Notifier notifier (argc, argv); + + // Loop forever handling events. + notifier.run (); + + return 0; +} diff --git a/TAO/examples/Event_Comm/supplier.cpp b/TAO/examples/Event_Comm/supplier.cpp new file mode 100644 index 00000000000..204ab7cb998 --- /dev/null +++ b/TAO/examples/Event_Comm/supplier.cpp @@ -0,0 +1,90 @@ +// $Id$ + +#include "Notifier_Handler.h" +#include "Supplier_Input_Handler.h" + +ACE_RCSID(Supplier, supplier, "$Id$") + +class Supplier : public ACE_Event_Handler +{ + // = TITLE + // Supplier driver for the TAO Publish/Subscribe example. + // + // = DESCRIPTION + // The executable file generated from this code should be + // registered (under the name 'logger'). +public: + Supplier (int argc, char *argv[]); + ~Supplier (void); + + void run (void); + // Execute the supplier. + +private: + virtual int handle_signal (int signum, siginfo_t *, ucontext_t *); + + virtual int handle_close (ACE_HANDLE, ACE_Reactor_Mask); + + Supplier_Input_Handler *ih_; + // Handler for keyboard input. +}; + +int +Supplier::handle_close (ACE_HANDLE, ACE_Reactor_Mask) +{ + ACE_DEBUG ((LM_DEBUG, + "closing down Supplier\n")); + return 0; +} + +int +Supplier::handle_signal (int signum, siginfo_t *, ucontext_t *) +{ + ACE_DEBUG ((LM_DEBUG, + "%S\n", + signum)); + ACE_Reactor::end_event_loop (); + return 0; +} + +void +Supplier::run (void) +{ + if (ACE_Reactor::run_event_loop () == -1) + ACE_ERROR ((LM_ERROR, + "%p\n", + "run_reactor_event_loop")); +} + +Supplier::Supplier (int argc, char *argv[]) + : ih_ (0) +{ + ACE_DEBUG ((LM_DEBUG, + "no config file, using static binding\n")); + + ACE_NEW (this->ih_, + Supplier_Input_Handler (this->nh_)); + + if (ACE_Reactor::instance ()->register_handler (SIGINT, this) == -1) + ACE_ERROR ((LM_ERROR, + "%p\n", + "register_handler")); +} + +Supplier::~Supplier (void) +{ + // Free up the handlers if they were statically bound. + this->ih_->handle_close (); +} + +int +main (int argc, char *argv[]) +{ + // Initialize server daemon. + Supplier supplier (argc, argv); + + // Loop forever handling events. + supplier.run (); + + return 0; +} |