ext/lwes_ext/emitter.c

   1 #include "lwes_ruby.h"
   2 #ifdef HAVE_RUBY_UTIL_H
   3 #  include <ruby/util.h>
   4 #else
   5 #  include "util.h"
   6 #endif
   7
   8 static VALUE ENC; /* LWES_ENCODING */
   9 static ID id_TYPE_DB, id_TYPE_LIST, id_NAME, id_HAVE_ENCODING;
  10 static ID id_new, id_enc, id_size, id_to_a;
  11 static VALUE sym_enc;
  12
  13 static void dump_name(char *name, LWES_BYTE_P buf, size_t *off)
  14 {
  15         if (marshall_SHORT_STRING(name, buf, MAX_MSG_SIZE, off) > 0)
  16                 return;
  17         rb_raise(rb_eRuntimeError, "failed to dump name=%s", name);
  18 }
  19
  20 static int dump_bool(char *name, VALUE val, LWES_BYTE_P buf, size_t *off)
  21 {
  22         dump_name(name, buf, off);
  23         lwesrb_dump_type(LWES_BOOLEAN_TOKEN, buf, off);
  24         return marshall_BOOLEAN(lwesrb_boolean(val), buf, MAX_MSG_SIZE, off);
  25 }
  26
  27 static int dump_string(char *name, VALUE val, LWES_BYTE_P buf, size_t *off)
  28 {
  29         char *dst;
  30
  31         switch (TYPE(val)) {
  32         case T_BIGNUM:
  33         case T_FIXNUM:
  34                 val = rb_obj_as_string(val);
  35         }
  36         dst = StringValueCStr(val);
  37
  38         dump_name(name, buf, off);
  39         lwesrb_dump_type(LWES_STRING_TOKEN, buf, off);
  40         return marshall_LONG_STRING(dst, buf, MAX_MSG_SIZE, off);
  41 }
  42
  43 static void dump_enc(VALUE enc, LWES_BYTE_P buf, size_t *off)
  44 {
  45         dump_name((char *)LWES_ENCODING, buf, off);
  46         lwesrb_dump_num(LWES_INT_16_TOKEN, enc, buf, off);
  47 }
  48
  49 /* the underlying struct for LWES::Emitter */
  50 struct _rb_lwes_emitter {
  51         struct lwes_emitter *emitter;
  52         char *address;
  53         char *iface;
  54         LWES_U_INT_32 port;
  55         LWES_BOOLEAN emit_heartbeat;
  56         LWES_INT_16 freq;
  57         LWES_U_INT_32 ttl;
  58 };
  59
  60 /* gets the _rb_lwes_emitter struct pointer from self */
  61 static struct _rb_lwes_emitter * _rle(VALUE self)
  62 {
  63         struct _rb_lwes_emitter *rle;
  64
  65         Data_Get_Struct(self, struct _rb_lwes_emitter, rle);
  66
  67         return rle;
  68 }
  69
  70 /* GC automatically calls this when object is finalized */
  71 static void rle_free(void *ptr)
  72 {
  73         struct _rb_lwes_emitter *rle = ptr;
  74
  75         if (rle->emitter)
  76                 lwes_emitter_destroy(rle->emitter);
  77         xfree(rle->address);
  78         xfree(rle->iface);
  79         xfree(ptr);
  80 }
  81
  82 /* called by the GC when object is allocated */
  83 static VALUE rle_alloc(VALUE klass)
  84 {
  85         struct _rb_lwes_emitter *rle;
  86
  87         return Data_Make_Struct(klass, struct _rb_lwes_emitter,
  88                                 NULL, rle_free, rle);
  89 }
  90
  91 struct hash_memo {
  92         size_t off;
  93         LWES_BYTE_P buf;
  94 };
  95
  96 /*
  97  * kv - Array:
  98  *   key => String,
  99  *   key => [ numeric_type, Numeric ],
 100  *   key => true,
 101  *   key => false,
 102  * memo - lwes_event pointer
 103  */
 104 static VALUE event_hash_iter_i(VALUE kv, VALUE memo)
 105 {
 106         volatile VALUE raise_inspect;
 107         struct hash_memo *hash_memo = (struct hash_memo *)NUM2ULONG(memo);
 108         VALUE val;
 109         VALUE name;
 110         char *attr_name;
 111         int rv = 0;
 112         LWES_BYTE_P buf = hash_memo->buf;
 113         size_t *off = &hash_memo->off;
 114
 115         if (TYPE(kv) != T_ARRAY || RARRAY_LEN(kv) != 2)
 116                 rb_raise(rb_eTypeError,
 117                          "hash iteration not giving key-value pairs");
 118
 119         name = rb_ary_entry(kv, 0);
 120
 121         if (name == sym_enc) return Qnil; /* already dumped first */
 122
 123         name = rb_obj_as_string(name);
 124         attr_name = StringValueCStr(name);
 125
 126         if (strcmp(attr_name, LWES_ENCODING) == 0)
 127                 return Qnil;
 128
 129         val = rb_ary_entry(kv, 1);
 130
 131         switch (TYPE(val)) {
 132         case T_TRUE:
 133         case T_FALSE:
 134                 rv = dump_bool(attr_name, val, buf, off);
 135                 break;
 136         case T_ARRAY:
 137                 dump_name(attr_name, buf, off);
 138                 lwesrb_dump_num_ary(val, buf, off);
 139                 return Qnil;
 140         case T_STRING:
 141                 rv = dump_string(attr_name, val, buf, off);
 142                 break;
 143         }
 144
 145         if (rv > 0)
 146                 return Qnil;
 147
 148         rb_raise(rb_eArgError, "unhandled type %s=%s",
 149                  attr_name, RAISE_INSPECT(val));
 150         return Qfalse;
 151 }
 152
 153 static VALUE emit_hash(VALUE self, VALUE name, VALUE event)
 154 {
 155         struct _rb_lwes_emitter *rle = _rle(self);
 156         struct hash_memo hash_memo;
 157         LWES_BYTE_P buf;
 158         size_t *off;
 159         VALUE memo = ULONG2NUM((unsigned long)&hash_memo);
 160         VALUE enc;
 161         LWES_U_INT_16 size = lwesrb_uint16(rb_funcall(event, id_size, 0, 0));
 162         int rv;
 163         char *event_name = StringValueCStr(name);
 164
 165         buf = hash_memo.buf = rle->emitter->buffer;
 166         hash_memo.off = 0;
 167         off = &hash_memo.off;
 168
 169         /* event name first */
 170         dump_name(event_name, buf, off);
 171
 172         /* number of attributes second */
 173         rv = marshall_U_INT_16(size, buf, MAX_MSG_SIZE, off);
 174         if (rv <= 0)
 175                 rb_raise(rb_eRuntimeError, "failed to dump num_attrs");
 176
 177         /* dump encoding before other fields */
 178         enc = rb_hash_aref(event, sym_enc);
 179         if (NIL_P(enc))
 180                 enc = rb_hash_aref(event, ENC);
 181         if (! NIL_P(enc))
 182                 dump_enc(enc, buf, off);
 183
 184         /* the rest of the fields */
 185         rb_iterate(rb_each, event, event_hash_iter_i, memo);
 186
 187         if (lwes_emitter_emit_bytes(rle->emitter, buf, *off) < 0)
 188                 rb_raise(rb_eRuntimeError, "failed to emit event");
 189
 190         return event;
 191 }
 192
 193 static void
 194 marshal_field(
 195         char *name,
 196         LWES_TYPE type,
 197         VALUE val,
 198         LWES_BYTE_P buf,
 199         size_t *off)
 200 {
 201         volatile VALUE raise_inspect;
 202
 203         switch (type) {
 204         case LWES_TYPE_STRING:
 205                 if (dump_string(name, val, buf, off) > 0)
 206                         return;
 207                 break;
 208         case LWES_TYPE_BOOLEAN:
 209                 if (dump_bool(name, val, buf, off) > 0)
 210                         return;
 211                 break;
 212         default:
 213                 dump_name(name, buf, off);
 214                 lwesrb_dump_num(type, val, buf, off);
 215                 return;
 216         }
 217
 218         rb_raise(rb_eRuntimeError, "failed to set %s=%s",
 219                  name, RAISE_INSPECT(val));
 220 }
 221
 222 static void lwes_struct_class(
 223         VALUE *event_class,
 224         VALUE *name,
 225         VALUE *type_list,
 226         VALUE *have_enc,
 227         VALUE event)
 228 {
 229         VALUE type_db;
 230
 231         *event_class = CLASS_OF(event);
 232         type_db = rb_const_get(*event_class, id_TYPE_DB);
 233
 234         if (CLASS_OF(type_db) != cLWES_TypeDB)
 235                 rb_raise(rb_eArgError, "class does not have valid TYPE_DB");
 236
 237         *name = rb_const_get(*event_class, id_NAME);
 238         Check_Type(*name, T_STRING);
 239         *type_list = rb_const_get(*event_class, id_TYPE_LIST);
 240         Check_Type(*type_list, T_ARRAY);
 241
 242         *have_enc = rb_const_get(*event_class, id_HAVE_ENCODING);
 243 }
 244
 245 static VALUE emit_struct(VALUE self, VALUE event)
 246 {
 247         VALUE event_class, name, type_list, have_enc, event_ary;
 248         struct _rb_lwes_emitter *rle = _rle(self);
 249         LWES_BYTE_P buf = rle->emitter->buffer;
 250         size_t off = 0;
 251         long i, len;
 252         LWES_U_INT_16 num_attr = 0;
 253         size_t num_attr_off;
 254         char *str;
 255
 256         lwes_struct_class(&event_class, &name, &type_list, &have_enc, event);
 257
 258         /* event name */
 259         str = StringValueCStr(name);
 260         dump_name(str, buf, &off);
 261
 262         /* number of attributes, use a placeholder until we've iterated */
 263         num_attr_off = off;
 264         if (marshall_U_INT_16(0, buf, MAX_MSG_SIZE, &off) < 0)
 265                 rb_raise(rb_eRuntimeError,
 266                          "failed to marshal number_of_attributes");
 267
 268         /* dump encoding before other fields */
 269         if (have_enc == Qtrue) {
 270                 VALUE enc = rb_funcall(event, id_enc, 0, 0);
 271                 if (! NIL_P(enc)) {
 272                         ++num_attr;
 273                         dump_enc(enc, buf, &off);
 274                 }
 275         }
 276
 277         len = RARRAY_LEN(type_list);
 278         event_ary = rb_funcall(event, id_to_a, 0, 0);
 279         for (i = 0; i < len; i++) {
 280                 /* type_list [ [ :field_sym, "field_name", ltype ] ] */
 281                 VALUE tlent = rb_ary_entry(type_list, i);
 282                 VALUE field_sym = rb_ary_entry(tlent, 0);
 283                 VALUE field_name;
 284                 VALUE val;
 285                 LWES_TYPE type;
 286
 287                 if (field_sym == sym_enc) /* encoding was already dumped */
 288                         continue;
 289
 290                 val = rb_ary_entry(event_ary, i);
 291                 if (NIL_P(val))
 292                         continue; /* LWES doesn't know nil */
 293
 294                 field_name = rb_ary_entry(tlent, 1);
 295                 str = StringValueCStr(field_name);
 296                 type = NUM2INT(rb_ary_entry(tlent, 2));
 297                 ++num_attr;
 298                 marshal_field(str, type, val, buf, &off);
 299         }
 300
 301         /* now we've iterated, we can accurately give num_attr */
 302         if (marshall_U_INT_16(num_attr, buf, MAX_MSG_SIZE, &num_attr_off) <= 0)
 303                 rb_raise(rb_eRuntimeError, "failed to marshal num_attr");
 304
 305         if (lwes_emitter_emit_bytes(rle->emitter, buf, off) < 0)
 306                 rb_raise(rb_eRuntimeError, "failed to emit event");
 307
 308         return event;
 309 }
 310
 311 static VALUE emit_event(VALUE self, VALUE event)
 312 {
 313         struct lwes_event *e = lwesrb_get_event(event);
 314
 315         if (lwes_emitter_emit(_rle(self)->emitter, e) < 0)
 316                 rb_raise(rb_eRuntimeError, "failed to emit event");
 317
 318         return event;
 319 }
 320 /*
 321  * call-seq:
 322  *   emitter << event
 323  *
 324  * Emits the given +event+ which much be an LWES::Event or
 325  * LWES::Struct-derived object
 326  */
 327 static VALUE emitter_ltlt(VALUE self, VALUE event)
 328 {
 329         if (rb_obj_is_kind_of(event, cLWES_Event)) {
 330                 return emit_event(self, event);
 331         } else {
 332                 Check_Type(event, T_STRUCT);
 333
 334                 return emit_struct(self, event);
 335         }
 336 }
 337
 338 /*
 339  * call-seq:
 340  *      emitter.emit("EventName", :foo => "HI")
 341  *      emitter.emit("EventName", :foo => [ :int32, 123 ])
 342  *      emitter.emit(EventClass, :foo => "HI")
 343  *      emitter.emit(event)
 344  *
 345  * Emits a hash.  If EventName is given as a string, it will expect a hash
 346  * as its second argument and will do its best to serialize a Ruby Hash
 347  * to an LWES Event.  If a type is ambiguous, a two-element array may be
 348  * specified as its value, including the LWES type information and the
 349  * Ruby value.
 350  *
 351  * If an EventClass is given, the second argument should be a hash with
 352  * the values given to the class.   This will emit the event named by
 353  * EventClass.
 354  *
 355  * If only one argument is given, it behaves just like LWES::Emitter#<<
 356  */
 357 static VALUE emitter_emit(int argc, VALUE *argv, VALUE self)
 358 {
 359         volatile VALUE raise_inspect;
 360         char *err;
 361         VALUE name = Qnil;
 362         VALUE event = Qnil;
 363         argc = rb_scan_args(argc, argv, "11", &name, &event);
 364
 365         switch (TYPE(name)) {
 366         case T_STRING:
 367                 if (TYPE(event) == T_HASH)
 368                         return emit_hash(self, name, event);
 369                 rb_raise(rb_eTypeError,
 370                          "second argument must be a hash when first "
 371                          "is a String");
 372         case T_STRUCT:
 373                 if (argc >= 2)
 374                         rb_raise(rb_eArgError,
 375                                  "second argument not allowed when first"
 376                                  " is a Struct");
 377                 event = name;
 378                 return emit_struct(self, event);
 379         case T_CLASS:
 380                 if (TYPE(event) != T_HASH)
 381                         rb_raise(rb_eTypeError,
 382                                  "second argument must be a Hash when first"
 383                                  " is a Class");
 384
 385                 /*
 386                  * we can optimize this so there's no intermediate
 387                  * struct created
 388                  */
 389                 event = rb_funcall(name, id_new, 1, event);
 390                 if (TYPE(event) == T_STRUCT)
 391                         return emit_struct(self, event);
 392                 if (rb_obj_is_kind_of(event, cLWES_Event))
 393                         return emit_event(self, event);
 394                 name = rb_class_name(name);
 395                 err = StringValuePtr(name);
 396                 rb_raise(rb_eArgError,
 397                          "%s created a bad event: %s",
 398                          err, RAISE_INSPECT(event));
 399         default:
 400                 if (rb_obj_is_kind_of(name, cLWES_Event))
 401                         return emit_event(self, name);
 402                 rb_raise(rb_eArgError,
 403                          "bad argument: %s, must be a String, Struct or Class",
 404                          RAISE_INSPECT(name));
 405         }
 406
 407         assert(0 && "should never get here");
 408         return event;
 409 }
 410
 411 /*
 412  * call-seq:
 413  *      emitter.close   -> nil
 414  *
 415  * Destroys the associated lwes_emitter and the associated socket.  This
 416  * method is rarely needed as Ruby garbage collection will take care of
 417  * closing for you, but may be useful in odd cases when it is desirable
 418  * to release file descriptors ASAP.
 419  */
 420 static VALUE emitter_close(VALUE self)
 421 {
 422         struct _rb_lwes_emitter *rle = _rle(self);
 423
 424         if (rle->emitter)
 425                 lwes_emitter_destroy(rle->emitter);
 426         rle->emitter = NULL;
 427
 428         return Qnil;
 429 }
 430
 431 static void lwesrb_emitter_create(struct _rb_lwes_emitter *rle)
 432 {
 433         int gc_retry = 1;
 434 retry:
 435         if (rle->ttl == UINT32_MAX)
 436                 rle->emitter = lwes_emitter_create(
 437                          rle->address, rle->iface, rle->port,
 438                          rle->emit_heartbeat, rle->freq);
 439         else
 440                 rle->emitter = lwes_emitter_create_with_ttl(
 441                          rle->address, rle->iface, rle->port,
 442                          rle->emit_heartbeat, rle->freq, rle->ttl);
 443
 444         if (!rle->emitter) {
 445                 if (--gc_retry == 0) {
 446                         rb_gc();
 447                         goto retry;
 448                 }
 449                 rb_raise(rb_eRuntimeError, "failed to create LWES emitter");
 450         }
 451 }
 452
 453 /* :nodoc: */
 454 static VALUE init_copy(VALUE dest, VALUE obj)
 455 {
 456         struct _rb_lwes_emitter *dst = _rle(dest);
 457         struct _rb_lwes_emitter *src = _rle(obj);
 458
 459         memcpy(dst, src, sizeof(*dst));
 460         dst->address = ruby_strdup(src->address);
 461         if (dst->iface)
 462                 dst->iface = ruby_strdup(src->iface);
 463         lwesrb_emitter_create(dst);
 464
 465         assert(dst->emitter && dst->emitter != src->emitter &&
 466                "emitter not a copy");
 467
 468         return dest;
 469 }
 470
 471 /* :nodoc: should only used internally by #initialize */
 472 static VALUE _create(VALUE self, VALUE options)
 473 {
 474         struct _rb_lwes_emitter *rle = _rle(self);
 475         VALUE address, iface, port, heartbeat, ttl;
 476
 477         rle->emit_heartbeat = FALSE;
 478         rle->freq = 0;
 479         rle->ttl = UINT32_MAX; /* nobody sets a ttl this long, right? */
 480
 481         if (rle->emitter)
 482                 rb_raise(rb_eRuntimeError, "already created lwes_emitter");
 483         if (TYPE(options) != T_HASH)
 484                 rb_raise(rb_eTypeError, "options must be a hash");
 485
 486         address = rb_hash_aref(options, ID2SYM(rb_intern("address")));
 487         if (TYPE(address) != T_STRING)
 488                 rb_raise(rb_eTypeError, ":address must be a string");
 489         rle->address = ruby_strdup(StringValueCStr(address));
 490
 491         iface = rb_hash_aref(options, ID2SYM(rb_intern("iface")));
 492         switch (TYPE(iface)) {
 493         case T_NIL:
 494                 rle->iface = NULL;
 495                 break;
 496         case T_STRING:
 497                 rle->iface = ruby_strdup(StringValueCStr(iface));
 498                 break;
 499         default:
 500                 rb_raise(rb_eTypeError, ":iface must be a String or nil");
 501         }
 502
 503         port = rb_hash_aref(options, ID2SYM(rb_intern("port")));
 504         if (TYPE(port) != T_FIXNUM)
 505                 rb_raise(rb_eTypeError, ":port must be a Fixnum");
 506         rle->port = NUM2UINT(port);
 507
 508         heartbeat = rb_hash_aref(options, ID2SYM(rb_intern("heartbeat")));
 509         if (TYPE(heartbeat) == T_FIXNUM) {
 510                 int tmp = NUM2INT(heartbeat);
 511                 if (tmp > INT16_MAX)
 512                         rb_raise(rb_eArgError,":heartbeat > INT16_MAX seconds");
 513                 rle->emit_heartbeat = TRUE;
 514                 rle->freq = (LWES_INT_16)tmp;
 515         } else if (NIL_P(heartbeat)) { /* do nothing, use defaults */
 516         } else
 517                 rb_raise(rb_eTypeError, ":heartbeat must be a Fixnum or nil");
 518
 519         ttl = rb_hash_aref(options, ID2SYM(rb_intern("ttl")));
 520         if (TYPE(ttl) == T_FIXNUM) {
 521                 unsigned LONG_LONG tmp = NUM2ULL(ttl);
 522                 if (tmp >= UINT32_MAX)
 523                         rb_raise(rb_eArgError, ":ttl >= UINT32_MAX seconds");
 524                 rle->ttl = (LWES_U_INT_32)tmp;
 525         } else if (NIL_P(ttl)) { /* do nothing, no ttl */
 526         } else
 527                 rb_raise(rb_eTypeError, ":ttl must be a Fixnum or nil");
 528
 529         lwesrb_emitter_create(rle);
 530
 531         return self;
 532 }
 533
 534 /* Init_lwes_ext will call this */
 535 void lwesrb_init_emitter(void)
 536 {
 537         VALUE mLWES = rb_define_module("LWES");
 538         VALUE cLWES_Emitter =
 539                           rb_define_class_under(mLWES, "Emitter", rb_cObject);
 540
 541         rb_define_method(cLWES_Emitter, "<<", emitter_ltlt, 1);
 542         rb_define_method(cLWES_Emitter, "emit", emitter_emit, -1);
 543         rb_define_method(cLWES_Emitter, "_create", _create, 1);
 544         rb_define_method(cLWES_Emitter, "close", emitter_close, 0);
 545         rb_define_method(cLWES_Emitter, "initialize_copy", init_copy, 1);
 546         rb_define_alloc_func(cLWES_Emitter, rle_alloc);
 547         LWESRB_MKID(TYPE_DB);
 548         LWESRB_MKID(TYPE_LIST);
 549         LWESRB_MKID(NAME);
 550         LWESRB_MKID(HAVE_ENCODING);
 551         LWESRB_MKID(new);
 552         LWESRB_MKID(size);
 553         id_enc = rb_intern(LWES_ENCODING);
 554         id_to_a = rb_intern("to_a");
 555         sym_enc = ID2SYM(id_enc);
 556
 557         ENC = rb_obj_freeze(rb_str_new2(LWES_ENCODING));
 558
 559         /*
 560          * the key in an LWES::Event to designate the encoding of
 561          * an event, currently "enc"
 562          */
 563         rb_define_const(mLWES, "ENCODING", ENC);
 564 }