ACE/examples/ASX/UPIPE_Event_Server/Consumer_Router.cpp

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