Class: Iodine::PubSub::Redis

Inherits:
Engine
  • Object
show all
Defined in:
ext/iodine/iodine_pubsub.c

Instance Method Summary collapse

Methods inherited from Engine

#publish, #subscribe, #unsubscribe

Constructor Details

#initialize(*args) ⇒ Object

Initializes a new Iodine::PubSub::Redis engine.

Iodine::PubSub::Redis.new(url, opt = {})

use:

REDIS_URL = "redis://localhost:6379/"
Iodine::PubSub::Redis.new(REDIS_URL, ping: 50) #pings every 50 seconds

To use Redis authentication, add the password to the URL. i.e.:

REDIS_URL = "redis://redis:password@localhost:6379/"
Iodine::PubSub::Redis.new(REDIS_URL, ping: 50) #pings every 50 seconds

The options hash accepts:

:ping:: the PING interval up to 255 seconds. Default: 0 (~5 minutes).



359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
# File 'ext/iodine/iodine_pubsub.c', line 359

static VALUE iodine_pubsub_redis_new(int argc, VALUE *argv, VALUE self) {
  if (!argc) {
    rb_raise(rb_eArgError, "Iodine::PubSub::Redis.new(address, opt={}) "
                           "requires at least 1 argument.");
  }
  VALUE url = argv[0];
  Check_Type(url, T_STRING);
  if (RSTRING_LEN(url) > 4096) {
    rb_raise(rb_eArgError, "Redis URL too long.");
  }
  uint8_t ping = 0;

  iodine_pubsub_s *e = iodine_pubsub_CData(self);
  if (!e) {
    rb_raise(rb_eTypeError, "not a valid engine");
    return Qnil;
  }

  /* extract options */
  if (argc == 2) {
    Check_Type(argv[1], T_HASH);
    VALUE tmp = rb_hash_aref(argv[1], rb_id2sym(rb_intern2("ping", 4)));
    if (tmp != Qnil) {
      Check_Type(tmp, T_FIXNUM);
      if (NUM2SIZET(tmp) > 255) {
        rb_raise(rb_eArgError,
                 ":ping must be a non-negative integer under 255 seconds.");
      }
      ping = (uint8_t)NUM2SIZET(tmp);
    }
  }

  /* parse URL assume redis://redis:password@localhost:6379 */
  fio_url_s info = fio_url_parse(RSTRING_PTR(url), RSTRING_LEN(url));

  FIO_LOG_INFO("Initializing Redis engine for address: %.*s",
               (int)RSTRING_LEN(url), RSTRING_PTR(url));
  /* create engine */
  e->engine = redis_engine_create(.address = info.host, .port = info.port,
                                  .auth = info.password, .ping_interval = ping);
  if (!e->engine) {
    e->engine = &e->do_not_touch;
  } else {
    e->dealloc = redis_engine_destroy;
  }

  if (e->engine == &e->do_not_touch) {
    rb_raise(rb_eArgError,
             "Error initializing the Redis engine - malformed URL?");
  }
  return self;
  (void)self;
  (void)argc;
  (void)argv;
}

Instance Method Details

#cmd(*args) ⇒ Object

Sends a Redis command. Accepts an optional block that will recieve the response.

i.e.:

REDIS_URL = "redis://redis:password@localhost:6379/"
redis = Iodine::PubSub::Redis.new(REDIS_URL, ping: 50) #pings every 50 seconds
Iodine::PubSub.default = redis
redis.cmd("KEYS", "*") {|result| p result

}



458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
# File 'ext/iodine/iodine_pubsub.c', line 458

static VALUE iodine_pubsub_redis_cmd(int argc, VALUE *argv, VALUE self) {
  // clang-format on
  if (argc <= 0) {
    rb_raise(rb_eArgError, "Iodine::PubSub::Redis#cmd(command, ...) is missing "
                           "the required command argument.");
  }
  iodine_pubsub_s *e = iodine_pubsub_CData(self);
  if (!e || !e->engine || e->engine == &e->do_not_touch) {
    rb_raise(rb_eTypeError,
             "Iodine::PubSub::Redis internal error - obsolete object?");
  }
  VALUE block = Qnil;
  if (rb_block_given_p()) {
    block = IodineStore.add(rb_block_proc());
  }
  FIOBJ data = fiobj_ary_new2((size_t)argc);
  for (int i = 0; i < argc; ++i) {
    switch (TYPE(argv[i])) {
    case T_SYMBOL:
      argv[i] = rb_sym2str(argv[i]);
    /* overflow */
    case T_STRING:
      fiobj_ary_push(data,
                     fiobj_str_new(RSTRING_PTR(argv[i]), RSTRING_LEN(argv[i])));
      break;
    case T_FIXNUM:
      fiobj_ary_push(data, fiobj_num_new(NUM2SSIZET(argv[i])));
      break;
    case T_FLOAT:
      fiobj_ary_push(data, fiobj_float_new(rb_float_value(argv[i])));
      break;
    case T_NIL:
      fiobj_ary_push(data, fiobj_null());
      break;
    case T_TRUE:
      fiobj_ary_push(data, fiobj_true());
      break;
    case T_FALSE:
      fiobj_ary_push(data, fiobj_false());
      break;
    default:
      goto wrong_type;
    }
  }
  if (redis_engine_send(e->engine, data, iodine_pubsub_redis_callback,
                        (void *)block)) {
    iodine_pubsub_redis_callback(e->engine, fiobj_null(), (void *)block);
  }
  fiobj_free(data);
  return self;

wrong_type:
  fiobj_free(data);
  rb_raise(rb_eArgError,
           "only String, Number (with limits), Symbol, true, false and nil "
           "arguments can be used.");
}