LUA_FUNCTION_DEF (redis, make_request);
LUA_FUNCTION_DEF (redis, make_request_sync);
+LUA_FUNCTION_DEF (redis, connect);
+LUA_FUNCTION_DEF (redis, connect_sync);
+LUA_FUNCTION_DEF (redis, add_cmd);
+LUA_FUNCTION_DEF (redis, exec);
-static const struct luaL_reg redislib_m[] = {
+static const struct luaL_reg redislib_f[] = {
LUA_INTERFACE_DEF (redis, make_request),
LUA_INTERFACE_DEF (redis, make_request_sync),
- {"__tostring", rspamd_lua_class_tostring},
+ LUA_INTERFACE_DEF (redis, connect),
+ LUA_INTERFACE_DEF (redis, connect_sync),
{NULL, NULL}
};
+static const struct luaL_reg redislib_m[] = {
+ LUA_INTERFACE_DEF (redis, add_cmd),
+ LUA_INTERFACE_DEF (redis, exec),
+ {"__tostring", rspamd_lua_class_tostring},
+};
+
#ifdef WITH_HIREDIS
/**
* Struct for userdata representation
guint16 terminated;
};
+struct lua_redis_ctx {
+ gboolean async;
+ union {
+ struct lua_redis_userdata async;
+ redisContext *sync;
+ } d;
+ ref_entry_t ref;
+};
+
+static struct lua_redis_ctx *
+lua_check_redis (lua_State * L, gint pos)
+{
+ void *ud = luaL_checkudata (L, pos, "rspamd{redis}");
+ luaL_argcheck (L, ud != NULL, pos, "'redis' expected");
+ return ud ? *((struct lua_redis_ctx **)ud) : NULL;
+}
+
static void
lua_redis_free_args (char **args, guint nargs)
{
}
static void
-lua_redis_fin (void *arg)
+lua_redis_dtor (struct lua_redis_ctx *ctx)
{
- struct lua_redis_userdata *ud = arg;
-
- if (ud->ctx) {
- ud->terminated = 1;
- redisAsyncFree (ud->ctx);
- lua_redis_free_args (ud->args, ud->nargs);
- event_del (&ud->timeout);
- luaL_unref (ud->L, LUA_REGISTRYINDEX, ud->cbref);
+ struct lua_redis_userdata *ud;
+
+ if (ctx->async) {
+ ud = &ctx->d.async;
+
+ if (ud->ctx) {
+ ud->terminated = 1;
+ redisAsyncFree (ud->ctx);
+ lua_redis_free_args (ud->args, ud->nargs);
+ event_del (&ud->timeout);
+ luaL_unref (ud->L, LUA_REGISTRYINDEX, ud->cbref);
+ }
}
+ else {
+ if (ctx->d.sync) {
+ redisFree (ctx->d.sync);
+ }
+ }
+
+ g_slice_free1 (sizeof (*ctx), ctx);
+}
+
+static void
+lua_redis_fin (void *arg)
+{
+ struct lua_redis_ctx *ctx = arg;
+
+ REF_RELEASE (ctx);
}
/**
lua_redis_callback (redisAsyncContext *c, gpointer r, gpointer priv)
{
redisReply *reply = r;
- struct lua_redis_userdata *ud = priv;
+ struct lua_redis_ctx *ctx = priv;
+ struct lua_redis_userdata *ud;
+
+ REF_RETAIN (ctx);
+
+ ud = &ctx->d.async;
if (ud->terminated) {
/* We are already at the termination stage, just go out */
+ REF_RELEASE (ctx);
return;
}
lua_redis_push_error (c->errstr, ud, TRUE);
}
}
+ REF_RELEASE (ctx);
}
static void
lua_redis_timeout (int fd, short what, gpointer u)
{
- struct lua_redis_userdata *ud = u;
+ struct lua_redis_ctx *ctx = u;
+ struct lua_redis_userdata *ud;
+ REF_RETAIN (ctx);
+ ud = &ctx->d.async;
msg_info ("timeout while querying redis server");
lua_redis_push_error ("timeout while connecting the server", ud, TRUE);
+ REF_RELEASE (ctx);
}
* @function rspamd_redis.make_request({params})
* Make request to redis server, params is a table of key=value arguments in any order
* @param {task} task worker task object
- * @param {ip} host server address
+ * @param {ip|string} host server address
* @param {function} callback callback to be called in form `function (task, err, data)`
* @param {string} cmd command to be sent to redis
* @param {table} args numeric array of strings used as redis arguments
static int
lua_redis_make_request (lua_State *L)
{
+ struct lua_redis_ctx *ctx;
+ rspamd_inet_addr_t *ip = NULL;
struct lua_redis_userdata *ud;
struct rspamd_lua_ip *addr = NULL;
struct rspamd_task *task = NULL;
- const gchar *cmd = NULL;
+ const gchar *cmd = NULL, *host;
gint top, cbref = -1;
struct timeval tv;
gboolean ret = FALSE;
lua_pushstring (L, "host");
lua_gettable (L, -2);
+
if (lua_type (L, -1) == LUA_TUSERDATA) {
addr = lua_check_ip (L, -1);
}
+ else if (lua_type (L, -1) == LUA_TSTRING) {
+ host = lua_tostring (L, -1);
+
+ if (rspamd_parse_inet_address (&ip, host, strlen (host))) {
+ addr = g_alloca (sizeof (*addr));
+ addr->addr = ip;
+
+ if (rspamd_inet_address_get_port (ip) == 0) {
+ rspamd_inet_address_set_port (ip, 6379);
+ }
+
+ if (task) {
+ rspamd_mempool_add_destructor (task->task_pool,
+ (rspamd_mempool_destruct_t)rspamd_inet_address_destroy,
+ ip);
+ }
+ }
+ }
+
lua_pop (L, 1);
lua_pushstring (L, "timeout");
lua_gettable (L, -2);
- timeout = lua_tonumber (L, -1);
+ if (lua_type (L, -1) == LUA_TNUMBER) {
+ timeout = lua_tonumber (L, -1);
+ }
lua_pop (L, 1);
if (task != NULL && addr != NULL && cbref != -1 && cmd != NULL) {
- ud =
- rspamd_mempool_alloc (task->task_pool,
- sizeof (struct lua_redis_userdata));
+ ctx = g_slice_alloc0 (sizeof (struct lua_redis_ctx));
+ REF_INIT_RETAIN (ctx, lua_redis_dtor);
+ ctx->async = TRUE;
+ ud = &ctx->d.async;
ud->task = task;
ud->L = L;
ud->cbref = cbref;
else if ((task = lua_check_task (L, 1)) != NULL) {
addr = lua_check_ip (L, 2);
top = lua_gettop (L);
+
/* Now get callback */
if (lua_isfunction (L, 3) && addr != NULL && addr->addr && top >= 4) {
/* Create userdata */
- ud =
- rspamd_mempool_alloc (task->task_pool,
- sizeof (struct lua_redis_userdata));
+ ctx = g_slice_alloc0 (sizeof (struct lua_redis_ctx));
+ REF_INIT_RETAIN (ctx, lua_redis_dtor);
+ ctx->async = TRUE;
+ ud = &ctx->d.async;
ud->task = task;
ud->L = L;
redisAsyncSetConnectCallback (ud->ctx, lua_redis_connect_cb);
if (ud->ctx == NULL || ud->ctx->err) {
- ud->terminated = 1;
- redisAsyncFree (ud->ctx);
- lua_redis_free_args (ud->args, ud->nargs);
- luaL_unref (ud->L, LUA_REGISTRYINDEX, ud->cbref);
+ REF_RELEASE (ctx);
lua_pushboolean (L, FALSE);
return 1;
}
+
redisLibeventAttach (ud->ctx, ud->task->ev_base);
ret = redisAsyncCommandArgv (ud->ctx,
lua_redis_callback,
- ud,
+ ctx,
ud->nargs,
(const gchar **)ud->args,
NULL);
+
if (ret == REDIS_OK) {
rspamd_session_add_event (ud->task->s,
lua_redis_fin,
g_quark_from_static_string ("lua redis"));
double_to_tv (timeout, &tv);
- event_set (&ud->timeout, -1, EV_TIMEOUT, lua_redis_timeout, ud);
+ event_set (&ud->timeout, -1, EV_TIMEOUT, lua_redis_timeout, ctx);
event_base_set (ud->task->ev_base, &ud->timeout);
event_add (&ud->timeout, &tv);
}
else {
msg_info ("call to redis failed: %s", ud->ctx->errstr);
- ud->terminated = 1;
- lua_redis_free_args (ud->args, ud->nargs);
- redisAsyncFree (ud->ctx);
- luaL_unref (ud->L, LUA_REGISTRYINDEX, ud->cbref);
+ REF_RELEASE (ctx);
}
}
/***
* @function rspamd_redis.make_request_sync({params})
* Make blocking request to redis server, params is a table of key=value arguments in any order
- * @param {task} task worker task object
- * @param {ip} host server address
+ * @param {ip|string} host server address
* @param {string} cmd command to be sent to redis
* @param {table} args numeric array of strings used as redis arguments
* @param {number} timeout timeout in seconds for request (1.0 by default)
return 1;
}
+
+/***
+ * @function rspamd_redis.connect({params})
+ * Make request to redis server, params is a table of key=value arguments in any order
+ * @param {task} task worker task object
+ * @param {ip|string} host server address
+ * @param {number} timeout timeout in seconds for request (1.0 by default)
+ * @return {redis} new connection object or nil if connection failed
+ */
+static int
+lua_redis_connect (lua_State *L)
+{
+ struct rspamd_lua_ip *addr = NULL;
+ rspamd_inet_addr_t *ip = NULL;
+ const gchar *host;
+ struct timeval tv;
+ struct lua_redis_ctx *ctx = NULL, **pctx;
+ struct lua_redis_userdata *ud;
+ struct rspamd_task *task = NULL;
+ gboolean ret = FALSE;
+ gdouble timeout = REDIS_DEFAULT_TIMEOUT;
+
+ if (lua_istable (L, 1)) {
+ /* Table version */
+ lua_pushstring (L, "task");
+ lua_gettable (L, -2);
+ if (lua_type (L, -1) == LUA_TUSERDATA) {
+ task = lua_check_task (L, -1);
+ }
+ lua_pop (L, 1);
+
+
+ lua_pushstring (L, "host");
+ lua_gettable (L, -2);
+
+ if (lua_type (L, -1) == LUA_TUSERDATA) {
+ addr = lua_check_ip (L, -1);
+ }
+ else if (lua_type (L, -1) == LUA_TSTRING) {
+ host = lua_tostring (L, -1);
+
+ if (rspamd_parse_inet_address (&ip, host, strlen (host))) {
+ addr = g_alloca (sizeof (*addr));
+ addr->addr = ip;
+
+ if (rspamd_inet_address_get_port (ip) == 0) {
+ rspamd_inet_address_set_port (ip, 6379);
+ }
+
+ if (task) {
+ rspamd_mempool_add_destructor (task->task_pool,
+ (rspamd_mempool_destruct_t)rspamd_inet_address_destroy,
+ ip);
+ }
+ }
+ }
+
+ lua_pop (L, 1);
+
+ lua_pushstring (L, "timeout");
+ if (lua_type (L, -1) == LUA_TNUMBER) {
+ lua_gettable (L, -2);
+ }
+ timeout = lua_tonumber (L, -1);
+ lua_pop (L, 1);
+
+ if (task != NULL && addr != NULL) {
+ ctx = g_slice_alloc0 (sizeof (struct lua_redis_ctx));
+ REF_INIT_RETAIN (ctx, lua_redis_dtor);
+ ctx->async = TRUE;
+ ud = &ctx->d.async;
+ ud->task = task;
+ ud->L = L;
+ ud->cbref = -1;
+ ret = TRUE;
+ }
+ }
+
+ if (ret && ctx) {
+ ud->terminated = 0;
+ ud->ctx = redisAsyncConnect (rspamd_inet_address_to_string (addr->addr),
+ rspamd_inet_address_get_port (addr->addr));
+ redisAsyncSetConnectCallback (ud->ctx, lua_redis_connect_cb);
+
+ if (ud->ctx == NULL || ud->ctx->err) {
+ REF_RELEASE (ctx);
+ lua_pushboolean (L, FALSE);
+
+ return 1;
+ }
+
+ redisLibeventAttach (ud->ctx, ud->task->ev_base);
+ pctx = lua_newuserdata (L, sizeof (ctx));
+ *pctx = ctx;
+ rspamd_lua_setclass (L, "rspamd{redis}", -1);
+ }
+ else {
+ lua_pushnil (L);
+ }
+
+ return 1;
+}
+
+static int
+lua_redis_connect_sync (lua_State *L)
+{
+ struct rspamd_lua_ip *addr = NULL;
+ rspamd_inet_addr_t *ip = NULL;
+ const gchar *host;
+ struct timeval tv;
+ gboolean ret = FALSE;
+ gdouble timeout = REDIS_DEFAULT_TIMEOUT;
+ struct lua_redis_ctx *ctx, **pctx;
+ redisReply *r;
+
+ if (lua_istable (L, 1)) {
+ lua_pushstring (L, "host");
+ lua_gettable (L, -2);
+ if (lua_type (L, -1) == LUA_TUSERDATA) {
+ addr = lua_check_ip (L, -1);
+ }
+ else if (lua_type (L, -1) == LUA_TSTRING) {
+ host = lua_tostring (L, -1);
+ if (rspamd_parse_inet_address (&ip, host, strlen (host))) {
+ addr = g_alloca (sizeof (*addr));
+ addr->addr = ip;
+
+ if (rspamd_inet_address_get_port (ip) == 0) {
+ rspamd_inet_address_set_port (ip, 6379);
+ }
+ }
+ }
+ lua_pop (L, 1);
+
+ lua_pushstring (L, "timeout");
+ lua_gettable (L, -2);
+ if (lua_type (L, -1) == LUA_TNUMBER) {
+ timeout = lua_tonumber (L, -1);
+ }
+ lua_pop (L, 1);
+
+ if (addr) {
+ ret = TRUE;
+ }
+ }
+
+ if (ret) {
+ double_to_tv (timeout, &tv);
+ ctx = g_slice_alloc0 (sizeof (struct lua_redis_ctx));
+ REF_INIT_RETAIN (ctx, lua_redis_dtor);
+ ctx->async = FALSE;
+ ctx->d.sync = redisConnectWithTimeout (
+ rspamd_inet_address_to_string (addr->addr),
+ rspamd_inet_address_get_port (addr->addr), tv);
+
+ if (ip) {
+ rspamd_inet_address_destroy (ip);
+ }
+
+ if (ctx->d.sync == NULL || ctx->d.sync->err) {
+ REF_RELEASE (ctx);
+ lua_pushboolean (L, FALSE);
+
+ return 1;
+ }
+
+ pctx = lua_newuserdata (L, sizeof (ctx));
+ *pctx = ctx;
+ rspamd_lua_setclass (L, "rspamd{redis}", -1);
+
+ }
+ else {
+ if (ip) {
+ rspamd_inet_address_destroy (ip);
+ }
+ msg_err ("bad arguments for redis request");
+ lua_pushboolean (L, FALSE);
+ }
+
+ return 1;
+}
+
+static int
+lua_redis_add_cmd (lua_State *L)
+{
+ msg_warn ("rspamd is compiled with no redis support");
+
+ lua_pushboolean (L, FALSE);
+
+ return 1;
+}
+static int
+lua_redis_exec (lua_State *L)
+{
+ msg_warn ("rspamd is compiled with no redis support");
+
+ lua_pushboolean (L, FALSE);
+
+ return 1;
+}
#else
static int
lua_redis_make_request (lua_State *L)
return 1;
}
+static int
+lua_redis_connect (lua_State *L)
+{
+ msg_warn ("rspamd is compiled with no redis support");
+
+ lua_pushboolean (L, FALSE);
+
+ return 1;
+}
+static int
+lua_redis_connect_sync (lua_State *L)
+{
+ msg_warn ("rspamd is compiled with no redis support");
+
+ lua_pushboolean (L, FALSE);
+
+ return 1;
+}
+static int
+lua_redis_add_cmd (lua_State *L)
+{
+ msg_warn ("rspamd is compiled with no redis support");
+
+ lua_pushboolean (L, FALSE);
+
+ return 1;
+}
+static int
+lua_redis_exec (lua_State *L)
+{
+ msg_warn ("rspamd is compiled with no redis support");
+
+ lua_pushboolean (L, FALSE);
+
+ return 1;
+}
#endif
static gint
lua_load_redis (lua_State * L)
{
lua_newtable (L);
- luaL_register (L, NULL, redislib_m);
+ luaL_register (L, NULL, redislib_f);
return 1;
}
void
luaopen_redis (lua_State * L)
{
+ luaL_newmetatable (L, "rspamd{redis}");
+ lua_pushstring (L, "__index");
+ lua_pushvalue (L, -2);
+ lua_settable (L, -3);
+
+ lua_pushstring (L, "class");
+ lua_pushstring (L, "rspamd{redis}");
+ lua_rawset (L, -3);
+
+ luaL_register (L, NULL, redislib_m);
+ lua_pop (L, 1);
+
rspamd_lua_add_preload (L, "rspamd_redis", lua_load_redis);
}
projects / rspamd.git / commitdiff