diff options
Diffstat (limited to 'ACE/examples/ASX/UPIPE_Event_Server/Supplier_Router.cpp')
-rw-r--r-- | ACE/examples/ASX/UPIPE_Event_Server/Supplier_Router.cpp | 137 |
1 files changed, 137 insertions, 0 deletions
diff --git a/ACE/examples/ASX/UPIPE_Event_Server/Supplier_Router.cpp b/ACE/examples/ASX/UPIPE_Event_Server/Supplier_Router.cpp new file mode 100644 index 00000000000..8b4a53d6331 --- /dev/null +++ b/ACE/examples/ASX/UPIPE_Event_Server/Supplier_Router.cpp @@ -0,0 +1,137 @@ +// $Id$ + +#include "ace/OS_NS_stdio.h" +#include "ace/OS_NS_string.h" +#include "Options.h" +#include "Supplier_Router.h" + +ACE_RCSID(UPIPE_Event_Server, Supplier_Router, "$Id$") + +#if defined (ACE_HAS_THREADS) + +typedef Acceptor_Factory<Supplier_Handler, SUPPLIER_KEY> SUPPLIER_FACTORY; + +int +Supplier_Handler::open (void *a) +{ + SUPPLIER_FACTORY *af = (SUPPLIER_FACTORY *) a; + this->router_task_ = af->router (); + return this->Peer_Handler<SUPPLIER_ROUTER, SUPPLIER_KEY>::open (a); +} + +Supplier_Handler::Supplier_Handler (ACE_Thread_Manager *tm) + : Peer_Handler<SUPPLIER_ROUTER, SUPPLIER_KEY> (tm) +{ +} + +// Create a new router and associate it with the REACTOR parameter.. + +Supplier_Router::Supplier_Router (ACE_Thread_Manager *tm) + : SUPPLIER_ROUTER (tm) +{ +} + +// Handle incoming messages in a separate thread.. + +int +Supplier_Router::svc (void) +{ + ACE_ASSERT (this->is_writer ()); + + ACE_Message_Block *message_block = 0; + + if (options.debug ()) + ACE_DEBUG ((LM_DEBUG, "(%t) starting svc in %s\n", this->name ())); + + while (this->getq (message_block) > 0) + { + if (this->put_next (message_block) == -1) + ACE_ERROR_RETURN ((LM_ERROR, "(%t) put_next failed in %s\n", this->name ()), -1); + } + + return 0; +} + +// Initialize the Router.. + +int +Supplier_Router::open (void *) +{ + ACE_ASSERT (this->is_writer ()); + ACE_TCHAR *argv[3]; + + argv[0] = (ACE_TCHAR *)this->name (); + argv[1] = (ACE_TCHAR *)options.supplier_file (); + argv[2] = 0; + + if (this->init (1, &argv[1]) == -1) + return -1; + + // Make this an active object. + // return this->activate (options.t_flags ()); + + // Until that's done, return 1 to indicate that the object wasn't activated. + return 1; +} + +// Close down the router.. + +int +Supplier_Router::close (u_long) +{ + ACE_ASSERT (this->is_writer ()); + this->peer_map_.close (); + this->msg_queue ()->deactivate(); + return 0; +} + +// Send a MESSAGE_BLOCK to the supplier(s).. + +int +Supplier_Router::put (ACE_Message_Block *mb, ACE_Time_Value *) +{ + ACE_ASSERT (this->is_writer ()); + + if (mb->msg_type () == ACE_Message_Block::MB_IOCTL) + { + this->control (mb); + return this->put_next (mb); + } + else + { +//printf("supplier-Router is routing: send_peers\n"); + return this->send_peers (mb); + } +} + +// Return information about the Supplier_Router ACE_Module.. + +int +Supplier_Router::info (ACE_TCHAR **strp, size_t length) const +{ + ACE_TCHAR buf[BUFSIZ]; + ACE_UPIPE_Addr addr; + const ACE_TCHAR *mod_name = this->name (); + ACE_UPIPE_Acceptor &sa = (ACE_UPIPE_Acceptor &) *this->acceptor_; + + if (sa.get_local_addr (addr) == -1) + return -1; + +#if !defined (ACE_WIN32) && defined (ACE_USES_WCHAR) +# define FMTSTR ACE_TEXT ("%ls\t %ls/ %ls") +#else +# define FMTSTR ACE_TEXT ("%s\t %s/ %s") +#endif + + ACE_OS::sprintf (buf, FMTSTR, + mod_name, ACE_TEXT ("upipe"), + ACE_TEXT ("# supplier 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 */ |