ACE/examples/APG/Reactor/HAStatus.cpp

   1 #include "ace/OS_NS_sys_time.h"
   2 #include "ace/os_include/os_netdb.h"
   3
   4 // Listing 1 code/ch07
   5 #include "ace/Log_Msg.h"
   6 #include "ace/INET_Addr.h"
   7 #include "ace/SOCK_Acceptor.h"
   8 #include "ace/Reactor.h"
   9 #include <memory>
  10
  11 class ClientAcceptor : public ACE_Event_Handler
  12 {
  13 public:
  14   virtual ~ClientAcceptor ();
  15
  16   //FUZZ: disable check_for_lack_ACE_OS
  17   int open (const ACE_INET_Addr &listen_addr);
  18   //FUZZ: enable check_for_lack_ACE_OS
  19
  20   // Get this handler's I/O handle.
  21   virtual ACE_HANDLE get_handle () const
  22     { return this->acceptor_.get_handle (); }
  23
  24   // Called when a connection is ready to accept.
  25   virtual int handle_input (ACE_HANDLE fd = ACE_INVALID_HANDLE);
  26
  27   // Called when this handler is removed from the ACE_Reactor.
  28   virtual int handle_close (ACE_HANDLE handle,
  29                             ACE_Reactor_Mask close_mask);
  30
  31 protected:
  32   ACE_SOCK_Acceptor acceptor_;
  33 };
  34 // Listing 1
  35
  36 // Listing 6 code/ch07
  37 #include "ace/Message_Block.h"
  38 #include "ace/Message_Queue.h"
  39 #include "ace/SOCK_Stream.h"
  40 #include "ace/Synch.h"
  41
  42 class ClientService : public ACE_Event_Handler
  43 {
  44 public:
  45   ACE_SOCK_Stream &peer () { return this->sock_; }
  46
  47   //FUZZ: disable check_for_lack_ACE_OS
  48   int open ();
  49   //FUZZ: enable check_for_lack_ACE_OS
  50
  51   // Get this handler's I/O handle.
  52   virtual ACE_HANDLE get_handle () const
  53     { return this->sock_.get_handle (); }
  54
  55   // Called when input is available from the client.
  56   virtual int handle_input (ACE_HANDLE fd = ACE_INVALID_HANDLE);
  57
  58   // Called when output is possible.
  59   virtual int handle_output (ACE_HANDLE fd = ACE_INVALID_HANDLE);
  60
  61   // Called when this handler is removed from the ACE_Reactor.
  62   virtual int handle_close (ACE_HANDLE handle,
  63                             ACE_Reactor_Mask close_mask);
  64
  65 protected:
  66   ACE_SOCK_Stream sock_;
  67   ACE_Message_Queue<ACE_NULL_SYNCH> output_queue_;
  68 };
  69 // Listing 6
  70
  71 // Listing 5 code/ch07
  72 ClientAcceptor::~ClientAcceptor ()
  73 {
  74   this->handle_close (ACE_INVALID_HANDLE, 0);
  75 }
  76 // Listing 5
  77
  78 // Listing 2 code/ch07
  79 int
  80 ClientAcceptor::open (const ACE_INET_Addr &listen_addr)
  81 {
  82   if (this->acceptor_.open (listen_addr, 1) == -1)
  83     ACE_ERROR_RETURN ((LM_ERROR,
  84                        ACE_TEXT ("%p\n"),
  85                        ACE_TEXT ("acceptor.open")),
  86                       -1);
  87   return this->reactor ()->register_handler
  88     (this, ACE_Event_Handler::ACCEPT_MASK);
  89 }
  90 // Listing 2
  91
  92 // Listing 3 code/ch07
  93 int
  94 ClientAcceptor::handle_input (ACE_HANDLE)
  95 {
  96   ClientService *client = 0;
  97   ACE_NEW_RETURN (client, ClientService, -1);
  98   std::unique_ptr<ClientService> p (client);
  99
 100   if (this->acceptor_.accept (client->peer ()) == -1)
 101     ACE_ERROR_RETURN ((LM_ERROR,
 102                        ACE_TEXT ("(%P|%t) %p\n"),
 103                        ACE_TEXT ("Failed to accept ")
 104                        ACE_TEXT ("client connection")),
 105                       -1);
 106   p.release ();
 107   client->reactor (this->reactor ());
 108   if (client->open () == -1)
 109     client->handle_close (ACE_INVALID_HANDLE, 0);
 110   return 0;
 111 }
 112 // Listing 3
 113
 114 // Listing 4 code/ch07
 115 int
 116 ClientAcceptor::handle_close (ACE_HANDLE, ACE_Reactor_Mask)
 117 {
 118   if (this->acceptor_.get_handle () != ACE_INVALID_HANDLE)
 119     {
 120       ACE_Reactor_Mask m = ACE_Event_Handler::ACCEPT_MASK |
 121                            ACE_Event_Handler::DONT_CALL;
 122       this->reactor ()->remove_handler (this, m);
 123       this->acceptor_.close ();
 124     }
 125   return 0;
 126 }
 127 // Listing 4
 128
 129 // Listing 7 code/ch07
 130 int
 131 ClientService::open ()
 132 {
 133   ACE_TCHAR peer_name[MAXHOSTNAMELEN];
 134   ACE_INET_Addr peer_addr;
 135   if (this->sock_.get_remote_addr (peer_addr) == 0 &&
 136       peer_addr.addr_to_string (peer_name, MAXHOSTNAMELEN) == 0)
 137     ACE_DEBUG ((LM_DEBUG,
 138                 ACE_TEXT ("(%P|%t) Connection from %s\n"),
 139                 peer_name));
 140   return this->reactor ()->register_handler
 141     (this, ACE_Event_Handler::READ_MASK);
 142 }
 143 // Listing 7
 144
 145 // Listing 8 code/ch07
 146 int
 147 ClientService::handle_input (ACE_HANDLE)
 148 {
 149   const size_t INPUT_SIZE = 4096;
 150   char buffer[INPUT_SIZE];
 151   ssize_t recv_cnt, send_cnt;
 152
 153   if ((recv_cnt = this->sock_.recv (buffer, sizeof(buffer))) <= 0)
 154     {
 155       ACE_DEBUG ((LM_DEBUG,
 156                   ACE_TEXT ("(%P|%t) Connection closed\n")));
 157       return -1;
 158     }
 159
 160   send_cnt =
 161     this->sock_.send (buffer, static_cast<size_t> (recv_cnt));
 162   if (send_cnt == recv_cnt)
 163     return 0;
 164   if (send_cnt == -1 && ACE_OS::last_error () != EWOULDBLOCK)
 165     ACE_ERROR_RETURN ((LM_ERROR,
 166                        ACE_TEXT ("(%P|%t) %p\n"),
 167                        ACE_TEXT ("send")),
 168                       0);
 169   if (send_cnt == -1)
 170     send_cnt = 0;
 171   ACE_Message_Block *mb = 0;
 172   size_t remaining =
 173     static_cast<size_t> ((recv_cnt - send_cnt));
 174   ACE_NEW_RETURN (mb, ACE_Message_Block (remaining), -1);
 175   mb->copy (&buffer[send_cnt], remaining);
 176   int output_off = this->output_queue_.is_empty ();
 177   ACE_Time_Value nowait (ACE_OS::gettimeofday ());
 178   if (this->output_queue_.enqueue_tail (mb, &nowait) == -1)
 179     {
 180       ACE_ERROR ((LM_ERROR,
 181                   ACE_TEXT ("(%P|%t) %p; discarding data\n"),
 182                   ACE_TEXT ("enqueue failed")));
 183       mb->release ();
 184       return 0;
 185     }
 186   if (output_off)
 187     return this->reactor ()->register_handler
 188       (this, ACE_Event_Handler::WRITE_MASK);
 189   return 0;
 190 }
 191 // Listing 8
 192
 193 // Listing 9 code/ch07
 194 int
 195 ClientService::handle_output (ACE_HANDLE)
 196 {
 197   ACE_Message_Block *mb = 0;
 198   ACE_Time_Value nowait (ACE_OS::gettimeofday ());
 199   while (0 <= this->output_queue_.dequeue_head
 200                                     (mb, &nowait))
 201     {
 202       ssize_t send_cnt =
 203         this->sock_.send (mb->rd_ptr (), mb->length ());
 204       if (send_cnt == -1)
 205         ACE_ERROR ((LM_ERROR,
 206                     ACE_TEXT ("(%P|%t) %p\n"),
 207                     ACE_TEXT ("send")));
 208       else
 209         mb->rd_ptr (static_cast<size_t> (send_cnt));
 210       if (mb->length () > 0)
 211         {
 212           this->output_queue_.enqueue_head (mb);
 213           break;
 214         }
 215       mb->release ();
 216     }
 217   return (this->output_queue_.is_empty ()) ? -1 : 0;
 218 }
 219 // Listing 9
 220
 221 // Listing 10 code/ch07
 222 int
 223 ClientService::handle_close (ACE_HANDLE, ACE_Reactor_Mask mask)
 224 {
 225   if (mask == ACE_Event_Handler::WRITE_MASK)
 226     return 0;
 227   mask = ACE_Event_Handler::ALL_EVENTS_MASK |
 228          ACE_Event_Handler::DONT_CALL;
 229   this->reactor ()->remove_handler (this, mask);
 230   this->sock_.close ();
 231   this->output_queue_.flush ();
 232   delete this;
 233   return 0;
 234 }
 235 // Listing 10
 236
 237 // Listing 12 code/ch07
 238 class LoopStopper : public ACE_Event_Handler
 239 {
 240 public:
 241   LoopStopper (int signum = SIGINT);
 242
 243   // Called when object is signaled by OS.
 244   virtual int handle_signal (int signum, siginfo_t * = 0, ucontext_t * = 0);
 245 };
 246
 247 LoopStopper::LoopStopper (int signum)
 248 {
 249   ACE_Reactor::instance ()->register_handler (signum, this);
 250 }
 251
 252 int
 253 LoopStopper::handle_signal (int, siginfo_t *, ucontext_t *)
 254 {
 255   ACE_Reactor::instance ()->end_reactor_event_loop ();
 256   return 0;
 257 }
 258 // Listing 12
 259
 260 // Listing 13 code/ch07
 261 #include "ace/Signal.h"
 262
 263 class LogSwitcher : public ACE_Event_Handler
 264 {
 265 public:
 266   LogSwitcher (int on_sig, int off_sig);
 267
 268   // Called when object is signaled by OS.
 269   virtual int handle_signal (int signum, siginfo_t * = 0, ucontext_t * = 0);
 270
 271   // Called when an exceptional event occurs.
 272   virtual int handle_exception (ACE_HANDLE fd = ACE_INVALID_HANDLE);
 273
 274 private:
 275   LogSwitcher () {}
 276
 277   int on_sig_;       // Signal to turn logging on
 278   int off_sig_;      // Signal to turn logging off
 279   int on_off_;       // 1 == turn on, 0 == turn off
 280 };
 281
 282 LogSwitcher::LogSwitcher (int on_sig, int off_sig)
 283   : on_sig_ (on_sig), off_sig_ (off_sig)
 284 {
 285   ACE_Sig_Set sigs;
 286   sigs.sig_add (on_sig);
 287   sigs.sig_add (off_sig);
 288   ACE_Reactor::instance ()->register_handler (sigs, this);
 289 }
 290 // Listing 13
 291
 292 // Listing 14 code/ch07
 293 int
 294 LogSwitcher::handle_signal (int signum, siginfo_t *, ucontext_t *)
 295 {
 296   if (signum == this->on_sig_ || signum == this->off_sig_)
 297     {
 298       this->on_off_ = signum == this->on_sig_;
 299       ACE_Reactor::instance ()->notify (this);
 300     }
 301   return 0;
 302 }
 303 // Listing 14
 304
 305 // Listing 15 code/ch07
 306 int
 307 LogSwitcher::handle_exception (ACE_HANDLE)
 308 {
 309   if (this->on_off_)
 310     ACE_LOG_MSG->clr_flags (ACE_Log_Msg::SILENT);
 311   else
 312     ACE_LOG_MSG->set_flags (ACE_Log_Msg::SILENT);
 313   return 0;
 314 }
 315 // Listing 15
 316
 317 // Listing 11 code/ch07
 318 int ACE_TMAIN (int, ACE_TCHAR *[])
 319 {
 320   ACE_INET_Addr port_to_listen ("HAStatus");
 321   ClientAcceptor acceptor;
 322   acceptor.reactor (ACE_Reactor::instance ());
 323   if (acceptor.open (port_to_listen) == -1)
 324     return 1;
 325
 326   ACE_Reactor::instance ()->run_reactor_event_loop ();
 327
 328   return (0);
 329 }
 330 // Listing 11