ACE/examples/IPC_SAP/UPIPE_SAP/ex1.cpp

   1
   2 //=============================================================================
   3 /**
   4  *  @file   ex1.cpp
   5  *
   6  *  Example for using <ACE_UPIPE_SAP> and <ACE_Thread> for
   7  *  intra-process communication.
   8  *
   9  *  @author Gerhard Lenzer and Douglas C. Schmidt
  10  */
  11 //=============================================================================
  12
  13
  14 #include "ace/OS_main.h"
  15 #include "ace/OS_NS_string.h"
  16 #include "ace/Stream.h"
  17 #include "ace/UPIPE_Acceptor.h"
  18 #include "ace/UPIPE_Connector.h"
  19
  20
  21 #if defined (ACE_HAS_THREADS)
  22
  23 // Global pattern
  24 static ACE_UPIPE_Addr addr (ACE_TEXT("pattern"));
  25
  26 // peer1 thread entry point.
  27
  28 static void *
  29 peer1 (void *)
  30 {
  31   ACE_UPIPE_Stream c_stream;
  32
  33   ACE_DEBUG ((LM_DEBUG,
  34               "(%t) peer1 starting connect\n"));
  35   ACE_UPIPE_Connector con;
  36
  37   if (con.connect (c_stream, addr) == -1)
  38     ACE_ERROR ((LM_ERROR,
  39                 "(%t) peer1 ACE_UPIPE_Connector failed\n"));
  40
  41   ACE_Message_Block *mb = 0;
  42   ACE_NEW_RETURN (mb,
  43                   ACE_Message_Block (20),
  44                   0);
  45
  46   mb->copy ("hello", 6);
  47
  48   if (c_stream.send (mb) == -1)
  49     ACE_ERROR ((LM_ERROR,
  50                 "(%t) error peer1 send\n"));
  51
  52   if (c_stream.recv (mb) == -1)
  53     ACE_ERROR ((LM_ERROR,
  54                 "(%t) error peer1 recv\n"));
  55
  56   ACE_ERROR ((LM_ERROR,
  57               "(%t) peer1 ack is \"%s\"\n",
  58               mb->rd_ptr ()));
  59
  60   // Free up the memory block.
  61   mb->release ();
  62
  63   // Now try the send()/recv() interface.
  64   char mytext[] = "This string is sent by peer1 as buffer";
  65
  66   ACE_ERROR ((LM_ERROR,
  67               "(%t) peer1 sending text\n"));
  68   if (c_stream.send (mytext, sizeof mytext) == -1)
  69     ACE_ERROR ((LM_ERROR,
  70                 "(%t) buffer send from peer1 failed\n"));
  71
  72   char conbuf[30];  // Buffer to receive response.
  73
  74   int i = 0;
  75
  76   for (char c = ' '; c != '!'; i++)
  77     {
  78       if (c_stream.recv (&c, 1) == -1)
  79         ACE_ERROR ((LM_ERROR,
  80                     "(%t) buffer recv from peer1 failed\n"));
  81       else
  82         conbuf[i] = c;
  83     }
  84
  85   conbuf[i] = '\0';
  86   ACE_DEBUG ((LM_DEBUG,
  87               "(%t) peer1 received buffer with \"%s\"\n",
  88               conbuf));
  89   c_stream.close ();
  90   return 0;
  91 }
  92
  93 static void *
  94 peer2 (void *)
  95 {
  96   ACE_UPIPE_Acceptor acc (addr);
  97   ACE_UPIPE_Stream s_stream;
  98
  99   // Spawn a peer1 thread.
 100   if (ACE_Thread_Manager::instance ()->spawn (ACE_THR_FUNC (peer1),
 101                                               (void *) 0,
 102                                               THR_NEW_LWP | THR_DETACHED) == -1)
 103     ACE_ERROR_RETURN ((LM_ERROR,
 104                        "%p\n",
 105                        "spawn"),
 106                       0);
 107
 108   ACE_DEBUG ((LM_DEBUG,
 109               "(%t) peer2 starting accept\n"));
 110
 111   if (acc.accept (s_stream) == -1)
 112     ACE_ERROR ((LM_ERROR,
 113                 "(%t) ACE_UPIPE_Acceptor.accept failed\n"));
 114
 115   ACE_Message_Block *mb = 0;
 116
 117   if (s_stream.recv (mb) == -1)
 118     ACE_ERROR ((LM_ERROR,
 119                 "(%t) peer2 recv failed\n"));
 120
 121   ACE_DEBUG ((LM_DEBUG, "(%t) peer2 recv is \"%s\"\n",
 122               mb->rd_ptr ()));
 123
 124   mb->wr_ptr (mb->rd_ptr ());
 125   mb->copy ("thanks", 7);
 126
 127   if (s_stream.send (mb) == -1)
 128     ACE_ERROR ((LM_ERROR,
 129                 "(%t) peer2 send failed\n"));
 130
 131   char s_buf[42];
 132   ACE_DEBUG ((LM_DEBUG,
 133               "(%t) peer2 sleeping on recv\n"));
 134
 135   if (s_stream.recv (s_buf, sizeof s_buf) == -1)
 136     ACE_ERROR ((LM_ERROR,
 137                 "(%t) peer2 recv failed\n"));
 138   else
 139     ACE_DEBUG ((LM_DEBUG,
 140                 "(%t) peer2 received buffer with \"%s\"\n",
 141                 s_buf));
 142
 143   ACE_OS::strcpy (s_buf,
 144                   "this is the peer2 response!");
 145
 146   if (s_stream.send (s_buf, 30) == -1)
 147     ACE_ERROR ((LM_ERROR,
 148                 "(%t) peer2 send failed\n"));
 149   s_stream.close ();
 150   return 0;
 151 }
 152
 153 int
 154 ACE_TMAIN (int, ACE_TCHAR *[])
 155 {
 156   // Spawn a peer2 thread.
 157   if (ACE_Thread_Manager::instance ()->spawn (ACE_THR_FUNC (peer2),
 158                                               (void *) 0,
 159                                               THR_NEW_LWP | THR_DETACHED) == -1)
 160     ACE_ERROR_RETURN ((LM_ERROR,
 161                        "%p\n",
 162                        "spawn"),
 163                       1);
 164   // Wait for peer2 and peer1 threads to exit.
 165   ACE_Thread_Manager::instance ()->wait ();
 166   return 0;
 167 }
 168 #else
 169 int
 170 ACE_TMAIN (int, ACE_TCHAR *[])
 171 {
 172   ACE_ERROR_RETURN ((LM_ERROR,
 173                      "threads not supported on this platform\n"),
 174                     -1);
 175 }
 176 #endif /* ACE_HAS_THREADS */