diff options
Diffstat (limited to 'examples/ASX/UPIPE_Event_Server/Consumer_Router.cpp')
-rw-r--r-- | examples/ASX/UPIPE_Event_Server/Consumer_Router.cpp | 126 |
1 files changed, 126 insertions, 0 deletions
diff --git a/examples/ASX/UPIPE_Event_Server/Consumer_Router.cpp b/examples/ASX/UPIPE_Event_Server/Consumer_Router.cpp new file mode 100644 index 00000000000..e679de5390e --- /dev/null +++ b/examples/ASX/UPIPE_Event_Server/Consumer_Router.cpp @@ -0,0 +1,126 @@ +#include "Consumer_Router.h" +// @(#)Consumer_Router.cpp 1.1 10/18/96 + +#include "Options.h" + +#if defined (ACE_HAS_THREADS) + +typedef Acceptor_Factory<Consumer_Handler, CONSUMER_KEY> CONSUMER_FACTORY; + +int +Consumer_Handler::open (void *a) +{ + CONSUMER_FACTORY *af = (CONSUMER_FACTORY *) a; + this->router_task_ = af->router (); + return this->Peer_Handler<CONSUMER_ROUTER, CONSUMER_KEY>::open (a); +} + +Consumer_Handler::Consumer_Handler (ACE_Thread_Manager *tm) + : Peer_Handler<CONSUMER_ROUTER, CONSUMER_KEY> (tm) +{ +} + +// Create a new handler that will interact with a consumer and point +// its ROUTER_TASK_ data member to the CONSUMER_ROUTER. + +Consumer_Router::Consumer_Router (ACE_Thread_Manager *tm) + : CONSUMER_ROUTER (tm) +{ +} + +// Initialize the Router.. + +int +Consumer_Router::open (void *) +{ + ACE_ASSERT (this->is_reader ()); + char *argv[3]; + + argv[0] = (char *) this->name (); + argv[1] = options.consumer_file (); + argv[2] = 0; + + if (this->init (1, &argv[1]) == -1) + return -1; + + // Make this an active object. +// return this->activate (options.t_flags ()); +} + +int +Consumer_Router::close (u_long) +{ + ACE_ASSERT (this->is_reader ()); + this->peer_map_.close (); + this->msg_queue ()->deactivate(); + return 0; +} + + +// Handle incoming messages in a separate thread.. + +int +Consumer_Router::svc (void) +{ + ACE_Thread_Control tc (this->thr_mgr ()); + ACE_Message_Block *mb = 0; + + ACE_ASSERT (this->is_reader ()); + + if (options.debug ()) + ACE_DEBUG ((LM_DEBUG, "(%t) starting svc in %s\n", this->name ())); + + while (this->getq (mb) > 0) + { + if (this->put_next (mb) == -1) + ACE_ERROR_RETURN ((LM_ERROR, "(%t) put_next failed in %s\n", this->name ()), -1); + + } + return 0; + // Note the implicit ACE_OS::thr_exit() via destructor. +} + +// Send a MESSAGE_BLOCK to the supplier(s).. + +int +Consumer_Router::put (ACE_Message_Block *mb, ACE_Time_Value *) +{ + ACE_ASSERT (this->is_reader ()); + + if (mb->msg_type () == ACE_Message_Block::MB_IOCTL) + { + this->control (mb); + return this->put_next (mb); + } + else +{ +//printf("consumer-Router is routing : send_peers\n"); + return this->send_peers (mb); +} +} + +// Return information about the Client_Router ACE_Module.. + +int +Consumer_Router::info (char **strp, size_t length) const +{ + char buf[BUFSIZ]; + ACE_UPIPE_Addr addr; + const char *mod_name = this->name (); + ACE_UPIPE_Acceptor &sa = (ACE_UPIPE_Acceptor &) *this->acceptor_; + + if (sa.get_local_addr (addr) == -1) + return -1; + + ACE_OS::sprintf (buf, "%s\t /%s %s", + mod_name, "upipe", + "# consumer router\n"); + + if (*strp == 0 && (*strp = ACE_OS::strdup (mod_name)) == 0) + return -1; + else + ACE_OS::strncpy (*strp, mod_name, length); + return ACE_OS::strlen (mod_name); +} + +#endif /* ACE_HAS_THREADS */ |