ext/lwes/emitter.c

   1 #include "lwes_ruby.h"
   2
   3 static VALUE cLWES_Emitter;
   4 static ID id_TYPE_DB, id_TYPE_LIST, id_NAME;
   5 static ID id_new;
   6
   7 /* the underlying struct for LWES::Emitter */
   8 struct _rb_lwes_emitter {
   9         struct lwes_emitter *emitter;
  10         LWES_CONST_SHORT_STRING address;
  11         LWES_CONST_SHORT_STRING iface;
  12         LWES_U_INT_32 port;
  13         LWES_BOOLEAN emit_heartbeat;
  14         LWES_INT_16 freq;
  15         LWES_U_INT_32 ttl;
  16 };
  17
  18 /* gets the _rb_lwes_emitter struct pointer from self */
  19 static struct _rb_lwes_emitter * _rle(VALUE self)
  20 {
  21         struct _rb_lwes_emitter *rle;
  22
  23         Data_Get_Struct(self, struct _rb_lwes_emitter, rle);
  24
  25         return rle;
  26 }
  27
  28 /* GC automatically calls this when object is finalized */
  29 static void rle_free(void *ptr)
  30 {
  31         struct _rb_lwes_emitter *rle = ptr;
  32
  33         if (rle->emitter)
  34                 lwes_emitter_destroy(rle->emitter);
  35         xfree(ptr);
  36 }
  37
  38 static struct lwes_event *
  39 lwesrb_event_create(struct lwes_event_type_db *db, VALUE name)
  40 {
  41         int gc_retry = 1;
  42         const char *event_name = RSTRING_PTR(name);
  43         struct lwes_event *event;
  44
  45 retry:
  46         event = lwes_event_create(db, event_name);
  47         if (!event) {
  48                 if (--gc_retry == 0) {
  49                         rb_gc();
  50                         goto retry;
  51                 }
  52                 rb_raise(rb_eRuntimeError, "failed to create lwes_event");
  53         }
  54         return event;
  55 }
  56
  57 /* called by the GC when object is allocated */
  58 static VALUE rle_alloc(VALUE klass)
  59 {
  60         struct _rb_lwes_emitter *rle;
  61
  62         return Data_Make_Struct(klass, struct _rb_lwes_emitter,
  63                                 NULL, rle_free, rle);
  64 }
  65
  66 /*
  67  * kv - Array:
  68  *   key => String,
  69  *   key => [ numeric_type, Numeric ],
  70  *   key => true,
  71  *   key => false,
  72  * memo - lwes_event pointer
  73  */
  74 static VALUE event_hash_iter_i(VALUE kv, VALUE memo)
  75 {
  76         VALUE *tmp;
  77         VALUE val;
  78         struct lwes_event *event = (struct lwes_event *)memo;
  79         LWES_CONST_SHORT_STRING name;
  80         int rv = 0;
  81         int gc_retry = 1;
  82
  83         assert(TYPE(kv) == T_ARRAY &&
  84                "hash iteration not giving key-value pairs");
  85         tmp = RARRAY_PTR(kv);
  86         name = RSTRING_PTR(rb_obj_as_string(tmp[0]));
  87         val = tmp[1];
  88
  89 retry:
  90         switch (TYPE(val)) {
  91         case T_TRUE:
  92                 rv = lwes_event_set_BOOLEAN(event, name, TRUE);
  93                 break;
  94         case T_FALSE:
  95                 rv = lwes_event_set_BOOLEAN(event, name, FALSE);
  96                 break;
  97         case T_ARRAY:
  98                 rv = lwesrb_event_set_numeric(event, name, val);
  99                 break;
 100         case T_STRING:
 101                 rv = lwes_event_set_STRING(event, name, RSTRING_PTR(val));
 102                 break;
 103         }
 104         if (rv > 0) {
 105                 return Qnil;
 106         } else if (rv == 0) {
 107                 rb_raise(rb_eArgError, "unhandled type %s=%s for event=%s",
 108                          name, RSTRING_PTR(rb_inspect(val)), event->eventName);
 109         } else {
 110                 /* looking at the lwes source code, -3 is allocation errors */
 111                 if (rv == -3 && --gc_retry == 0) {
 112                         rb_gc();
 113                         goto retry;
 114                 }
 115                 rb_raise(rb_eRuntimeError, "failed to set %s=%s for event=%s",
 116                          name, RSTRING_PTR(rb_inspect(val)), event->eventName);
 117         }
 118         return Qfalse;
 119 }
 120
 121 static VALUE _emit_hash(VALUE _tmp)
 122 {
 123         VALUE *tmp = (VALUE *)_tmp;
 124         VALUE self = tmp[0];
 125         VALUE _event = tmp[1];
 126         struct lwes_event *event = (struct lwes_event *)tmp[2];
 127
 128         rb_iterate(rb_each, _event, event_hash_iter_i, (VALUE)event);
 129         if (lwes_emitter_emit(_rle(self)->emitter, event) < 0)
 130                 rb_raise(rb_eRuntimeError, "failed to emit event");
 131
 132         return _event;
 133 }
 134
 135 static void
 136 set_field(
 137         struct lwes_event *event,
 138         LWES_CONST_SHORT_STRING name,
 139         LWES_TYPE type,
 140         VALUE val)
 141 {
 142         int gc_retry = 1;
 143         int rv;
 144
 145 retry:
 146         switch (type) {
 147         case LWES_TYPE_BOOLEAN:
 148                 if (val == Qfalse)
 149                         rv = lwes_event_set_BOOLEAN(event, name, FALSE);
 150                 else if (val == Qtrue)
 151                         rv = lwes_event_set_BOOLEAN(event, name, TRUE);
 152                 else
 153                         rb_raise(rb_eTypeError, "non-boolean set for %s: %s",
 154                                  name, RSTRING_PTR(rb_inspect(val)));
 155                 break;
 156         case LWES_TYPE_STRING:
 157                 if (TYPE(val) != T_STRING)
 158                         rb_raise(rb_eTypeError, "non-String set for %s: %s",
 159                                  name, RSTRING_PTR(rb_inspect(val)));
 160                 rv = lwes_event_set_STRING(event, name, RSTRING_PTR(val));
 161                 break;
 162         default:
 163                 rv = lwesrb_event_set_num(event, name, type, val);
 164         }
 165         if (rv > 0) {
 166                 return;
 167         } else {
 168                 if (rv == -3 && --gc_retry == 0) {
 169                         rb_gc();
 170                         goto retry;
 171                 }
 172                 rb_raise(rb_eRuntimeError,
 173                          "failed to set %s=%s for event=%s (error: %d)",
 174                          name, RSTRING_PTR(rb_inspect(val)),
 175                          event->eventName, rv);
 176         }
 177
 178         assert(0 && "you should never get here (set_field)");
 179 }
 180
 181 static VALUE _emit_struct(VALUE _argv)
 182 {
 183         VALUE *argv = (VALUE *)_argv;
 184         VALUE self = argv[0];
 185         VALUE _event = argv[1];
 186         struct lwes_event *event = (struct lwes_event *)argv[2];
 187         VALUE type_list = argv[3];
 188         long i;
 189         VALUE *tmp;
 190
 191         if (TYPE(type_list) != T_ARRAY)
 192                 rb_raise(rb_eArgError, "could not get TYPE_LIST const");
 193
 194         i = RARRAY_LEN(type_list);
 195         for (tmp = RARRAY_PTR(type_list); --i >= 0; tmp++) {
 196                 /* inner: [ :field_sym, "field_name", type ] */
 197                 VALUE *inner = RARRAY_PTR(*tmp);
 198                 VALUE val = rb_struct_aref(_event, inner[0]);
 199                 LWES_CONST_SHORT_STRING name;
 200                 LWES_TYPE type;
 201
 202                 if (NIL_P(val))
 203                         continue;
 204
 205                 name = RSTRING_PTR(inner[1]);
 206                 type = NUM2INT(inner[2]);
 207                 set_field(event, name, type, val);
 208         }
 209
 210         if (lwes_emitter_emit(_rle(self)->emitter, event) < 0)
 211                 rb_raise(rb_eRuntimeError, "failed to emit event");
 212
 213         return _event;
 214 }
 215
 216 static VALUE _destroy_event(VALUE _event)
 217 {
 218         struct lwes_event *event = (struct lwes_event *)_event;
 219
 220         assert(event && "destroying NULL event");
 221         lwes_event_destroy(event);
 222
 223         return Qnil;
 224 }
 225
 226 static VALUE emit_hash(VALUE self, VALUE name, VALUE _event)
 227 {
 228         VALUE tmp[3];
 229         struct lwes_event *event = lwesrb_event_create(NULL, name);
 230
 231         tmp[0] = self;
 232         tmp[1] = _event;
 233         tmp[2] = (VALUE)event;
 234         rb_ensure(_emit_hash, (VALUE)&tmp, _destroy_event, (VALUE)event);
 235
 236         return _event;
 237 }
 238
 239 static struct lwes_event_type_db * get_type_db(VALUE event_class)
 240 {
 241         VALUE type_db = rb_const_get(event_class, id_TYPE_DB);
 242
 243         if (CLASS_OF(type_db) != cLWES_TypeDB)
 244                 rb_raise(rb_eArgError, "class does not have valid TYPE_DB");
 245
 246         return lwesrb_get_type_db(type_db);
 247 }
 248
 249 static VALUE emit_struct(VALUE self, VALUE _event)
 250 {
 251         VALUE argv[4];
 252         VALUE event_class = CLASS_OF(_event);
 253         struct lwes_event_type_db *db = get_type_db(event_class);
 254         struct lwes_event *event;
 255         VALUE name = rb_const_get(event_class, id_NAME);
 256         VALUE type_list = rb_const_get(event_class, id_TYPE_LIST);
 257
 258         if (TYPE(name) != T_STRING || TYPE(type_list) != T_ARRAY)
 259                 rb_raise(rb_eArgError,
 260                          "could not get class NAME or TYPE_LIST from: %s",
 261                          RSTRING_PTR(rb_inspect(_event)));
 262
 263         event = lwesrb_event_create(db, name);
 264
 265         argv[0] = self;
 266         argv[1] = _event;
 267         argv[2] = (VALUE)event;
 268         argv[3] = type_list;
 269         rb_ensure(_emit_struct, (VALUE)&argv, _destroy_event, (VALUE)event);
 270
 271         return _event;
 272 }
 273
 274 /*
 275  * call-seq:
 276  *   emitter = LWES::Emitter.new
 277  *   event = EventStruct.new
 278  *   event.foo = "bar"
 279  *   emitter << event
 280  */
 281 static VALUE emitter_ltlt(VALUE self, VALUE event)
 282 {
 283         if (TYPE(event) != T_STRUCT)
 284                 rb_raise(rb_eArgError,
 285                          "Must be a Struct: %s",
 286                          RSTRING_PTR(rb_inspect(event)));
 287
 288         return emit_struct(self, event);
 289 }
 290
 291 /*
 292  * call-seq:
 293  *   emitter = LWES::Emitter.new
 294  *
 295  *   emitter.emit("EventName", :foo => "HI")
 296  *
 297  *   emitter.emit(EventStruct, :foo => "HI")
 298  *
 299  *   struct = EventStruct.new
 300  *   struct.foo = "HI"
 301  *   emitter.emit(struct)
 302  */
 303 static VALUE emitter_emit(int argc, VALUE *argv, VALUE self)
 304 {
 305         VALUE name = Qnil;
 306         VALUE event = Qnil;
 307         argc = rb_scan_args(argc, argv, "11", &name, &event);
 308
 309         switch (TYPE(name)) {
 310         case T_STRING:
 311                 if (TYPE(event) == T_HASH)
 312                         return emit_hash(self, name, event);
 313                 rb_raise(rb_eArgError,
 314                          "second argument must be a hash when first "
 315                          "is a String");
 316         case T_STRUCT:
 317                 if (argc >= 2)
 318                         rb_raise(rb_eArgError,
 319                                  "second argument not allowed when first"
 320                                  " is a Struct");
 321                 event = name;
 322                 return emit_struct(self, event);
 323         case T_CLASS:
 324                 if (TYPE(event) != T_HASH)
 325                         rb_raise(rb_eArgError,
 326                                  "second argument must be a Hash when first"
 327                                  " is a Class");
 328
 329                 /*
 330                  * we can optimize this so there's no intermediate
 331                  * struct created
 332                  */
 333                 event = rb_funcall(name, id_new, 1, event);
 334                 return emit_struct(self, event);
 335         default:
 336                 rb_raise(rb_eArgError,
 337                          "bad argument: %s, must be a String, Struct or Class",
 338                          RSTRING_PTR(rb_inspect(name)));
 339         }
 340
 341         assert(0 && "should never get here");
 342         return event;
 343 }
 344
 345 /*
 346  * Destroys the associated lwes_emitter and the associated socket.  This
 347  * method is rarely needed as Ruby garbage collection will take care of
 348  * closing for you, but may be useful in odd cases when it is desirable
 349  * to release file descriptors ASAP.
 350  */
 351 static VALUE emitter_close(VALUE self)
 352 {
 353         struct _rb_lwes_emitter *rle = _rle(self);
 354
 355         if (rle->emitter)
 356                 lwes_emitter_destroy(rle->emitter);
 357         rle->emitter = NULL;
 358
 359         return Qnil;
 360 }
 361
 362 static void lwesrb_emitter_create(struct _rb_lwes_emitter *rle)
 363 {
 364         int gc_retry = 1;
 365 retry:
 366         if (rle->ttl == UINT32_MAX)
 367                 rle->emitter = lwes_emitter_create(
 368                          rle->address, rle->iface, rle->port,
 369                          rle->emit_heartbeat, rle->freq);
 370         else
 371                 rle->emitter = lwes_emitter_create_with_ttl(
 372                          rle->address, rle->iface, rle->port,
 373                          rle->emit_heartbeat, rle->freq, rle->ttl);
 374
 375         if (!rle->emitter) {
 376                 if (--gc_retry == 0) {
 377                         rb_gc();
 378                         goto retry;
 379                 }
 380                 rb_raise(rb_eRuntimeError, "failed to create LWES emitter");
 381         }
 382 }
 383
 384 /* should only used internally by #initialize */
 385 static VALUE _create(VALUE self, VALUE options)
 386 {
 387         struct _rb_lwes_emitter *rle = _rle(self);
 388         VALUE address, iface, port, heartbeat, ttl;
 389
 390         rle->emit_heartbeat = FALSE;
 391         rle->freq = 0;
 392         rle->ttl = UINT32_MAX; /* nobody sets a ttl this long, right? */
 393
 394         if (rle->emitter)
 395                 rb_raise(rb_eRuntimeError, "already created lwes_emitter");
 396         if (TYPE(options) != T_HASH)
 397                 rb_raise(rb_eTypeError, "options must be a hash");
 398
 399         address = rb_hash_aref(options, ID2SYM(rb_intern("address")));
 400         if (TYPE(address) != T_STRING)
 401                 rb_raise(rb_eTypeError, ":address must be a string");
 402         rle->address = RSTRING_PTR(address);
 403
 404         iface = rb_hash_aref(options, ID2SYM(rb_intern("iface")));
 405         switch (TYPE(iface)) {
 406         case T_NIL:
 407                 rle->iface = NULL;
 408                 break;
 409         case T_STRING:
 410                 rle->iface = RSTRING_PTR(iface);
 411                 break;
 412         default:
 413                 rb_raise(rb_eTypeError, ":iface must be a String or nil");
 414         }
 415
 416         port = rb_hash_aref(options, ID2SYM(rb_intern("port")));
 417         if (TYPE(port) != T_FIXNUM)
 418                 rb_raise(rb_eTypeError, ":port must be a Fixnum");
 419         rle->port = NUM2UINT(port);
 420
 421         heartbeat = rb_hash_aref(options, ID2SYM(rb_intern("heartbeat")));
 422         if (TYPE(heartbeat) == T_FIXNUM) {
 423                 int tmp = NUM2INT(heartbeat);
 424                 if (tmp > INT16_MAX)
 425                         rb_raise(rb_eArgError,":heartbeat > INT16_MAX seconds");
 426                 rle->emit_heartbeat = TRUE;
 427                 rle->freq = (LWES_INT_16)tmp;
 428         } else if (NIL_P(heartbeat)) { /* do nothing, use defaults */
 429         } else
 430                 rb_raise(rb_eTypeError, ":heartbeat must be a Fixnum or nil");
 431
 432         ttl = rb_hash_aref(options, ID2SYM(rb_intern("ttl")));
 433         if (TYPE(ttl) == T_FIXNUM) {
 434                 unsigned LONG_LONG tmp = NUM2ULL(ttl);
 435                 if (tmp >= UINT32_MAX)
 436                         rb_raise(rb_eArgError, ":ttl >= UINT32_MAX seconds");
 437                 rle->ttl = (LWES_U_INT_32)tmp;
 438         } else if (NIL_P(ttl)) { /* do nothing, no ttl */
 439         } else
 440                 rb_raise(rb_eTypeError, ":ttl must be a Fixnum or nil");
 441
 442         lwesrb_emitter_create(rle);
 443
 444         return self;
 445 }
 446
 447 /* Init_lwes_ext will call this */
 448 void lwesrb_init_emitter(void)
 449 {
 450         VALUE mLWES = rb_define_module("LWES");
 451         cLWES_Emitter = rb_define_class_under(mLWES, "Emitter", rb_cObject);
 452
 453         rb_define_method(cLWES_Emitter, "<<", emitter_ltlt, 1);
 454         rb_define_method(cLWES_Emitter, "emit", emitter_emit, -1);
 455         rb_define_method(cLWES_Emitter, "_create", _create, 1);
 456         rb_define_method(cLWES_Emitter, "close", emitter_close, 0);
 457         rb_define_alloc_func(cLWES_Emitter, rle_alloc);
 458         LWESRB_MKID(TYPE_DB);
 459         LWESRB_MKID(TYPE_LIST);
 460         LWESRB_MKID(NAME);
 461         LWESRB_MKID(new);
 462 }