ext/lwes/emitter.c

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