ACE/examples/ASX/UPIPE_Event_Server/Supplier_Router.cpp

   1 #include "ace/OS_NS_stdio.h"
   2 #include "ace/OS_NS_string.h"
   3 #include "ace/Truncate.h"
   4 #include "Options.h"
   5 #include "Supplier_Router.h"
   6
   7
   8 #if defined (ACE_HAS_THREADS)
   9
  10 typedef Acceptor_Factory<Supplier_Handler, SUPPLIER_KEY> SUPPLIER_FACTORY;
  11
  12 int
  13 Supplier_Handler::open (void *a)
  14 {
  15   SUPPLIER_FACTORY *af = (SUPPLIER_FACTORY *) a;
  16   this->router_task_ = af->router ();
  17   return this->Peer_Handler<SUPPLIER_ROUTER, SUPPLIER_KEY>::open (a);
  18 }
  19
  20 Supplier_Handler::Supplier_Handler (ACE_Thread_Manager *tm)
  21   : Peer_Handler<SUPPLIER_ROUTER, SUPPLIER_KEY> (tm)
  22 {
  23 }
  24
  25 // Create a new router and associate it with the REACTOR parameter..
  26
  27 Supplier_Router::Supplier_Router (ACE_Thread_Manager *tm)
  28   : SUPPLIER_ROUTER (tm)
  29 {
  30 }
  31
  32 // Handle incoming messages in a separate thread..
  33
  34 int
  35 Supplier_Router::svc ()
  36 {
  37   ACE_ASSERT (this->is_writer ());
  38
  39   ACE_Message_Block *message_block = 0;
  40
  41   if (options.debug ())
  42     ACE_DEBUG ((LM_DEBUG, "(%t) starting svc in %s\n", this->name ()));
  43
  44   while (this->getq (message_block) > 0)
  45   {
  46     if (this->put_next (message_block) == -1)
  47       ACE_ERROR_RETURN ((LM_ERROR, "(%t) put_next failed in %s\n", this->name ()), -1);
  48   }
  49
  50   return 0;
  51 }
  52
  53 // Initialize the Router..
  54
  55 int
  56 Supplier_Router::open (void *)
  57 {
  58   ACE_ASSERT (this->is_writer ());
  59   ACE_TCHAR *argv[3];
  60
  61   argv[0] = (ACE_TCHAR *)this->name ();
  62   argv[1] = (ACE_TCHAR *)options.supplier_file ();
  63   argv[2] = 0;
  64
  65   if (this->init (1, &argv[1]) == -1)
  66     return -1;
  67
  68   // Make this an active object.
  69   //   return this->activate (options.t_flags ());
  70
  71   // Until that's done, return 1 to indicate that the object wasn't activated.
  72   return 1;
  73 }
  74
  75 // Close down the router..
  76
  77 int
  78 Supplier_Router::close (u_long)
  79 {
  80   ACE_ASSERT (this->is_writer ());
  81   this->peer_map_.close ();
  82   this->msg_queue ()->deactivate();
  83   return 0;
  84 }
  85
  86 // Send a MESSAGE_BLOCK to the supplier(s)..
  87
  88 int
  89 Supplier_Router::put (ACE_Message_Block *mb, ACE_Time_Value *)
  90 {
  91   ACE_ASSERT (this->is_writer ());
  92
  93   if (mb->msg_type () == ACE_Message_Block::MB_IOCTL)
  94     {
  95       this->control (mb);
  96       return this->put_next (mb);
  97     }
  98   else
  99     {
 100 //printf("supplier-Router is routing: send_peers\n");
 101       return this->send_peers (mb);
 102     }
 103 }
 104
 105 // Return information about the Supplier_Router ACE_Module..
 106
 107 int
 108 Supplier_Router::info (ACE_TCHAR **strp, size_t length) const
 109 {
 110   ACE_TCHAR       buf[BUFSIZ];
 111   ACE_UPIPE_Addr  addr;
 112   const ACE_TCHAR *module_name = this->name ();
 113   ACE_UPIPE_Acceptor &sa = (ACE_UPIPE_Acceptor &) *this->acceptor_;
 114
 115   if (sa.get_local_addr (addr) == -1)
 116     return -1;
 117
 118   ACE_OS::sprintf (buf,
 119                    ACE_TEXT ("%") ACE_TEXT_PRIs
 120                    ACE_TEXT ("\t %") ACE_TEXT_PRIs
 121                    ACE_TEXT ("/ %") ACE_TEXT_PRIs,
 122                    module_name, ACE_TEXT ("upipe"),
 123                    ACE_TEXT ("# supplier router\n"));
 124
 125   if (*strp == 0 && (*strp = ACE_OS::strdup (module_name)) == 0)
 126     return -1;
 127   else
 128     ACE_OS::strncpy (*strp, module_name, length);
 129
 130   return ACE_Utils::truncate_cast<int> (ACE_OS::strlen (module_name));
 131 }
 132
 133 #endif /* ACE_HAS_THREADS */