ACE/examples/ASX/UPIPE_Event_Server/event_server.cpp

   1 // Test the event server.
   2
   3 #include "ace/OS_main.h"
   4 #include "ace/Stream.h"
   5 #include "ace/Service_Config.h"
   6 #include "ace/UPIPE_Acceptor.h"
   7 #include "ace/UPIPE_Connector.h"
   8 #include "ace/Truncate.h"
   9
  10 // FUZZ: disable check_for_streams_include
  11 #include "ace/streams.h"
  12
  13 #include "Options.h"
  14 #include "Consumer_Router.h"
  15 #include "Event_Analyzer.h"
  16 #include "Supplier_Router.h"
  17 #include "ace/Sig_Adapter.h"
  18 #include "ace/OS_NS_unistd.h"
  19
  20 #if defined (ACE_HAS_THREADS)
  21
  22 typedef ACE_Stream<ACE_MT_SYNCH> MT_Stream;
  23 typedef ACE_Module<ACE_MT_SYNCH> MT_Module;
  24
  25 // Handle SIGINT and terminate the entire application.
  26
  27 class Quit_Handler : public ACE_Sig_Adapter
  28 {
  29 public:
  30   Quit_Handler ();
  31   virtual int handle_input (ACE_HANDLE fd);
  32 };
  33
  34 Quit_Handler::Quit_Handler ()
  35   : ACE_Sig_Adapter (ACE_Sig_Handler_Ex (ACE_Reactor::end_event_loop))
  36 {
  37   // Register to trap input from the user.
  38   if (ACE_Event_Handler::register_stdin_handler (this,
  39                                                  ACE_Reactor::instance (),
  40                                                  ACE_Thread_Manager::instance ()) == -1)
  41     ACE_ERROR ((LM_ERROR, "%p\n", "register_stdin_handler"));
  42   // Register to trap the SIGINT signal.
  43   else if (ACE_Reactor::instance ()->register_handler
  44            (SIGINT, this) == -1)
  45     ACE_ERROR ((LM_ERROR, "%p\n", "register_handler"));
  46 }
  47
  48 int
  49 Quit_Handler::handle_input (ACE_HANDLE)
  50 {
  51   options.stop_timer ();
  52   ACE_DEBUG ((LM_INFO, " (%t) closing down the test\n"));
  53   options.print_results ();
  54
  55   ACE_Reactor::end_event_loop();
  56   return 0;
  57 }
  58
  59 static void *
  60 consumer (void *)
  61 {
  62   ACE_UPIPE_Stream c_stream;
  63   ACE_UPIPE_Addr c_addr (ACE_TEXT ("/tmp/conupipe"));
  64
  65   int verb = options.verbose ();
  66   int msiz = ACE_Utils::truncate_cast<int> (options.message_size ());
  67   time_t secs, par1, par2;
  68   time_t currsec;
  69
  70   if (verb)
  71     cout << "consumer starting connect" << endl;
  72
  73   ACE_UPIPE_Connector con;
  74
  75   if (con.connect (c_stream, c_addr) == -1)
  76     ACE_DEBUG ((LM_INFO, " (%t) connect failed\n"));
  77   else
  78     cout << "consumer :we're connected" << endl;
  79
  80   ACE_Message_Block *mb_p;
  81
  82   int done = 0;
  83   int cnt = 0;
  84   ACE_OS::time (&currsec);
  85
  86   par1 = currsec;
  87
  88   while (done == 0 && (c_stream.recv (mb_p) != -1))
  89     {
  90       if (mb_p->length () > 1)
  91         {
  92           cnt++;
  93           if (verb)
  94             {
  95               cout << " consumer received message !!!!!! "
  96                   << mb_p->rd_ptr () << endl;
  97             }
  98         }
  99       else
 100         {
 101           if (verb)
 102             {
 103               cout << "consumer got last mb"
 104                   << (char) * (mb_p->rd_ptr ()) << endl;
 105             }
 106           c_stream.close ();
 107           done = 1;
 108         }
 109     }
 110
 111     ACE_OS::time (&currsec);
 112     par2 = currsec;
 113
 114     secs = par2 - par1;
 115
 116     if (secs <= 0)
 117       secs=1;
 118
 119     ACE_DEBUG ((LM_INFO,
 120                 ACE_TEXT ("consumer got %d messages of size %d ")
 121                 ACE_TEXT ("within %: seconds\n"),
 122                 cnt, msiz, secs));
 123
 124     ACE_OS::sleep (2);
 125     cout << "consumer terminating " << endl;
 126     return 0;
 127 }
 128
 129 static void *
 130 supplier (void *dummy)
 131 {
 132   ACE_UPIPE_Stream s_stream;
 133   ACE_UPIPE_Addr serv_addr (ACE_TEXT ("/tmp/supupipe"));
 134   ACE_UPIPE_Connector con;
 135
 136   int iter = ACE_Utils::truncate_cast<int> (options.iterations ());
 137   int verb = options.verbose ();
 138   int msiz = ACE_Utils::truncate_cast<int> (options.message_size ());
 139   cout << "supplier starting connect" << endl;
 140
 141   if (con.connect (s_stream, serv_addr) == -1)
 142     ACE_DEBUG ((LM_INFO, " (%t) connect failed\n"));
 143
 144   cout << "supplier : we're connected" << endl;
 145   int      n;
 146   n = 0;
 147   ACE_Message_Block * mb_p;
 148
 149   while (n < iter)
 150     {
 151       mb_p = new ACE_Message_Block (msiz);
 152       ACE_OS::strcpy (mb_p->rd_ptr (), (char *) dummy);
 153       mb_p->length (msiz);
 154       if (verb)
 155         cout << "supplier sending 1 message_block" << endl;
 156       if (s_stream.send (mb_p) == -1)
 157         {
 158           cout << "supplier send failed" << endl;
 159           return (void *) -1;
 160         }
 161       n++;
 162     }
 163
 164   mb_p = new ACE_Message_Block (10);
 165   mb_p->length (1);
 166   *mb_p->rd_ptr () = 'g';
 167
 168   cout << "supplier sending last message_block" << endl;
 169
 170   if (s_stream.send (mb_p) == -1)
 171     {
 172       cout << "supplier send last mb failed" << endl;
 173       return (void *) -1;
 174     }
 175   mb_p = new ACE_Message_Block (10);
 176   mb_p->length (0);
 177
 178   if (verb)
 179     cout << "supplier sending very last message_block" << endl;
 180
 181   if (s_stream.send (mb_p) == -1)
 182     {
 183       cout << "supplier send very last mb failed" << endl;
 184       return (void *) -1;
 185     }
 186
 187   ACE_OS::sleep (2);
 188   cout << "supplier terminating" << endl;
 189   return 0;
 190 }
 191
 192 int
 193 ACE_TMAIN (int argc, ACE_TCHAR *argv[])
 194 {
 195   ACE_Service_Config daemon;
 196
 197   options.parse_args (argc, argv);
 198   options.start_timer ();
 199
 200   // Primary ACE_Stream for EVENT_SERVER application.
 201   MT_Stream event_server;
 202
 203   // Enable graceful shutdowns....
 204   Quit_Handler quit_handler;
 205
 206   // Create the modules..
 207
 208   MT_Module *sr = new MT_Module (ACE_TEXT ("Supplier_Router"),
 209           new Supplier_Router (ACE_Thread_Manager::instance ()));
 210   MT_Module *ea = new MT_Module (ACE_TEXT ("Event_Analyzer"),
 211                                  new Event_Analyzer,
 212                                  new Event_Analyzer);
 213   MT_Module *cr = new MT_Module (ACE_TEXT ("Consumer_Router"),
 214                                  0, // 0 triggers the creation of a ACE_Thru_Task...
 215                                  new Consumer_Router (ACE_Thread_Manager::instance ()));
 216
 217   // Push the modules onto the event_server stream.
 218
 219   if (event_server.push (sr) == -1)
 220     ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"),
 221                        ACE_TEXT ("push (Supplier_Router)")), -1);
 222
 223   if (event_server.push (ea) == -1)
 224     ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"),
 225                        ACE_TEXT ("push (Event_Analyzer)")), -1);
 226
 227   if (event_server.push (cr) == -1)
 228     ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"),
 229                        ACE_TEXT ("push (Consumer_Router)")), -1);
 230
 231   // Set the high and low water marks appropriately.
 232
 233   int wm = ACE_Utils::truncate_cast<int> (options.low_water_mark ());
 234
 235   if (event_server.control (ACE_IO_Cntl_Msg::SET_LWM, &wm) == -1)
 236     ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"),
 237                        ACE_TEXT ("push (setting low watermark)")), -1);
 238
 239   wm = ACE_Utils::truncate_cast<int> (options.high_water_mark ());
 240
 241   if (event_server.control (ACE_IO_Cntl_Msg::SET_HWM, &wm) == -1)
 242     ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"),
 243                        ACE_TEXT ("push (setting high watermark)")), -1);
 244
 245   // spawn the two threads.
 246
 247   if (ACE_Thread_Manager::instance ()->spawn (ACE_THR_FUNC (consumer), (void *) 0,
 248                                              THR_NEW_LWP | THR_DETACHED) == -1)
 249     ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("spawn")), 1);
 250
 251   else if (ACE_Thread_Manager::instance ()->spawn (ACE_THR_FUNC (supplier), (void *) "hello",
 252                                                   THR_NEW_LWP | THR_DETACHED) == -1)
 253     ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("spawn")), 1);
 254
 255   // Perform the main event loop waiting for the user to type ^C or to
 256   // enter a line on the ACE_STDIN.
 257
 258   ACE_Reactor::instance ()->run_reactor_event_loop ();
 259
 260   ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("main exiting\n")));
 261
 262   return 0;
 263 }
 264 #else
 265 int
 266 ACE_TMAIN (int, ACE_TCHAR *[])
 267 {
 268   ACE_ERROR_RETURN ((LM_ERROR,
 269                      ACE_TEXT ("test not defined for this platform\n")),
 270                     -1);
 271 }
 272 #endif /* ACE_HAS_THREADS */