ACE/examples/C++NPv2/AIO_Client_Logging_Daemon.cpp

   1 /*
   2 ** Copyright 2002 Addison Wesley. All Rights Reserved.
   3 */
   4
   5 #include "ace/config-all.h"
   6
   7 #if defined (ACE_WIN32) || (defined (ACE_HAS_AIO_CALLS))
   8
   9 #include "ace/OS_NS_string.h"
  10 #include "ace/OS_NS_sys_socket.h"
  11 #include "ace/Asynch_Acceptor.h"
  12 #include "ace/Asynch_Connector.h"
  13 #include "ace/Asynch_IO.h"
  14 #include "ace/CDR_Stream.h"
  15 #include "ace/Get_Opt.h"
  16 #include "ace/INET_Addr.h"
  17 #include "ace/Message_Block.h"
  18 #include "ace/Null_Condition.h"
  19 #include "ace/Null_Mutex.h"
  20 #include "ace/Proactor.h"
  21 #include "ace/SOCK_Stream.h"
  22 #include "ace/Service_Object.h"
  23 #include "ace/Signal.h"
  24 #include "ace/Singleton.h"
  25 #include "ace/Task.h"
  26 #include "ace/Thread_Manager.h"
  27 #include "ace/Unbounded_Set.h"
  28 #include "ace/os_include/os_netdb.h"
  29 #include "AIO_CLD_export.h"
  30 #include "AIO_Client_Logging_Daemon.h"
  31 #include <openssl/ssl.h>
  32
  33 class AIO_CLD_Acceptor
  34   : public ACE_Asynch_Acceptor<AIO_Input_Handler> {
  35 public:
  36   //FUZZ: disable check_for_lack_ACE_OS
  37   // Cancel accept and close all clients.
  38   void close ();
  39   //FUZZ: enable check_for_lack_ACE_OS
  40
  41   // Remove handler from client set.
  42   void remove (AIO_Input_Handler *ih)
  43   { clients_.remove (ih); }
  44
  45 protected:
  46   // Service handler factory method.
  47   virtual AIO_Input_Handler *make_handler ();
  48
  49   // Set of all connected clients
  50   ACE_Unbounded_Set<AIO_Input_Handler *> clients_;
  51 };
  52
  53
  54 class AIO_Client_Logging_Daemon : public ACE_Task<ACE_NULL_SYNCH> {
  55 protected:
  56   ACE_INET_Addr cld_addr_;
  57   ACE_INET_Addr sld_addr_;
  58
  59   // Factory that passively connects the <AIO_Input_Handler>.
  60   AIO_CLD_Acceptor acceptor_;
  61
  62 public:
  63   // Service Configurator hook methods.
  64   virtual int init (int argc, ACE_TCHAR *argv[]);
  65   virtual int fini ();
  66   virtual int svc ();
  67 };
  68
  69 /******************************************************/
  70
  71 AIO_Output_Handler::~AIO_Output_Handler () {
  72   reader_.cancel ();
  73   writer_.cancel ();
  74   ACE_OS::closesocket (handle ());
  75 }
  76
  77 int AIO_Output_Handler::put (ACE_Message_Block *mb,
  78                              ACE_Time_Value *timeout) {
  79   if (can_write_) { start_write (mb); return 0; }
  80   return putq (mb, timeout);
  81 }
  82
  83 void AIO_Output_Handler::open
  84   (ACE_HANDLE new_handle, ACE_Message_Block &) {
  85   ACE_SOCK_Stream peer (new_handle);
  86   int bufsiz = ACE_DEFAULT_MAX_SOCKET_BUFSIZ;
  87   peer.set_option (SOL_SOCKET, SO_SNDBUF,
  88                    &bufsiz, sizeof bufsiz);
  89
  90   reader_.open (*this, new_handle, 0, proactor ());
  91   writer_.open (*this, new_handle, 0, proactor ());
  92
  93   ACE_Message_Block *mb = 0;
  94   ACE_NEW (mb, ACE_Message_Block (1));
  95   reader_.read (*mb, 1);
  96   ACE_Sig_Action no_sigpipe ((ACE_SignalHandler) SIG_IGN);
  97   no_sigpipe.register_action (SIGPIPE, 0);
  98   can_write_ = 1;
  99   start_write (0);
 100 }
 101
 102 void AIO_Output_Handler::start_write (ACE_Message_Block *mblk) {
 103   if (mblk == 0) {
 104     ACE_Time_Value nonblock (0);
 105     getq (mblk, &nonblock);
 106   }
 107   if (mblk != 0) {
 108     can_write_ = 0;
 109     if (writer_.write (*mblk, mblk->length ()) == -1)
 110       ungetq (mblk);
 111   }
 112 }
 113
 114 void AIO_Output_Handler::handle_read_stream
 115        (const ACE_Asynch_Read_Stream::Result &result) {
 116   result.message_block ().release ();
 117   writer_.cancel ();
 118   ACE_OS::closesocket (result.handle ());
 119   handle (ACE_INVALID_HANDLE);
 120   can_write_ = 0;
 121   CLD_CONNECTOR::instance ()->reconnect ();
 122 }
 123
 124 void AIO_Output_Handler::handle_write_stream
 125        (const ACE_Asynch_Write_Stream::Result &result) {
 126   ACE_Message_Block &mblk = result.message_block ();
 127   if (!result.success ()) {
 128     mblk.rd_ptr (mblk.base ());
 129     ungetq (&mblk);
 130   }
 131   else {
 132     can_write_ = handle () == result.handle ();
 133     if (mblk.length () == 0) {
 134       mblk.release ();
 135       if (can_write_) start_write ();
 136     }
 137     else if (can_write_) start_write (&mblk);
 138     else { mblk.rd_ptr (mblk.base ()); ungetq (&mblk); }
 139   }
 140 }
 141
 142 /******************************************************/
 143
 144 AIO_Input_Handler::~AIO_Input_Handler () {
 145   reader_.cancel ();
 146   ACE_OS::closesocket (handle ());
 147   if (mblk_ != 0) mblk_->release ();
 148   mblk_ = 0;
 149   acceptor_->remove (this);
 150 }
 151
 152 void AIO_Input_Handler::open
 153   (ACE_HANDLE new_handle, ACE_Message_Block &) {
 154   reader_.open (*this, new_handle, 0, proactor ());
 155   ACE_NEW_NORETURN
 156     (mblk_, ACE_Message_Block (ACE_DEFAULT_CDR_BUFSIZE));
 157   // Align the Message Block for a CDR stream
 158   ACE_CDR::mb_align (mblk_);
 159   reader_.read (*mblk_, LOG_HEADER_SIZE);
 160 }
 161
 162 void AIO_Input_Handler::handle_read_stream
 163     (const ACE_Asynch_Read_Stream::Result &result) {
 164   if (!result.success () || result.bytes_transferred () == 0)
 165     delete this;
 166   else if (result.bytes_transferred () < result.bytes_to_read ())
 167     reader_.read (*mblk_, result.bytes_to_read () -
 168                   result.bytes_transferred ());
 169   else if (mblk_->length () == LOG_HEADER_SIZE) {
 170     ACE_InputCDR cdr (mblk_);
 171
 172     ACE_CDR::Boolean byte_order;
 173     cdr >> ACE_InputCDR::to_boolean (byte_order);
 174     cdr.reset_byte_order (byte_order);
 175
 176     ACE_CDR::ULong length;
 177     cdr >> length;
 178
 179     mblk_->size (length + LOG_HEADER_SIZE);
 180     reader_.read (*mblk_, length);
 181   }
 182   else {
 183     if (OUTPUT_HANDLER::instance ()->put (mblk_) == -1)
 184       mblk_->release ();
 185
 186     ACE_NEW_NORETURN
 187       (mblk_, ACE_Message_Block (ACE_DEFAULT_CDR_BUFSIZE));
 188     ACE_CDR::mb_align (mblk_);
 189     reader_.read (*mblk_, LOG_HEADER_SIZE);
 190   }
 191 }
 192
 193 /********************************************************/
 194
 195 void AIO_CLD_Acceptor::close () {
 196   ACE_Unbounded_Set_Iterator<AIO_Input_Handler *>
 197     iter (clients_.begin ());
 198   AIO_Input_Handler **ih;
 199   for (; iter.next (ih); ++iter)
 200     delete *ih;
 201 }
 202
 203 AIO_Input_Handler * AIO_CLD_Acceptor::make_handler () {
 204   AIO_Input_Handler *ih;
 205   ACE_NEW_RETURN (ih, AIO_Input_Handler (this), 0);
 206   if (clients_.insert (ih) == -1)
 207     { delete ih; return 0; }
 208   return ih;
 209 }
 210
 211 /*******************************************************/
 212
 213 #if !defined (CLD_CERTIFICATE_FILENAME)
 214 #  define CLD_CERTIFICATE_FILENAME "cld-cert.pem"
 215 #endif /* !CLD_CERTIFICATE_FILENAME */
 216 #if !defined (CLD_KEY_FILENAME)
 217 #  define CLD_KEY_FILENAME "cld-key.pem"
 218 #endif /* !CLD_KEY_FILENAME */
 219
 220
 221 int AIO_CLD_Connector::validate_connection
 222   (const ACE_Asynch_Connect::Result& result,
 223    const ACE_INET_Addr &remote, const ACE_INET_Addr&) {
 224   remote_addr_ = remote;
 225   if (!result.success ()) {
 226     ACE_Time_Value delay (retry_delay_);
 227     retry_delay_ *= 2;
 228     if (retry_delay_ > MAX_RETRY_DELAY)
 229       retry_delay_ = MAX_RETRY_DELAY;
 230     proactor ()->schedule_timer (*this, 0, delay);
 231     return -1;
 232   }
 233   retry_delay_ = INITIAL_RETRY_DELAY;
 234
 235   if (ssl_ctx_ == 0) {
 236     OpenSSL_add_ssl_algorithms ();
 237     ssl_ctx_ = SSL_CTX_new (SSLv23_client_method ());
 238     if (ssl_ctx_ == 0) return -1;
 239
 240     if (SSL_CTX_use_certificate_file (ssl_ctx_,
 241                                       CLD_CERTIFICATE_FILENAME,
 242                                       SSL_FILETYPE_PEM) <= 0
 243        || SSL_CTX_use_PrivateKey_file (ssl_ctx_,
 244                                        CLD_KEY_FILENAME,
 245                                        SSL_FILETYPE_PEM) <= 0
 246        || !SSL_CTX_check_private_key (ssl_ctx_)) {
 247       SSL_CTX_free (ssl_ctx_);
 248       ssl_ctx_ = 0;
 249       return -1;
 250     }
 251     ssl_ = SSL_new (ssl_ctx_);
 252     if (ssl_ == 0) {
 253       SSL_CTX_free (ssl_ctx_); ssl_ctx_ = 0;
 254       return -1;
 255     }
 256   }
 257
 258   SSL_clear (ssl_);
 259 #if defined (ACE_WIN32)
 260   // ACE_WIN32 is the only platform where ACE_HANDLE is not an int.
 261   // See ace/config-lite.h for the typedefs.
 262   SSL_set_fd (ssl_, reinterpret_cast<int> (result.connect_handle ()));
 263 #else
 264   SSL_set_fd (ssl_, result.connect_handle ());
 265 #endif /* ACE_WIN32 */
 266   SSL_set_verify (ssl_, SSL_VERIFY_PEER, 0);
 267
 268   if (SSL_connect (ssl_) == -1
 269       || SSL_shutdown (ssl_) == -1) return -1;
 270   return 0;
 271 }
 272
 273 void AIO_CLD_Connector::handle_time_out
 274 (const ACE_Time_Value&, const void *) {
 275   this->connect (remote_addr_);
 276 }
 277
 278 /******************************************************/
 279
 280 int AIO_Client_Logging_Daemon::init
 281       (int argc, ACE_TCHAR *argv[]) {
 282   u_short cld_port = ACE_DEFAULT_SERVICE_PORT;
 283   u_short sld_port = ACE_DEFAULT_LOGGING_SERVER_PORT;
 284   ACE_TCHAR sld_host[MAXHOSTNAMELEN];
 285   ACE_OS::strcpy (sld_host, ACE_LOCALHOST);
 286
 287   ACE_Get_Opt get_opt (argc, argv, ACE_TEXT ("p:r:s:"), 0);
 288   get_opt.long_option (ACE_TEXT ("client_port"), 'p',
 289                        ACE_Get_Opt::ARG_REQUIRED);
 290   get_opt.long_option (ACE_TEXT ("server_port"), 'r',
 291                        ACE_Get_Opt::ARG_REQUIRED);
 292   get_opt.long_option (ACE_TEXT ("server_name"), 's',
 293                        ACE_Get_Opt::ARG_REQUIRED);
 294
 295   for (int c; (c = get_opt ()) != -1;)
 296     switch (c) {
 297     case 'p': // Client logging daemon acceptor port number.
 298       cld_port = static_cast<u_short> (ACE_OS::atoi (get_opt.opt_arg ()));
 299       break;
 300     case 'r': // Server logging daemon acceptor port number.
 301       sld_port = static_cast<u_short> (ACE_OS::atoi (get_opt.opt_arg ()));
 302       break;
 303     case 's': // Server logging daemon hostname.
 304       ACE_OS::strsncpy
 305         (sld_host, get_opt.opt_arg (), MAXHOSTNAMELEN);
 306       break;
 307     }
 308
 309   if (cld_addr_.set (cld_port) == -1 ||
 310       sld_addr_.set (sld_port, sld_host) == -1)
 311     return -1;
 312   return activate ();
 313 }
 314
 315 int AIO_Client_Logging_Daemon::fini () {
 316   ACE_Proactor::instance ()->proactor_end_event_loop ();
 317   this->wait ();
 318   return 0;
 319 }
 320
 321 int AIO_Client_Logging_Daemon::svc () {
 322   if (acceptor_.open (cld_addr_) == -1) return -1;
 323   if (CLD_CONNECTOR::instance ()->connect (sld_addr_) == 0)
 324     ACE_Proactor::instance ()->proactor_run_event_loop ();
 325   acceptor_.close ();
 326   CLD_CONNECTOR::close ();
 327   OUTPUT_HANDLER::close ();
 328   return 0;
 329 }
 330
 331 #else   /* There's no AIO support on this platform */
 332
 333 #include "ace/Task.h"
 334 #include "ace/Service_Object.h"
 335 #include "ace/Synch_Traits.h"
 336 #include "AIO_CLD_export.h"
 337
 338 class AIO_Client_Logging_Daemon : public ACE_Task<ACE_NULL_SYNCH> {
 339 public:
 340   // Service Configurator hook methods.
 341   virtual int init (int, ACE_TCHAR *[]);
 342   virtual int fini ();
 343 };
 344
 345 int AIO_Client_Logging_Daemon::init (int, ACE_TCHAR *[]) {
 346   ACE_ERROR_RETURN
 347     ((LM_ERROR, ACE_TEXT ("This service requires AIO support\n")), -1);
 348 }
 349
 350 int AIO_Client_Logging_Daemon::fini () {
 351   return 0;
 352 }
 353
 354 #endif /* ACE_WIN32 || ACE_HAS_AIO_CALLS */
 355
 356 ACE_FACTORY_DEFINE (AIO_CLD, AIO_Client_Logging_Daemon)
 357
 358