ACE/protocols/tests/RMCast/Receiver.cpp

   1 // file      : Receiver.cpp
   2 // author    : Boris Kolpackov <boris@kolpackov.net>
   3 #include "ace/Vector_T.h"
   4 #include "ace/Log_Msg.h"
   5 #include "ace/OS_NS_string.h"
   6
   7 #include "ace/RMCast/Socket.h"
   8
   9 #include "Protocol.h"
  10
  11 typedef
  12 ACE_Vector<unsigned char, ACE_VECTOR_DEFAULT_SIZE>
  13 Status_List;
  14
  15 class args {};
  16
  17 int
  18 ACE_TMAIN (int argc, ACE_TCHAR* argv[])
  19 {
  20   try
  21   {
  22     if (argc < 2) throw args ();
  23
  24     ACE_INET_Addr addr (argv[1]);
  25
  26     //FUZZ: disable check_for_lack_ACE_OS
  27     // Turn on message loss and reordering simulation.
  28     //
  29     ACE_RMCast::Socket socket (addr, false, true);
  30     //FUZZ: enable check_for_lack_ACE_OS
  31
  32
  33     Message expected_msg;
  34     expected_msg.sn = 0;
  35
  36     // VC6 does not know about the new rules.
  37     //
  38     {
  39       for (unsigned short i = 0; i < payload_size; i++)
  40       {
  41         expected_msg.payload[i] = i;
  42       }
  43     }
  44
  45     Status_List received (message_count);
  46     Status_List damaged (message_count);
  47     Status_List duplicate (message_count);
  48
  49     // VC6 does not know about new rules.
  50     //
  51     {
  52       for (unsigned long i = 0; i < message_count; ++i)
  53       {
  54         received.push_back (0);
  55         damaged.push_back (0);
  56         duplicate.push_back (0);
  57       }
  58     }
  59
  60
  61     Message msg;
  62
  63     while (true)
  64     {
  65       ssize_t s = socket.size ();
  66
  67       if (s == -1 && errno == ENOENT)
  68       {
  69         ACE_ERROR ((LM_ERROR, "unavailable message detected\n"));
  70
  71         // Receive it.
  72         //
  73         socket.recv (&msg, sizeof (msg));
  74
  75         continue;
  76       }
  77
  78       if (s != (ssize_t)sizeof (msg))
  79       {
  80         ACE_ERROR ((LM_ERROR, "unexpected message size %d, expected %d\n",
  81                     s, sizeof (msg)));
  82         continue;
  83       }
  84
  85       if (socket.recv (&msg, sizeof (msg)) != s)
  86       {
  87         ACE_ERROR ((LM_ERROR,
  88                     "recv() reported different size than size()\n"));
  89         continue;
  90       }
  91
  92       if (received[msg.sn] == 1)
  93       {
  94         duplicate[msg.sn] = 1;
  95       }
  96       else
  97       {
  98         received[msg.sn] = 1;
  99
 100         if (ACE_OS::memcmp (expected_msg.payload,
 101                             msg.payload,
 102                             payload_size * sizeof(short)) != 0)
 103         {
 104           damaged[msg.sn] = 1;
 105         }
 106       }
 107
 108       if (msg.sn + 1 == message_count) break;
 109     }
 110
 111     unsigned long lost_count (0), damaged_count (0), duplicate_count (0);
 112
 113     {
 114       for (Status_List::Iterator i (received); !i.done (); i.advance ())
 115       {
 116         unsigned char* e;
 117         i.next (e);
 118
 119         if (*e == 0) ++lost_count;
 120       }
 121     }
 122
 123
 124     {
 125       for (Status_List::Iterator i (damaged); !i.done (); i.advance ())
 126       {
 127         unsigned char* e;
 128         i.next (e);
 129
 130         if (*e == 1) ++damaged_count;
 131       }
 132     }
 133
 134
 135     {
 136       for (Status_List::Iterator i (duplicate); !i.done (); i.advance ())
 137       {
 138         unsigned char* e;
 139         i.next (e);
 140
 141         if (*e == 1) ++duplicate_count;
 142       }
 143     }
 144
 145     ACE_DEBUG ((LM_DEBUG,
 146                 "lost      : %d\n"
 147                 "damaged   : %d\n"
 148                 "duplicate : %d\n",
 149                 lost_count,
 150                 damaged_count,
 151                 duplicate_count));
 152
 153     if (damaged_count != 0 || duplicate_count != 0)
 154       return 1;
 155
 156     if (lost_count != 0)
 157     {
 158       // It is ok to lose some number of messages at the beginning.
 159       //
 160       unsigned long count (0);
 161
 162       for (Status_List::Iterator i (received); !i.done (); i.advance ())
 163       {
 164         unsigned char* e;
 165         i.next (e);
 166
 167         if (*e == 0)
 168           ++count;
 169         else
 170           break; // Seen first ok message.
 171       }
 172
 173       if (lost_count != count)
 174         return 1;
 175     }
 176
 177     return 0;
 178   }
 179   catch (args const&)
 180   {
 181     ACE_ERROR ((LM_ERROR,
 182                 "usage: %s <IPv4 multicast address>:<port>\n", argv[0]));
 183   }
 184
 185   return 1;
 186 }