src/lwes_emitter.c

   1 /*======================================================================*
   2  * Copyright (C) 2008 Light Weight Event System                         *
   3  * All rights reserved.                                                 *
   4  *                                                                      *
   5  * This program is free software; you can redistribute it and/or modify *
   6  * it under the terms of the GNU General Public License as published by *
   7  * the Free Software Foundation; either version 2 of the License, or    *
   8  * (at your option) any later version.                                  *
   9  *                                                                      *
  10  * This program is distributed in the hope that it will be useful,      *
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of       *
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the        *
  13  * GNU General Public License for more details.                         *
  14  *                                                                      *
  15  * You should have received a copy of the GNU General Public License    *
  16  * along with this program; if not, write to the Free Software          *
  17  * Foundation, Inc., 51 Franklin Street, Fifth Floor,                   *
  18  * Boston, MA 02110-1301 USA.                                           *
  19  *======================================================================*/
  20 #include "lwes_emitter.h"
  21
  22 /*************************************************************************
  23   PRIVATE API prototypes, shouldn't be called by a user of the library.
  24  *************************************************************************/
  25 int
  26 lwes_emitter_emit_event
  27   (struct lwes_emitter *emitter,
  28    struct lwes_event *event);
  29
  30 int
  31 lwes_emitter_collect_statistics
  32   (struct lwes_emitter *emitter);
  33
  34 void lwes_emitter_calculate_and_send_statistics
  35   (struct lwes_emitter *emitter,
  36    struct lwes_event *stats_event,
  37    time_t current_time);
  38
  39 /*************************************************************************
  40   PUBLIC API
  41  *************************************************************************/
  42 struct lwes_emitter *
  43 lwes_emitter_create
  44   (LWES_CONST_SHORT_STRING address,
  45    LWES_CONST_SHORT_STRING iface,
  46    LWES_U_INT_32 port,
  47    LWES_BOOLEAN emit_heartbeat,
  48    LWES_INT_16 freq)
  49 {
  50   return lwes_emitter_create_with_ttl (address,
  51                                        iface,
  52                                        port,
  53                                        emit_heartbeat,
  54                                        freq,
  55                                        3);
  56 }
  57
  58 struct lwes_emitter *
  59 lwes_emitter_create_with_ttl
  60   (LWES_CONST_SHORT_STRING address,
  61    LWES_CONST_SHORT_STRING iface,
  62    LWES_U_INT_32 port,
  63    LWES_BOOLEAN emit_heartbeat,
  64    LWES_INT_16 freq,
  65    LWES_U_INT_32 ttl)
  66 {
  67   struct lwes_event* tmp_event;
  68   struct lwes_emitter* emitter =
  69     (struct lwes_emitter *) malloc (sizeof (struct lwes_emitter));
  70
  71   if (emitter == NULL)
  72     {
  73       return NULL;
  74     }
  75
  76   if (lwes_net_open (&(emitter->connection),address,iface,port) < 0)
  77     {
  78       free (emitter);
  79       return NULL;
  80     }
  81
  82   if (lwes_net_set_ttl (&(emitter->connection), ttl) == -1)
  83     {
  84       free (emitter);
  85       return NULL;
  86     }
  87
  88   emitter->buffer = (LWES_BYTE_P) malloc (sizeof (LWES_BYTE)*MAX_MSG_SIZE);
  89   if (emitter->buffer == NULL)
  90     {
  91       free(emitter);
  92       return NULL;
  93     }
  94
  95   emitter->count = 0;
  96   emitter->count_since_last_beat = 0;
  97   emitter->sequence = 0;
  98   emitter->frequency = freq;
  99   emitter->emitHeartbeat = emit_heartbeat;
 100
 101   /* Send an event saying we are starting up */
 102   if (emitter->emitHeartbeat)
 103     {
 104       tmp_event = lwes_event_create (NULL,(LWES_SHORT_STRING)"System::Startup");
 105       if ( tmp_event != NULL )
 106         {
 107           emitter->last_beat_time = time (NULL);
 108           lwes_emitter_emit_event (emitter,tmp_event);
 109           lwes_event_destroy (tmp_event);
 110         }
 111     }
 112
 113   return emitter;
 114 }
 115
 116 int
 117 lwes_emitter_emit
 118   (struct lwes_emitter *emitter,
 119    struct lwes_event *event)
 120 {
 121   int error=0;
 122
 123   if(emitter == NULL)
 124   {
 125     return -1;
 126   }
 127
 128   /* Send an event */
 129   error = lwes_emitter_emit_event (emitter,event);
 130
 131   lwes_emitter_collect_statistics (emitter);
 132
 133   return error;
 134 }
 135
 136 int
 137 lwes_emitter_emitto
 138   (LWES_SHORT_STRING address,
 139    LWES_SHORT_STRING iface,
 140    LWES_U_INT_32 port,
 141    struct lwes_emitter *emitter,
 142    struct lwes_event *event)
 143 {
 144   int size;
 145
 146   if(emitter == NULL)
 147   {
 148     return -1;
 149   }
 150
 151   /* Send an event */
 152   if ((size = lwes_event_to_bytes (event,emitter->buffer,MAX_MSG_SIZE,0)) < 0)
 153     {
 154       return -1;
 155     }
 156
 157   if (lwes_net_sendto_bytes (&(emitter->connection),
 158                              address,
 159                              iface,
 160                              port,
 161                              emitter->buffer,size) < 0 )
 162   {
 163     return -2;
 164   }
 165
 166   return 0;
 167 }
 168
 169 int
 170 lwes_emitter_destroy
 171   (struct lwes_emitter *emitter)
 172 {
 173   /* Send an event saying we are shutting down */
 174   int ret = 0;
 175   if (emitter->emitHeartbeat)
 176     {
 177       struct lwes_event *tmp_event =
 178         lwes_event_create(NULL,(LWES_SHORT_STRING)"System::Shutdown");
 179       time_t current_time = time (NULL);
 180
 181       lwes_emitter_calculate_and_send_statistics (emitter,
 182                                                   tmp_event,
 183                                                   current_time);
 184     }
 185
 186   /* shutdown the network, use the return code here for library users */
 187   ret = lwes_net_close (&(emitter->connection));
 188
 189   /* free our memory */
 190   if ( emitter != NULL && emitter->buffer != NULL )
 191     {
 192       free(emitter->buffer);
 193     }
 194   if ( emitter != NULL )
 195     {
 196       free(emitter);
 197     }
 198
 199   return 0;
 200 }
 201
 202 int
 203 lwes_emitter_emit_bytes
 204   (struct lwes_emitter *emitter,
 205    LWES_BYTE_P bytes,
 206    size_t length)
 207 {
 208   return lwes_net_send_bytes (&(emitter->connection), bytes ,length);
 209 }
 210
 211 /*************************************************************************
 212   PRIVATE API
 213  *************************************************************************/
 214 int
 215 lwes_emitter_emit_event
 216   (struct lwes_emitter *emitter,
 217    struct lwes_event *event)
 218 {
 219   int size;
 220
 221   if ((size = lwes_event_to_bytes (event,emitter->buffer,MAX_MSG_SIZE,0)) < 0)
 222   {
 223     return -1;
 224   }
 225
 226   if (lwes_emitter_emit_bytes (emitter, emitter->buffer, size) == -1)
 227   {
 228     return -2;
 229   }
 230
 231   return 0;
 232 }
 233
 234 void lwes_emitter_calculate_and_send_statistics
 235   (struct lwes_emitter *emitter,
 236    struct lwes_event *stats_event,
 237    time_t current_time)
 238 {
 239   if (stats_event != NULL)
 240     {
 241       LWES_INT_16 frequency_this_period;
 242       time_t tmp = (current_time - emitter->last_beat_time);
 243       if ( tmp > 32767 )
 244         {
 245            frequency_this_period=32767;
 246         }
 247       else if ( tmp < 0 )
 248         {
 249           frequency_this_period=0;
 250         }
 251       else
 252         {
 253           frequency_this_period=(LWES_INT_16)tmp;
 254         }
 255
 256       lwes_event_set_INT_16(stats_event,(LWES_SHORT_STRING)"freq",
 257                             frequency_this_period);
 258       lwes_event_set_INT_64(stats_event,(LWES_SHORT_STRING)"seq",
 259                             emitter->sequence);
 260       lwes_event_set_INT_64(stats_event,(LWES_SHORT_STRING)"count",
 261                             emitter->count_since_last_beat);
 262       lwes_event_set_INT_64(stats_event,(LWES_SHORT_STRING)"total",
 263                             emitter->count);
 264       lwes_emitter_emit_event(emitter,stats_event);
 265       lwes_event_destroy(stats_event);
 266     }
 267 }
 268
 269 int
 270 lwes_emitter_collect_statistics
 271   (struct lwes_emitter *emitter)
 272 {
 273   time_t current_time = time (NULL);
 274
 275   /* Count it */
 276   emitter->count++;
 277   emitter->count_since_last_beat++;
 278
 279   /* Send a heartbeat event */
 280   if ( emitter->emitHeartbeat
 281        && (current_time - emitter->last_beat_time) >= emitter->frequency )
 282     {
 283       struct lwes_event *tmp_event =
 284         lwes_event_create (NULL,(LWES_SHORT_STRING)"System::Heartbeat");
 285
 286       if (tmp_event != NULL)
 287         {
 288           emitter->sequence++;
 289           lwes_emitter_calculate_and_send_statistics (emitter,
 290                                                       tmp_event,
 291                                                       current_time);
 292           emitter->last_beat_time = current_time;
 293           emitter->count_since_last_beat = 0;
 294         }
 295     }
 296   return 0;
 297 }