Class: Iodine::PubSub::Redis
- Defined in:
- ext/iodine/iodine_pubsub.c
Instance Method Summary collapse
-
#cmd(*args) ⇒ Object
Sends a Redis command.
-
#initialize(*args) ⇒ Object
constructor
Initializes a new Redis engine.
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).
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 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 |
# File 'ext/iodine/iodine_pubsub.c', line 380
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 */
http_url_s info = http_url_parse(RSTRING_PTR(url), RSTRING_LEN(url));
FIO_LOG_INFO(
"Initializing Redis engine for address: %s - port: %s - auth %s\n",
info.host.data ? info.host.data : "-",
info.port.data ? info.port.data : "-",
info.password.data ? info.password.data : "-");
/* 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
}
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 515 516 517 518 519 520 521 522 523 524 525 526 |
# File 'ext/iodine/iodine_pubsub.c', line 470
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.");
}
|