ext/lwes/emitter.c

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