Class: Iodine::PubSub::Redis

Inherits:

Engine

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

Defined in:

ext/iodine/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/iodine_pubsub.c', line 395

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.");
  }
  FIOBJ port = FIOBJ_INVALID;
  FIOBJ address = FIOBJ_INVALID;
  FIOBJ auth = FIOBJ_INVALID;
  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 */
  {
    size_t l = RSTRING_LEN(url);
    char *str = RSTRING_PTR(url);
    char *pointers[5];
    char *end = str + l;
    uint8_t flag = 1;
    uint8_t counter = 0;
    for (size_t i = 0; i < l; i++) {
      if (counter > 4)
        goto finish;
      if (str[i] == ':' && str[i + 1] == '/' && str[i + 2] == '/') {
        pointers[counter++] = str + i + 3;
        i = i + 2;
        flag = 0;
        continue;
      }
      if (str[i] == '@' && counter == 1 - flag) {
        rb_raise(rb_eArgError, "malformed URL");
      }
      if (str[i] == ':' || str[i] == '@') {
        pointers[counter++] = str + i + 1;
        continue;
      }
      if (str[i] == '/') {
        end = str + i;
        break;
      }
    }
    if (flag) {
      if (counter > 3) {
        rb_raise(rb_eArgError, "malformed URL");
      }
      /* move pointers one step forward and set 0 to str... */
      char *pointers_2[5];
      for (size_t i = 0; i < counter; ++i) {
        pointers_2[i + 1] = pointers[i];
      }
      pointers_2[0] = str;
      ++counter;
      for (size_t i = 0; i < counter; ++i) {
        pointers[i] = pointers_2[i];
      }
    }
    /* review results */
    switch (counter) {
    case 1:
      /* redis://localhost */
      if (pointers[0] == end) {
        goto finish;
      }
      address = fiobj_str_new(pointers[0], end - pointers[0]);
      break;
    case 2:
      /* redis://localhost:6379 */
      if (pointers[1] - pointers[0] - 1 == 0) {
        goto finish;
      }
      address = fiobj_str_new(pointers[0], pointers[1] - pointers[0] - 1);
      if (pointers[1] != end) {
        port = fiobj_str_new(pointers[1], end - pointers[1]);
      }
      break;
    case 3:
      /* redis://redis:password@localhost */
      if (pointers[2] - pointers[1] - 1 == 0 || end - pointers[2] == 0) {
        goto finish;
      }
      address = fiobj_str_new(pointers[2], end - pointers[2]);
      auth = fiobj_str_new(pointers[1], pointers[2] - pointers[1] - 1);
      break;
    case 4:
      /* redis://redis:password@localhost:6379 */
      if (pointers[2] - pointers[1] - 1 == 0 ||
          pointers[3] - pointers[2] - 1 == 0 || end - pointers[3] == 0) {
        goto finish;
      }
      port = fiobj_str_new(pointers[3], end - pointers[3]);
      address = fiobj_str_new(pointers[2], pointers[3] - pointers[2] - 1);
      auth = fiobj_str_new(pointers[1], pointers[2] - pointers[1] - 1);
      break;
    default:
      goto finish;
    }
  }
  fprintf(
      stderr,
      "INFO: Initializing Redis engine for address: %s - port: %s -  auth %s\n",
      fiobj_obj2cstr(address).data, fiobj_obj2cstr(port).data,
      fiobj_obj2cstr(auth).data);
  /* create engine */
  e->engine = redis_engine_create(
          .address = fiobj_obj2cstr(address)
          .data,
          .port = (port == FIOBJ_INVALID ? "6379" : fiobj_obj2cstr(port).data),
          .ping_interval = ping,
          .auth = (auth == FIOBJ_INVALID ? NULL : fiobj_obj2cstr(auth).data),
          .auth_len = (auth == FIOBJ_INVALID ? 0 : fiobj_obj2cstr(auth).len));
  if (!e->engine) {
    e->engine = &e->do_not_touch;
  } else {
    e->dealloc = redis_engine_destroy;
  }

finish:
  fiobj_free(port);
  fiobj_free(address);
  fiobj_free(auth);
  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/iodine_pubsub.c', line 579

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;
    }
  }
  FIOBJ cmd = fiobj_ary_shift(data);
  if (redis_engine_send(e->engine, cmd, data, iodine_pubsub_redis_callback,
                        (void *)block)) {
    iodine_pubsub_redis_callback(e->engine, fiobj_null(), (void *)block);
  }
  fiobj_free(cmd);
  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.");
}