Class: Iodine::PubSub::Redis

Inherits:

Engine

• Object
• Engine
• Iodine::PubSub::Redis

Defined in:

ext/iodine_ext/iodine_pubsub.c

Instance Method Summary

• #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).

# File 'ext/iodine_ext/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

}

# File 'ext/iodine_ext/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.");
}