ACE/protocols/tests/HTBP/Reactor_Tests/server.cpp

   1 /**
   2  * server for a reactor based connection establishment test using HTBP
   3  */
   4
   5 #include "ace/Log_Msg.h"
   6
   7 #include "ace/HTBP/HTBP_Session.h"
   8 #include "ace/HTBP/HTBP_Stream.h"
   9 #include "ace/HTBP/HTBP_Addr.h"
  10
  11 #include "ace/SOCK_Acceptor.h"
  12 #include "ace/SOCK_Stream.h"
  13 #include "ace/Event_Handler.h"
  14 #include "ace/Reactor.h"
  15 #include "ace/Get_Opt.h"
  16 #include "ace/OS_NS_stdio.h"
  17 #include "ace/OS_NS_unistd.h"
  18 #include "ace/OS_NS_sys_socket.h"
  19 #include "ace/os_include/os_netdb.h"
  20
  21 unsigned port = 8088;
  22 const ACE_TCHAR *notifier_file = 0;
  23
  24 int
  25 parse_args (int argc, ACE_TCHAR *argv[])
  26 {
  27   ACE_Get_Opt get_opts (argc, argv, ACE_TEXT("o:p:"));
  28   int c;
  29
  30   while ((c = get_opts ()) != -1)
  31     switch (c)
  32       {
  33       case 'o':
  34         notifier_file = get_opts.opt_arg();
  35         break;
  36
  37       case 'p':
  38         port = static_cast<unsigned>(ACE_OS::atoi (get_opts.opt_arg()));
  39         break;
  40       case '?':
  41       default:
  42         ACE_ERROR_RETURN ((LM_ERROR,
  43                            "usage:  %s "
  44                            "-p port "
  45                             "\n",
  46                            argv [0]),
  47                           -1);
  48       }
  49   // Indicates successful parsing of the command line
  50   return 0;
  51 }
  52
  53 class Accept_Handler : public ACE_Event_Handler
  54 {
  55 public:
  56   Accept_Handler (ACE_SOCK_Acceptor& a);
  57   virtual ~Accept_Handler ();
  58   virtual int handle_input (ACE_HANDLE );
  59 private:
  60   ACE_SOCK_Acceptor& acceptor_;
  61   ACE::HTBP::Channel *channels_[2];
  62 };
  63
  64 class Stream_Handler : public ACE_Event_Handler
  65 {
  66 public:
  67   Stream_Handler (ACE::HTBP::Stream &s);
  68   virtual ~Stream_Handler ();
  69   virtual int handle_input (ACE_HANDLE );
  70 private:
  71   ACE::HTBP::Stream &stream_;
  72 };
  73
  74
  75 Accept_Handler::Accept_Handler(ACE_SOCK_Acceptor &a)
  76   :ACE_Event_Handler(),
  77    acceptor_(a)
  78 {
  79   this->channels_[0] = this->channels_[1] = 0;
  80   if (this->reactor() == 0)
  81     this->reactor(ACE_Reactor::instance());
  82   this->reactor()->register_handler (acceptor_.get_handle(),
  83                                      this,
  84                                      ACE_Event_Handler::ACCEPT_MASK);
  85 }
  86
  87 Accept_Handler::~Accept_Handler()
  88 {
  89   this->reactor()->remove_handler (acceptor_.get_handle(),
  90                                    ACE_Event_Handler::ACCEPT_MASK|
  91                                    ACE_Event_Handler::DONT_CALL);
  92   acceptor_.close();
  93 }
  94
  95 int
  96 Accept_Handler::handle_input (ACE_HANDLE h)
  97 {
  98   ACE::HTBP::Channel **ch = 0;
  99   if (h == acceptor_.get_handle())
 100     {
 101       ACE_SOCK_Stream *sock = new ACE_SOCK_Stream;
 102       acceptor_.accept(*sock);
 103       ch = channels_[0] == 0 ? &channels_[0] :& channels_[1];
 104       *ch = new ACE::HTBP::Channel(*sock);
 105       this->reactor()->register_handler (sock->get_handle(),
 106                                          this,
 107                                          ACE_Event_Handler::READ_MASK);
 108       return 0;
 109     }
 110   for (int i = 0; i < 2; i++)
 111     if (channels_[i] && channels_[i]->get_handle() == h)
 112       {
 113         ch = &channels_[i];
 114         break;
 115       }
 116   if (ch == 0)
 117     ACE_ERROR_RETURN ((LM_ERROR,
 118                        ACE_TEXT ("(%P|%t) Server Accept_Handler::handle_input, ")
 119                        ACE_TEXT ("unknown handle %d\n") ,h),
 120                       -1);
 121   int result = (*ch)->pre_recv();
 122   if (result == 0)
 123     {
 124       this->reactor()->remove_handler (h,
 125                                       ACE_Event_Handler::READ_MASK |
 126                                       ACE_Event_Handler::DONT_CALL);
 127
 128       (*ch)->register_notifier(this->reactor());
 129       ACE::HTBP::Session *session = (*ch)->session();
 130
 131       ACE::HTBP::Stream *stream = new ACE::HTBP::Stream(session);
 132       ACE_Event_Handler *handler = session->handler();
 133
 134       if (handler == 0)
 135         {
 136           ACE_DEBUG ((LM_DEBUG,
 137                       ACE_TEXT ("(%P|%t) Server Accept_Handler::handle_input ")
 138                       ACE_TEXT ("Creating new stream handler for %d\n"),
 139                       stream->get_handle()));
 140           Stream_Handler *sh = new Stream_Handler(*stream);
 141           session->handler (sh);
 142         }
 143       else
 144         ACE_DEBUG ((LM_DEBUG,
 145                     ACE_TEXT ("(%P|%t) Server Accept_Handler::handle_input ")
 146                     ACE_TEXT ("There is already a handler for %d\n"),
 147                     stream->get_handle()));
 148
 149       if ((*ch)->state() == ACE::HTBP::Channel::Data_Queued)
 150         {
 151           ACE_DEBUG ((LM_DEBUG,
 152                       ACE_TEXT ("(%P|%t) Server Accept_Handler::handle_input\n"),
 153                       ACE_TEXT ("Issuing notification on handler\n")));
 154           this->reactor()->notify (session->handler(),
 155                                    ACE_Event_Handler::READ_MASK);
 156         }
 157
 158       *ch = 0;
 159     }
 160   return 0;
 161 }
 162
 163 Stream_Handler::Stream_Handler (ACE::HTBP::Stream &s)
 164   :stream_(s)
 165 {}
 166 Stream_Handler::~Stream_Handler ()
 167 {
 168 }
 169
 170 int
 171 Stream_Handler::handle_input (ACE_HANDLE h)
 172 {
 173   char buffer[1000];
 174   ssize_t n = this->stream_.recv (buffer,1000);
 175   if (n == -1)
 176     ACE_ERROR_RETURN ((LM_ERROR,
 177                        ACE_TEXT ("(%P|%t) Server Stream_Handler::handle_input %p\n"),
 178                        ACE_TEXT ("recv")),
 179                       0);
 180   buffer[n] = 0;
 181   ACE_DEBUG ((LM_DEBUG,
 182               ACE_TEXT ("(%P|%t) Server Stream_Handler::handle_input ")
 183               ACE_TEXT (" (%d) read %b:\n%C\n"),
 184               h, n, buffer));
 185
 186   const char *tok_loc = ACE_OS::strstr (buffer, "goodbye");
 187   if (tok_loc != 0)
 188     this->reactor()->end_event_loop();
 189   else
 190     {
 191       ACE::HTBP::Channel *ch = stream_.session()->outbound();
 192       if (ch != 0)
 193         ACE_DEBUG ((LM_DEBUG,
 194                     ACE_TEXT ("(%P|%t) Server Stream_Handler::handle_input ")
 195                     ACE_TEXT ("Sending reply on %d\n"),
 196                     ch->ace_stream().get_handle()));
 197       else
 198         ACE_DEBUG ((LM_DEBUG,
 199                     ACE_TEXT ("(%P|%t) Server Stream_Handler::handle_input ")
 200                     ACE_TEXT ("Can't send reply on nul channel\n")));
 201       this->stream_.send ("Back atcha!",11);
 202     }
 203   return 0;
 204 }
 205
 206 int
 207 ACE_TMAIN (int argc, ACE_TCHAR * argv[])
 208 {
 209   ACE_DEBUG ((LM_DEBUG,
 210               ACE_TEXT ("(%P|%t) Server: ")
 211               ACE_TEXT ("At start of main\n")));
 212   ACE_OS::socket_init (ACE_WSOCK_VERSION);
 213
 214   if (parse_args (argc, argv) != 0)
 215     return 1;
 216
 217   ACE_TCHAR host[MAXHOSTNAMELEN+1];
 218   if (ACE_OS::hostname (host, MAXHOSTNAMELEN) != 0)
 219     ACE_ERROR_RETURN ((LM_ERROR,
 220                        ACE_TEXT ("(%P|%t) Server failure: %p\n"),
 221                        ACE_TEXT ("hostname")),
 222                       1);
 223
 224   ACE_INET_Addr local (port, host);
 225   local.addr_to_string (host, MAXHOSTNAMELEN);
 226   ACE_DEBUG ((LM_DEBUG,
 227               ACE_TEXT ("(%P|%t) Server: ")
 228               ACE_TEXT ("listening at %s\n"),
 229               host));
 230
 231   ACE_SOCK_Acceptor acc (local, 1);
 232   ACE_DEBUG ((LM_DEBUG,
 233               ACE_TEXT ("(%P|%t) Server: ")
 234               ACE_TEXT ("opened listener\n")));
 235
 236   Accept_Handler handler (acc);
 237   ACE_DEBUG ((LM_DEBUG,
 238               ACE_TEXT ("(%P|%t) Server: ")
 239               ACE_TEXT ("server is ready\n")));
 240
 241   if (notifier_file != 0)
 242     {
 243       FILE *f = ACE_OS::fopen (notifier_file,ACE_TEXT("w+"));
 244       const char *msg = "server ready";
 245       ACE_OS::fwrite (msg,ACE_OS::strlen(msg),1,f);
 246       ACE_OS::fclose (f);
 247     }
 248
 249   ACE_Reactor::instance()->run_reactor_event_loop();
 250   return 0;
 251 }