src/lwes-event-counting-listener.c

   1 /*======================================================================*
   2  * Copyright (c) 2008, Yahoo! Inc. All rights reserved.                 *
   3  *                                                                      *
   4  * Licensed under the New BSD License (the "License"); you may not use  *
   5  * this file except in compliance with the License.  Unless required    *
   6  * by applicable law or agreed to in writing, software distributed      *
   7  * under the License is distributed on an "AS IS" BASIS, WITHOUT        *
   8  * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.     *
   9  * See the License for the specific language governing permissions and  *
  10  * limitations under the License. See accompanying LICENSE file.        *
  11  *======================================================================*/
  12
  13 #include "lwes_listener.h"
  14
  15 #ifdef HAVE_GETOPT_H
  16 #include <getopt.h>
  17 #endif
  18
  19 #include <signal.h>
  20 #include <string.h>
  21 #include <sys/types.h>
  22 #include <sys/socket.h>
  23 #include <netinet/in.h>
  24 #include <arpa/inet.h>
  25 #include <stdio.h>
  26 #include <time.h>
  27
  28 // prototypes
  29 static void signal_handler(int sig);
  30
  31 // global variable used to indicate what signal (if any) has been caught
  32 static volatile int done = 0;
  33
  34 static const char help[] =
  35   "lwes-event-counting-listener [options]"                             "\n"
  36   ""                                                                   "\n"
  37   "  where options are:"                                               "\n"
  38   ""                                                                   "\n"
  39   "    -m [one argument]"                                              "\n"
  40   "       The multicast ip address to listen on."                      "\n"
  41   "       (default: 224.1.1.11)"                                       "\n"
  42   ""                                                                   "\n"
  43   "    -p [one argument]"                                              "\n"
  44   "       The ip port to listen on."                                   "\n"
  45   "       (default: 12345)"                                            "\n"
  46   ""                                                                   "\n"
  47   "    -i [one argument]"                                              "\n"
  48   "       The interface to listen on."                                 "\n"
  49   "       (default: 0.0.0.0)"                                          "\n"
  50   ""                                                                   "\n"
  51   "    -q"                                                             "\n"
  52   "       Quiet mode, don't actually print out the message"            "\n"
  53   ""                                                                   "\n"
  54   "    -c"                                                             "\n"
  55   "       Count mode, count the packets which come in and print out"   "\n"
  56   "       the count every second"                                      "\n"
  57   ""                                                                   "\n"
  58   "    -h"                                                             "\n"
  59   "         show this message"                                         "\n"
  60   ""                                                                   "\n"
  61   "  arguments are specified as -option value or -optionvalue"         "\n"
  62   ""                                                                   "\n";
  63
  64 static
  65 int
  66 lwes_U_INT_16_to_stream
  67   (LWES_U_INT_16 a_uint16,
  68    FILE *stream)
  69 {
  70   return fprintf (stream,"%hu",a_uint16);
  71 }
  72
  73 static
  74 int
  75 lwes_INT_16_to_stream
  76   (LWES_INT_16 an_int16,
  77    FILE *stream)
  78 {
  79   return fprintf (stream,"%hd",an_int16);
  80 }
  81
  82 static
  83 int
  84 lwes_U_INT_32_to_stream
  85   (LWES_U_INT_32 a_uint32,
  86    FILE *stream)
  87 {
  88   return fprintf (stream,"%u",a_uint32);
  89 }
  90
  91 static
  92 int
  93 lwes_INT_32_to_stream
  94   (LWES_INT_32 an_int32,
  95    FILE *stream)
  96 {
  97   return fprintf (stream,"%d",an_int32);
  98 }
  99
 100 static
 101 int
 102 lwes_U_INT_64_to_stream
 103   (LWES_U_INT_64 a_uint64,
 104    FILE *stream)
 105 {
 106   return fprintf (stream,"%llu",a_uint64);
 107 }
 108
 109 static
 110 int
 111 lwes_INT_64_to_stream
 112   (LWES_INT_64 an_int64,
 113    FILE *stream)
 114 {
 115   return fprintf (stream,"%lld",an_int64);
 116 }
 117
 118 static
 119 int
 120 lwes_BOOLEAN_to_stream
 121   (LWES_BOOLEAN a_boolean,
 122    FILE *stream)
 123 {
 124   return fprintf (stream,"%s",((a_boolean==1)?"true":"false"));
 125 }
 126
 127 static
 128 int
 129 lwes_IP_ADDR_to_stream
 130   (LWES_IP_ADDR an_ipaddr,
 131    FILE *stream)
 132 {
 133   return fprintf (stream,"%s",inet_ntoa (an_ipaddr));
 134 }
 135
 136 static
 137 int
 138 lwes_SHORT_STRING_to_stream
 139   (LWES_SHORT_STRING a_string,
 140    FILE *stream)
 141 {
 142   return fprintf (stream,"%s",a_string);
 143 }
 144
 145 static
 146 int
 147 lwes_LONG_STRING_to_stream
 148   (LWES_LONG_STRING a_string,
 149    FILE *stream)
 150 {
 151   return fprintf (stream,"%s",a_string);
 152 }
 153
 154 static
 155 int
 156 lwes_event_attribute_to_stream
 157   (struct lwes_event_attribute *attribute,
 158    FILE *stream)
 159 {
 160   if (attribute->type == LWES_U_INT_16_TOKEN)
 161   {
 162     lwes_U_INT_16_to_stream (*((LWES_U_INT_16 *)attribute->value),stream);
 163   }
 164   else if (attribute->type == LWES_INT_16_TOKEN)
 165   {
 166     lwes_INT_16_to_stream (*((LWES_INT_16 *)attribute->value),stream);
 167   }
 168   else if (attribute->type == LWES_U_INT_32_TOKEN)
 169   {
 170     lwes_U_INT_32_to_stream (*((LWES_U_INT_32 *)attribute->value),stream);
 171   }
 172   else if (attribute->type == LWES_INT_32_TOKEN)
 173   {
 174     lwes_INT_32_to_stream (*((LWES_INT_32 *)attribute->value),stream);
 175   }
 176   else if (attribute->type == LWES_U_INT_64_TOKEN)
 177   {
 178     lwes_U_INT_64_to_stream (*((LWES_U_INT_64 *)attribute->value),stream);
 179   }
 180   else if (attribute->type == LWES_INT_64_TOKEN)
 181   {
 182     lwes_INT_64_to_stream (*((LWES_INT_64 *)attribute->value),stream);
 183   }
 184   else if (attribute->type == LWES_BOOLEAN_TOKEN)
 185   {
 186     lwes_BOOLEAN_to_stream (*((LWES_BOOLEAN *)attribute->value),stream);
 187   }
 188   else if (attribute->type == LWES_IP_ADDR_TOKEN)
 189   {
 190     lwes_IP_ADDR_to_stream (*((LWES_IP_ADDR *)attribute->value),stream);
 191   }
 192   else if (attribute->type == LWES_STRING_TOKEN)
 193   {
 194     lwes_LONG_STRING_to_stream ((LWES_LONG_STRING)attribute->value,stream);
 195   }
 196   else
 197   {
 198     /* should really do something here, but not sure what */
 199   }
 200   return 0;
 201 }
 202
 203 static
 204 int
 205 lwes_event_to_stream
 206   (struct lwes_event *event,
 207    FILE *stream)
 208 {
 209   struct lwes_event_attribute *tmp;
 210   struct lwes_hash_enumeration e;
 211
 212   lwes_SHORT_STRING_to_stream (event->eventName,stream);
 213   fprintf (stream,"[");
 214   fflush (stream);
 215   lwes_U_INT_16_to_stream (event->number_of_attributes,stream);
 216   fprintf (stream,"]");
 217   fflush (stream);
 218   fprintf (stream,"\n");
 219   fflush (stream);
 220   fprintf (stream,"{\n");
 221   fflush (stream);
 222
 223   if (lwes_hash_keys (event->attributes, &e))
 224     {
 225       while (lwes_hash_enumeration_has_more_elements (&e))
 226         {
 227           LWES_SHORT_STRING tmpAttrName =
 228             lwes_hash_enumeration_next_element (&e);
 229
 230           tmp =
 231             (struct lwes_event_attribute *)lwes_hash_get (event->attributes,
 232                                                           tmpAttrName);
 233
 234           fprintf (stream,"\t");
 235           fflush (stream);
 236           lwes_SHORT_STRING_to_stream (tmpAttrName,stream);
 237           fflush (stream);
 238           fprintf (stream," = ");
 239           fflush (stream);
 240           lwes_event_attribute_to_stream (tmp,stream);
 241           fflush (stream);
 242           fprintf (stream,";\n");
 243           fflush (stream);
 244         }
 245     }
 246   fprintf (stream,"}\n");
 247   fflush (stream);
 248   return 0;
 249 }
 250
 251 int main (int   argc,
 252           char *argv[])
 253 {
 254   const char *mcast_ip    = "224.1.1.11";
 255   const char *mcast_iface = NULL;
 256   int         mcast_port  = 12345;
 257   int         count       = 0;
 258   int         quiet       = 0;
 259
 260   sigset_t fullset;
 261   struct sigaction act;
 262
 263   struct lwes_listener * listener;
 264   int event_count = 0;
 265   time_t start_time = time (NULL);
 266   int frequency = 1;
 267
 268   /* turn off error messages, I'll handle them */
 269   opterr = 0;
 270   while (1)
 271     {
 272       char c = getopt (argc, argv, "m:p:i:qch");
 273
 274       if (c == -1)
 275         {
 276           break;
 277         }
 278
 279       switch (c)
 280         {
 281           case 'c':
 282             count = 1;
 283             break;
 284
 285           case 'q':
 286             quiet = 1;
 287             break;
 288
 289           case 'm':
 290             mcast_ip = optarg;
 291
 292             break;
 293
 294           case 'p':
 295             mcast_port = atoi(optarg);
 296
 297             break;
 298
 299           case 'h':
 300             fprintf (stderr, "%s", help);
 301
 302             return 1;
 303
 304           case 'i':
 305             mcast_iface = optarg;
 306
 307             break;
 308
 309           default:
 310             fprintf (stderr,
 311                      "error: unrecognized command line option -%c\n",
 312                      optopt);
 313
 314             return 1;
 315         }
 316     }
 317
 318   sigfillset (&fullset);
 319   sigprocmask (SIG_SETMASK, &fullset, NULL);
 320
 321   memset (&act, 0, sizeof (act));
 322   act.sa_handler = signal_handler;
 323   sigfillset (&act.sa_mask);
 324
 325   sigaction (SIGINT, &act, NULL);
 326   sigaction (SIGTERM, &act, NULL);
 327   sigaction (SIGPIPE, &act, NULL);
 328
 329   sigdelset (&fullset, SIGINT);
 330   sigdelset (&fullset, SIGTERM);
 331   sigdelset (&fullset, SIGPIPE);
 332
 333   sigprocmask (SIG_SETMASK, &fullset, NULL);
 334
 335   listener = lwes_listener_create ( (LWES_SHORT_STRING) mcast_ip,
 336                                     (LWES_SHORT_STRING) mcast_iface,
 337                                     (LWES_U_INT_32)     mcast_port );
 338
 339   while ( ! done )
 340     {
 341       struct lwes_event *event = lwes_event_create_no_name ( NULL );
 342
 343       if ( event != NULL )
 344         {
 345           time_t current_time;
 346           int ret = lwes_listener_recv_by (listener, event,1000);
 347           if ( ret > 0 )
 348             {
 349               if (count)
 350                 {
 351                   ++event_count;
 352                 }
 353               if (! quiet)
 354                 {
 355                   lwes_event_to_stream (event, stdout);
 356                 }
 357             }
 358           if (count)
 359             {
 360               current_time = time (NULL);
 361               if ((current_time - start_time) >= frequency)
 362                 {
 363                   char timebuff[20];
 364                   /* HH/MM/SS DD/MM/YYYY  */
 365                   /* 12345678901234567890 */
 366                   start_time = time (NULL);
 367
 368                   strftime (timebuff, 20, "%H:%M:%S %d/%m/%Y",
 369                             localtime (&start_time));
 370
 371                   printf ("%s : %d\n", timebuff, event_count);
 372                   event_count = 0;
 373                 }
 374             }
 375         }
 376       lwes_event_destroy (event);
 377     }
 378
 379   lwes_listener_destroy (listener);
 380
 381   return 0;
 382 }
 383
 384 static void signal_handler(int sig)
 385 {
 386   (void)sig; // appease compiler
 387   done = 1;
 388 }