ACE/examples/Reactor/Ntalker/ntalker.cpp

   1 // Listens to multicast address.  After first message received, will
   2 // listen for 5 more seconds.  Prints Mbits/sec received from client.
   3
   4 #include "ace/OS_main.h"
   5 #include "ace/OS_NS_unistd.h"
   6 #include "ace/INET_Addr.h"
   7 #include "ace/SOCK_Dgram_Mcast.h"
   8 #include "ace/Reactor.h"
   9 #include "ace/Get_Opt.h"
  10 #include "ace/Thread_Manager.h"
  11 #include "ace/Service_Config.h"
  12
  13 #if defined (ACE_HAS_IP_MULTICAST)
  14 // Network interface to subscribe to.  This is hardware specific.  use
  15 // netstat(1M) to find whether your interface is le0 or ie0
  16
  17 static const ACE_TCHAR *INTERFACE = 0;
  18 static const char *MCAST_ADDR = ACE_DEFAULT_MULTICAST_ADDR;
  19 static const u_short UDP_PORT = ACE_DEFAULT_MULTICAST_PORT;
  20
  21 class Handler : public ACE_Event_Handler
  22 {
  23   // = TITLE
  24   //     Handle both multicast and stdin events.
  25 public:
  26   Handler (u_short udp_port,
  27            const char *ip_addr,
  28            const ACE_TCHAR *a_interface,
  29            ACE_Reactor & );
  30   // Constructor.
  31
  32   ~Handler ();
  33   // Destructor.
  34
  35   // Event demuxer hooks.
  36   virtual int handle_input (ACE_HANDLE);
  37   virtual int handle_close (ACE_HANDLE,
  38                             ACE_Reactor_Mask);
  39   virtual ACE_HANDLE get_handle () const;
  40
  41 private:
  42   ACE_SOCK_Dgram_Mcast mcast_;
  43   // Multicast wrapper.
  44
  45   ACE_INET_Addr sockmc_addr_;
  46   // Address to multicast to.
  47 };
  48
  49 ACE_HANDLE
  50 Handler::get_handle () const
  51 {
  52   return this->mcast_.get_handle ();
  53 }
  54
  55 int
  56 Handler::handle_input (ACE_HANDLE h)
  57 {
  58   char buf[BUFSIZ];
  59
  60   if (h == ACE_STDIN)
  61     {
  62       ssize_t result = ACE_OS::read (h, buf, BUFSIZ);
  63
  64       if (result > 0)
  65         {
  66           if (this->mcast_.send (buf, result) != result)
  67             ACE_ERROR_RETURN ((LM_ERROR,
  68                                "%p\n",
  69                                "send error"),
  70                               -1);
  71           return 0;
  72         }
  73       else if (result == -1)
  74             ACE_ERROR_RETURN ((LM_ERROR,
  75                                "%p\n",
  76                                "can't read from STDIN"),
  77                               -1);
  78       else // result == 0
  79         {
  80           ACE_Reactor::end_event_loop ();
  81           return -1;
  82         }
  83     }
  84   else
  85     {
  86       ACE_INET_Addr remote_addr;
  87
  88       // Receive message from multicast group.
  89       ssize_t result = this->mcast_.recv (buf,
  90                                           sizeof buf,
  91                                           remote_addr);
  92
  93       if (result != -1)
  94         {
  95           ACE_DEBUG ((LM_DEBUG,
  96                       "received datagram from host %s on port %d bytes = %d\n",
  97                       remote_addr.get_host_name (),
  98                       remote_addr.get_port_number (),
  99                       result));
 100           ACE_OS::write (ACE_STDERR, buf, result);
 101           ACE_DEBUG ((LM_DEBUG,
 102                       "\n"));
 103           return 0;
 104         }
 105
 106       ACE_ERROR_RETURN ((LM_ERROR,
 107                          "%p\n",
 108                          "something amiss"),
 109                               -1);
 110     }
 111 }
 112
 113 int
 114 Handler::handle_close (ACE_HANDLE h, ACE_Reactor_Mask)
 115 {
 116   if (h == ACE_STDIN)
 117     {
 118       ACE_DEBUG ((LM_DEBUG,
 119                   "STDIN_Events handle removed from reactor.\n"));
 120       if (ACE_Reactor::instance ()->remove_handler
 121           (this, ACE_Event_Handler::READ_MASK) == -1)
 122         ACE_ERROR_RETURN ((LM_ERROR,
 123                            "%p\n",
 124                            "remove_handler"),
 125                            -1);
 126     }
 127   else
 128     ACE_DEBUG ((LM_DEBUG,
 129                 "Mcast_Events handle removed from reactor.\n"));
 130   return 0;
 131 }
 132
 133 Handler::~Handler ()
 134 {
 135   if (this->mcast_.leave (sockmc_addr_) == -1)
 136     ACE_ERROR ((LM_ERROR,
 137                 "%p\n",
 138                 "leave fails"));
 139 }
 140
 141 Handler::Handler (u_short udp_port,
 142                   const char *ip_addr,
 143                   const ACE_TCHAR *a_interface,
 144                   ACE_Reactor &reactor)
 145 {
 146   // Create multicast address to listen on.
 147
 148   this->sockmc_addr_ = ACE_INET_Addr (udp_port, ip_addr);
 149
 150   // subscribe to multicast group.
 151
 152   if (this->mcast_.join (sockmc_addr_, 1, a_interface) == -1)
 153     ACE_ERROR ((LM_ERROR,
 154                 "%p\n",
 155                 "can't subscribe to multicast group"));
 156
 157   // Disable loopbacks.
 158   // if (this->mcast_.set_option (IP_MULTICAST_LOOP, 0) == -1 )
 159   //   ACE_OS::perror (" can't disable loopbacks " ), ACE_OS::exit (1);
 160
 161   // Register callbacks with the ACE_Reactor.
 162   else if (reactor.register_handler (this->mcast_.get_handle (),
 163                                      this,
 164                                      ACE_Event_Handler::READ_MASK) == -1)
 165     ACE_ERROR ((LM_ERROR,
 166                 "%p\n",
 167                 "can't register with Reactor\n"));
 168   // Register the STDIN handler.
 169   else if (ACE_Event_Handler::register_stdin_handler (this,
 170                                                       ACE_Reactor::instance (),
 171                                                       ACE_Thread_Manager::instance ()) == -1)
 172     ACE_ERROR ((LM_ERROR,
 173                 "%p\n",
 174                 "register_stdin_handler"));
 175 }
 176
 177 static void
 178 parse_args (int argc, ACE_TCHAR *argv[])
 179 {
 180   ACE_Get_Opt get_opt (argc, argv, ACE_TEXT("i:u"));
 181
 182   int c;
 183
 184   while ((c = get_opt ()) != -1)
 185     switch (c)
 186       {
 187       case 'i':
 188         INTERFACE = get_opt.opt_arg ();
 189         break;
 190       case 'u':
 191         // usage same as unknown.
 192         ACE_FALLTHROUGH;
 193       default:
 194         ACE_DEBUG ((LM_DEBUG,
 195                     "%s -i interface\n",
 196                     argv[0]));
 197         ACE_OS::exit (1);
 198       }
 199 }
 200
 201 int
 202 ACE_TMAIN (int argc, ACE_TCHAR *argv[])
 203 {
 204   parse_args (argc, argv);
 205
 206   Handler handler (UDP_PORT,
 207                    MCAST_ADDR,
 208                    INTERFACE,
 209                    *ACE_Reactor::instance ());
 210
 211   // Run the event loop.
 212   ACE_Reactor::run_event_loop ();
 213
 214   ACE_DEBUG ((LM_DEBUG,
 215               "talker Done.\n"));
 216   return 0;
 217 }
 218
 219 #else
 220 int
 221 ACE_TMAIN (int, ACE_TCHAR *argv[])
 222 {
 223   ACE_ERROR_RETURN ((LM_ERROR,
 224                      "error: %s must be run on a platform that support IP multicast\n",
 225                      argv[0]),
 226                     0);
 227 }
 228 #endif /* ACE_HAS_IP_MULTICAST */
 229