12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619 |
- /*-
- * Copyright 2016 Vsevolod Stakhov
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
- #include "lua_common.h"
- #include "lua_thread_pool.h"
- #include "utlist.h"
-
- #include "contrib/hiredis/hiredis.h"
- #include "contrib/hiredis/async.h"
-
- #define REDIS_DEFAULT_TIMEOUT 1.0
-
- static const gchar *M = "rspamd lua redis";
-
- /***
- * @module rspamd_redis
- * This module implements redis asynchronous client for rspamd LUA API.
- * Here is an example of using of this module:
- * @example
- local rspamd_redis = require "rspamd_redis"
- local rspamd_logger = require "rspamd_logger"
-
- local function symbol_callback(task)
- local redis_key = 'some_key'
- local function redis_cb(err, data)
- if not err then
- rspamd_logger.infox('redis returned %1=%2', redis_key, data)
- end
- end
-
- rspamd_redis.make_request(task, "127.0.0.1:6379", redis_cb,
- 'GET', {redis_key})
- -- or in table form:
- -- rspamd_redis.make_request({task=task, host="127.0.0.1:6379,
- -- callback=redis_cb, timeout=2.0, cmd='GET', args={redis_key}})
- end
- */
-
- 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);
- LUA_FUNCTION_DEF (redis, gc);
-
- static const struct luaL_reg redislib_f[] = {
- LUA_INTERFACE_DEF (redis, make_request),
- LUA_INTERFACE_DEF (redis, make_request_sync),
- 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),
- {"__gc", lua_redis_gc},
- {"__tostring", rspamd_lua_class_tostring},
- {NULL, NULL}
- };
-
- #undef REDIS_DEBUG_REFS
- #ifdef REDIS_DEBUG_REFS
- #define REDIS_RETAIN(x) do { \
- msg_err ("retain ref %p, refcount: %d", (x), (x)->ref.refcount); \
- REF_RETAIN(x); \
- } while (0)
-
- #define REDIS_RELEASE(x) do { \
- msg_err ("release ref %p, refcount: %d", (x), (x)->ref.refcount); \
- REF_RELEASE(x); \
- } while (0)
- #else
- #define REDIS_RETAIN REF_RETAIN
- #define REDIS_RELEASE REF_RELEASE
- #endif
-
- #ifdef WITH_HIREDIS
- struct lua_redis_request_specific_userdata;
- /**
- * Struct for userdata representation
- */
- struct lua_redis_userdata {
- redisAsyncContext *ctx;
- struct rspamd_task *task;
- struct rspamd_symcache_item *item;
- struct rspamd_async_session *s;
- struct ev_loop *event_loop;
- struct rspamd_config *cfg;
- struct rspamd_redis_pool *pool;
- gchar *server;
- struct lua_redis_request_specific_userdata *specific;
- gdouble timeout;
- guint16 port;
- guint16 terminated;
- };
-
- #define LUA_REDIS_SPECIFIC_REPLIED (1 << 0)
- /* session was finished */
- #define LUA_REDIS_SPECIFIC_FINISHED (1 << 1)
- #define LUA_REDIS_ASYNC (1 << 0)
- #define LUA_REDIS_TEXTDATA (1 << 1)
- #define LUA_REDIS_TERMINATED (1 << 2)
- #define LUA_REDIS_NO_POOL (1 << 3)
- #define IS_ASYNC(ctx) ((ctx)->flags & LUA_REDIS_ASYNC)
-
- struct lua_redis_request_specific_userdata {
- gint cbref;
- guint nargs;
- gchar **args;
- gsize *arglens;
- struct lua_redis_userdata *c;
- struct lua_redis_ctx *ctx;
- struct lua_redis_request_specific_userdata *next;
- ev_timer timeout_ev;
- guint flags;
- };
-
- struct lua_redis_ctx {
- guint flags;
- struct lua_redis_userdata async;
- guint cmds_pending;
- ref_entry_t ref;
- GQueue *replies; /* for sync connection only */
- GQueue *events_cleanup; /* for sync connection only */
- struct thread_entry *thread; /* for sync mode, set only if there was yield */
- };
-
- struct lua_redis_result {
- gboolean is_error;
- gint result_ref;
- struct rspamd_symcache_item *item;
- struct rspamd_async_session *s;
- struct rspamd_task *task;
- struct lua_redis_request_specific_userdata *sp_ud;
- };
-
- static struct lua_redis_ctx *
- lua_check_redis (lua_State * L, gint pos)
- {
- void *ud = rspamd_lua_check_udata (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, gsize *arglens, guint nargs)
- {
- guint i;
-
- if (args) {
- for (i = 0; i < nargs; i ++) {
- g_free (args[i]);
- }
-
- g_free (args);
- g_free (arglens);
- }
- }
-
- static void
- lua_redis_dtor (struct lua_redis_ctx *ctx)
- {
- struct lua_redis_userdata *ud;
- struct lua_redis_request_specific_userdata *cur, *tmp;
- gboolean is_successful = TRUE;
- struct redisAsyncContext *ac;
-
- ud = &ctx->async;
- msg_debug ("desctructing %p", ctx);
-
- if (ud->ctx) {
-
- LL_FOREACH_SAFE (ud->specific, cur, tmp) {
- ev_timer_stop (ud->event_loop, &cur->timeout_ev);
-
- if (!(cur->flags & LUA_REDIS_SPECIFIC_REPLIED)) {
- is_successful = FALSE;
- }
-
- cur->flags |= LUA_REDIS_SPECIFIC_FINISHED;
- }
-
- ctx->flags |= LUA_REDIS_TERMINATED;
-
- ud->terminated = 1;
- ac = ud->ctx;
- ud->ctx = NULL;
-
- if (!is_successful) {
- rspamd_redis_pool_release_connection (ud->pool, ac,
- RSPAMD_REDIS_RELEASE_FATAL);
- }
- else {
- rspamd_redis_pool_release_connection (ud->pool, ac,
- (ctx->flags & LUA_REDIS_NO_POOL) ?
- RSPAMD_REDIS_RELEASE_ENFORCE : RSPAMD_REDIS_RELEASE_DEFAULT);
- }
-
- }
-
- LL_FOREACH_SAFE (ud->specific, cur, tmp) {
- lua_redis_free_args (cur->args, cur->arglens, cur->nargs);
-
- if (cur->cbref != -1) {
- luaL_unref (ud->cfg->lua_state, LUA_REGISTRYINDEX, cur->cbref);
- }
-
- g_free (cur);
- }
-
- if (ctx->events_cleanup) {
- g_queue_free (ctx->events_cleanup);
- ctx->events_cleanup = NULL;
- }
- if (ctx->replies) {
- g_queue_free (ctx->replies);
- ctx->replies = NULL;
- }
-
- g_free (ctx);
- }
-
- static gint
- lua_redis_gc (lua_State *L)
- {
- struct lua_redis_ctx *ctx = lua_check_redis (L, 1);
-
- if (ctx) {
- REDIS_RELEASE (ctx);
- }
-
- return 0;
- }
-
- static void
- lua_redis_fin (void *arg)
- {
- struct lua_redis_request_specific_userdata *sp_ud = arg;
- struct lua_redis_ctx *ctx;
-
- ctx = sp_ud->ctx;
-
- ev_timer_stop (sp_ud->ctx->async.event_loop, &sp_ud->timeout_ev);
-
- msg_debug ("finished redis query %p from session %p", sp_ud, ctx);
- sp_ud->flags |= LUA_REDIS_SPECIFIC_FINISHED;
-
- REDIS_RELEASE (ctx);
- }
-
- /**
- * Push error of redis request to lua callback
- * @param code
- * @param ud
- */
- static void
- lua_redis_push_error (const gchar *err,
- struct lua_redis_ctx *ctx,
- struct lua_redis_request_specific_userdata *sp_ud,
- gboolean connected)
- {
- struct lua_redis_userdata *ud = sp_ud->c;
- struct lua_callback_state cbs;
- lua_State *L;
-
- if (!(sp_ud->flags & (LUA_REDIS_SPECIFIC_REPLIED|LUA_REDIS_SPECIFIC_FINISHED))) {
- if (sp_ud->cbref != -1) {
-
- lua_thread_pool_prepare_callback (ud->cfg->lua_thread_pool, &cbs);
- L = cbs.L;
-
- lua_pushcfunction (L, &rspamd_lua_traceback);
- int err_idx = lua_gettop (L);
- /* Push error */
- lua_rawgeti (cbs.L, LUA_REGISTRYINDEX, sp_ud->cbref);
-
- /* String of error */
- lua_pushstring (cbs.L, err);
- /* Data is nil */
- lua_pushnil (cbs.L);
-
- if (ud->item) {
- rspamd_symcache_set_cur_item (ud->task, ud->item);
- }
-
- if (lua_pcall (cbs.L, 2, 0, err_idx) != 0) {
- msg_info ("call to callback failed: %s", lua_tostring (cbs.L, -1));
- }
-
- lua_settop (L, err_idx - 1);
- lua_thread_pool_restore_callback (&cbs);
- }
-
- sp_ud->flags |= LUA_REDIS_SPECIFIC_REPLIED;
-
- if (connected && ud->s) {
- if (ud->item) {
- rspamd_symcache_item_async_dec_check (ud->task, ud->item, M);
- }
-
- rspamd_session_remove_event (ud->s, lua_redis_fin, sp_ud);
- }
- else {
- lua_redis_fin (sp_ud);
- }
- }
- }
-
- static void
- lua_redis_push_reply (lua_State *L, const redisReply *r, gboolean text_data)
- {
- guint i;
- struct rspamd_lua_text *t;
-
- switch (r->type) {
- case REDIS_REPLY_INTEGER:
- lua_pushinteger (L, r->integer);
- break;
- case REDIS_REPLY_NIL:
- /* XXX: not the best approach */
- lua_newuserdata (L, sizeof (gpointer));
- break;
- case REDIS_REPLY_STRING:
- case REDIS_REPLY_STATUS:
- if (text_data) {
- t = lua_newuserdata (L, sizeof (*t));
- rspamd_lua_setclass (L, "rspamd{text}", -1);
- t->flags = 0;
- t->start = r->str;
- t->len = r->len;
- }
- else {
- lua_pushlstring (L, r->str, r->len);
- }
- break;
- case REDIS_REPLY_ARRAY:
- lua_createtable (L, r->elements, 0);
- for (i = 0; i < r->elements; ++i) {
- lua_redis_push_reply (L, r->element[i], text_data);
- lua_rawseti (L, -2, i + 1); /* Store sub-reply */
- }
- break;
- default: /* should not happen */
- msg_info ("unknown reply type: %d", r->type);
- break;
- }
- }
-
- /**
- * Push data of redis request to lua callback
- * @param r redis reply data
- * @param ud
- */
- static void
- lua_redis_push_data (const redisReply *r, struct lua_redis_ctx *ctx,
- struct lua_redis_request_specific_userdata *sp_ud)
- {
- struct lua_redis_userdata *ud = sp_ud->c;
- struct lua_callback_state cbs;
- lua_State *L;
-
- if (!(sp_ud->flags & (LUA_REDIS_SPECIFIC_REPLIED|LUA_REDIS_SPECIFIC_FINISHED))) {
- if (sp_ud->cbref != -1) {
- lua_thread_pool_prepare_callback (ud->cfg->lua_thread_pool, &cbs);
- L = cbs.L;
-
- lua_pushcfunction (L, &rspamd_lua_traceback);
- int err_idx = lua_gettop (L);
- /* Push error */
- lua_rawgeti (cbs.L, LUA_REGISTRYINDEX, sp_ud->cbref);
- /* Error is nil */
- lua_pushnil (cbs.L);
- /* Data */
- lua_redis_push_reply (cbs.L, r, ctx->flags & LUA_REDIS_TEXTDATA);
-
- if (ud->item) {
- rspamd_symcache_set_cur_item (ud->task, ud->item);
- }
-
- if (lua_pcall (cbs.L, 2, 0, err_idx) != 0) {
- msg_info ("call to callback failed: %s", lua_tostring (cbs.L, -1));
- }
-
- lua_settop (L, err_idx - 1);
- lua_thread_pool_restore_callback (&cbs);
- }
-
- sp_ud->flags |= LUA_REDIS_SPECIFIC_REPLIED;
-
- if (ud->s) {
- if (ud->item) {
- rspamd_symcache_item_async_dec_check (ud->task, ud->item, M);
- }
-
- rspamd_session_remove_event (ud->s, lua_redis_fin, sp_ud);
- }
- else {
- lua_redis_fin (sp_ud);
- }
- }
- }
-
- /**
- * Callback for redis replies
- * @param c context of redis connection
- * @param r redis reply
- * @param priv userdata
- */
- static void
- lua_redis_callback (redisAsyncContext *c, gpointer r, gpointer priv)
- {
- redisReply *reply = r;
- struct lua_redis_request_specific_userdata *sp_ud = priv;
- struct lua_redis_ctx *ctx;
- struct lua_redis_userdata *ud;
- redisAsyncContext *ac;
-
- ctx = sp_ud->ctx;
- ud = sp_ud->c;
-
- if (ud->terminated) {
- /* We are already at the termination stage, just go out */
- return;
- }
-
- msg_debug ("got reply from redis %p for query %p", sp_ud->c->ctx, sp_ud);
-
- REDIS_RETAIN (ctx);
-
- /* If session is finished, we cannot call lua callbacks */
- if (!(sp_ud->flags & LUA_REDIS_SPECIFIC_FINISHED)) {
- if (c->err == 0) {
- if (r != NULL) {
- if (reply->type != REDIS_REPLY_ERROR) {
- lua_redis_push_data (reply, ctx, sp_ud);
- }
- else {
- lua_redis_push_error (reply->str, ctx, sp_ud, TRUE);
- }
- }
- else {
- lua_redis_push_error ("received no data from server", ctx, sp_ud, TRUE);
- }
- }
- else {
- if (c->err == REDIS_ERR_IO) {
- lua_redis_push_error (strerror (errno), ctx, sp_ud, TRUE);
- }
- else {
- lua_redis_push_error (c->errstr, ctx, sp_ud, TRUE);
- }
- }
- }
-
- ctx->cmds_pending --;
-
- if (ctx->cmds_pending == 0 && !ud->terminated) {
- /* Disconnect redis early as we don't need it anymore */
- ud->terminated = 1;
- ac = ud->ctx;
- ud->ctx = NULL;
-
- if (ac) {
- rspamd_redis_pool_release_connection (ud->pool, ac,
- (ctx->flags & LUA_REDIS_NO_POOL) ?
- RSPAMD_REDIS_RELEASE_ENFORCE : RSPAMD_REDIS_RELEASE_DEFAULT);
- }
- }
-
- REDIS_RELEASE (ctx);
- }
-
- static gint
- lua_redis_push_results (struct lua_redis_ctx *ctx, lua_State *L)
- {
- gint results = g_queue_get_length (ctx->replies);
- gint i;
- gboolean can_use_lua = TRUE;
-
- results = g_queue_get_length (ctx->replies);
-
- if (!lua_checkstack (L, (results * 2) + 1)) {
- luaL_error (L, "cannot resize stack to fit %d commands",
- ctx->cmds_pending);
-
- can_use_lua = FALSE;
- }
-
- for (i = 0; i < results; i ++) {
- struct lua_redis_result *result = g_queue_pop_head (ctx->replies);
-
- if (can_use_lua) {
- lua_pushboolean (L, !result->is_error);
- lua_rawgeti (L, LUA_REGISTRYINDEX, result->result_ref);
- }
-
- luaL_unref (L, LUA_REGISTRYINDEX, result->result_ref);
-
- g_queue_push_tail (ctx->events_cleanup, result);
- }
-
- return can_use_lua ? results * 2 : 0;
- }
-
- static void
- lua_redis_cleanup_events (struct lua_redis_ctx *ctx)
- {
- REDIS_RETAIN (ctx); /* To avoid preliminary destruction */
-
- while (!g_queue_is_empty (ctx->events_cleanup)) {
- struct lua_redis_result *result = g_queue_pop_head (ctx->events_cleanup);
-
- if (result->item) {
- rspamd_symcache_item_async_dec_check (result->task, result->item, M);
- }
-
- if (result->s) {
- rspamd_session_remove_event (result->s, lua_redis_fin, result->sp_ud);
- }
- else {
- lua_redis_fin (result->sp_ud);
- }
-
- g_free (result);
- }
-
- REDIS_RELEASE (ctx);
- }
-
- /**
- * Callback for redis replies
- * @param c context of redis connection
- * @param r redis reply
- * @param priv userdata
- */
- static void
- lua_redis_callback_sync (redisAsyncContext *ac, gpointer r, gpointer priv)
- {
- redisReply *reply = r;
-
- struct lua_redis_request_specific_userdata *sp_ud = priv;
- struct lua_redis_ctx *ctx;
- struct lua_redis_userdata *ud;
- struct thread_entry* thread;
- gint results;
-
- ctx = sp_ud->ctx;
- ud = sp_ud->c;
- lua_State *L = ctx->async.cfg->lua_state;
-
- sp_ud->flags |= LUA_REDIS_SPECIFIC_REPLIED;
-
- if (ud->terminated) {
- /* We are already at the termination stage, just go out */
- /* TODO:
- if somebody is waiting for us (ctx->thread), return result,
- otherwise, indeed, ignore
- */
- return;
- }
-
- ev_timer_stop (ud->event_loop, &sp_ud->timeout_ev);
- msg_debug ("got reply from redis: %p for query %p", ac, sp_ud);
-
- struct lua_redis_result *result = g_malloc0 (sizeof *result);
-
- /* If session is finished, we cannot call lua callbacks */
- if (ac->err == 0) {
- if (r != NULL) {
- if (reply->type != REDIS_REPLY_ERROR) {
- result->is_error = FALSE;
- lua_redis_push_reply (L, reply, ctx->flags & LUA_REDIS_TEXTDATA);
- }
- else {
- result->is_error = TRUE;
- lua_pushstring (L, reply->str);
- }
- }
- else {
- result->is_error = TRUE;
- lua_pushliteral (L, "received no data from server");
- }
- }
- else {
- result->is_error = TRUE;
- if (ac->err == REDIS_ERR_IO) {
- lua_pushstring (L, strerror (errno));
- }
- else {
- lua_pushstring (L, ac->errstr);
- }
- }
- /* if error happened, we should terminate the connection,
- and release it */
-
- if (result->is_error && sp_ud->c->ctx) {
- ac = sp_ud->c->ctx;
- /* Set to NULL to avoid double free in dtor */
- sp_ud->c->ctx = NULL;
- ctx->flags |= LUA_REDIS_TERMINATED;
-
- /*
- * This will call all callbacks pending so the entire context
- * will be destructed
- */
- rspamd_redis_pool_release_connection (sp_ud->c->pool, ac,
- RSPAMD_REDIS_RELEASE_FATAL);
- }
-
- result->result_ref = luaL_ref (L, LUA_REGISTRYINDEX);
- result->s = ud->s;
- result->item = ud->item;
- result->task = ud->task;
- result->sp_ud = sp_ud;
-
- g_queue_push_tail (ctx->replies, result);
-
- ctx->cmds_pending --;
-
- if (ctx->cmds_pending == 0) {
- if (ctx->thread) {
- /* somebody yielded and waits for results */
- thread = ctx->thread;
- ctx->thread = NULL;
-
- results = lua_redis_push_results (ctx, thread->lua_state);
- lua_thread_resume (thread, results);
- lua_redis_cleanup_events (ctx);
- }
- }
- }
-
- static void
- lua_redis_timeout_sync (EV_P_ ev_timer *w, int revents)
- {
- struct lua_redis_request_specific_userdata *sp_ud =
- (struct lua_redis_request_specific_userdata *)w->data;
- struct lua_redis_ctx *ctx = sp_ud->ctx;
- redisAsyncContext *ac;
-
- msg_debug ("timeout while querying redis server: %p, redis: %p", sp_ud,
- sp_ud->c->ctx);
-
- if (sp_ud->c->ctx) {
- ac = sp_ud->c->ctx;
-
- /* Set to NULL to avoid double free in dtor */
- sp_ud->c->ctx = NULL;
- ac->err = REDIS_ERR_IO;
- errno = ETIMEDOUT;
- ctx->flags |= LUA_REDIS_TERMINATED;
-
- /*
- * This will call all callbacks pending so the entire context
- * will be destructed
- */
- rspamd_redis_pool_release_connection (sp_ud->c->pool, ac,
- RSPAMD_REDIS_RELEASE_FATAL);
- }
- }
-
- static void
- lua_redis_timeout (EV_P_ ev_timer *w, int revents)
- {
- struct lua_redis_request_specific_userdata *sp_ud =
- (struct lua_redis_request_specific_userdata *)w->data;
- struct lua_redis_ctx *ctx;
- redisAsyncContext *ac;
-
- if (sp_ud->flags & LUA_REDIS_SPECIFIC_FINISHED) {
- return;
- }
-
- ctx = sp_ud->ctx;
-
- REDIS_RETAIN (ctx);
- msg_debug ("timeout while querying redis server: %p, redis: %p", sp_ud,
- sp_ud->c->ctx);
- lua_redis_push_error ("timeout while connecting the server", ctx, sp_ud, TRUE);
-
- if (sp_ud->c->ctx) {
- ac = sp_ud->c->ctx;
- /* Set to NULL to avoid double free in dtor */
- sp_ud->c->ctx = NULL;
- ac->err = REDIS_ERR_IO;
- errno = ETIMEDOUT;
- /*
- * This will call all callbacks pending so the entire context
- * will be destructed
- */
- rspamd_redis_pool_release_connection (sp_ud->c->pool, ac,
- RSPAMD_REDIS_RELEASE_FATAL);
- }
-
- REDIS_RELEASE (ctx);
- }
-
-
- static void
- lua_redis_parse_args (lua_State *L, gint idx, const gchar *cmd,
- gchar ***pargs, gsize **parglens, guint *nargs)
- {
- gchar **args = NULL;
- gsize *arglens;
- gint top;
-
- if (idx != 0 && lua_type (L, idx) == LUA_TTABLE) {
- /* Get all arguments */
- lua_pushvalue (L, idx);
- lua_pushnil (L);
- top = 0;
-
- while (lua_next (L, -2) != 0) {
- gint type = lua_type (L, -1);
-
- if (type == LUA_TNUMBER || type == LUA_TSTRING ||
- type == LUA_TUSERDATA) {
- top ++;
- }
- lua_pop (L, 1);
- }
-
- args = g_malloc ((top + 1) * sizeof (gchar *));
- arglens = g_malloc ((top + 1) * sizeof (gsize));
- arglens[0] = strlen (cmd);
- args[0] = g_malloc (arglens[0]);
- memcpy (args[0], cmd, arglens[0]);
- top = 1;
- lua_pushnil (L);
-
- while (lua_next (L, -2) != 0) {
- gint type = lua_type (L, -1);
-
- if (type == LUA_TSTRING) {
- const gchar *s;
-
- s = lua_tolstring (L, -1, &arglens[top]);
- args[top] = g_malloc (arglens[top]);
- memcpy (args[top], s, arglens[top]);
- top ++;
- }
- else if (type == LUA_TUSERDATA) {
- struct rspamd_lua_text *t;
-
- t = lua_check_text (L, -1);
-
- if (t && t->start) {
- arglens[top] = t->len;
- args[top] = g_malloc (arglens[top]);
- memcpy (args[top], t->start, arglens[top]);
- top ++;
- }
- }
- else if (type == LUA_TNUMBER) {
- gdouble val = lua_tonumber (L, -1);
- gint r;
- gchar numbuf[64];
-
- if (val == (gdouble)((gint64)val)) {
- r = rspamd_snprintf (numbuf, sizeof (numbuf), "%L",
- (gint64)val);
- }
- else {
- r = rspamd_snprintf (numbuf, sizeof (numbuf), "%f",
- val);
- }
-
- arglens[top] = r;
- args[top] = g_malloc (arglens[top]);
- memcpy (args[top], numbuf, arglens[top]);
- top ++;
- }
-
- lua_pop (L, 1);
- }
-
- lua_pop (L, 1);
- }
- else {
- /* Use merely cmd */
-
- args = g_malloc (sizeof (gchar *));
- arglens = g_malloc (sizeof (gsize));
- arglens[0] = strlen (cmd);
- args[0] = g_malloc (arglens[0]);
- memcpy (args[0], cmd, arglens[0]);
- top = 1;
- }
-
- *pargs = args;
- *parglens = arglens;
- *nargs = top;
- }
-
- static struct lua_redis_ctx *
- rspamd_lua_redis_prepare_connection (lua_State *L, gint *pcbref, gboolean is_async)
- {
- struct lua_redis_ctx *ctx = NULL;
- rspamd_inet_addr_t *ip = NULL;
- struct lua_redis_userdata *ud = NULL;
- struct rspamd_lua_ip *addr = NULL;
- struct rspamd_task *task = NULL;
- const gchar *host;
- const gchar *password = NULL, *dbname = NULL;
- gint cbref = -1;
- struct rspamd_config *cfg = NULL;
- struct rspamd_async_session *session = NULL;
- struct ev_loop *ev_base = NULL;
- gboolean ret = FALSE;
- guint flags = 0;
-
- if (lua_istable (L, 1)) {
- /* Table version */
- lua_pushvalue (L, 1);
- lua_pushstring (L, "task");
- lua_gettable (L, -2);
- if (lua_type (L, -1) == LUA_TUSERDATA) {
- task = lua_check_task_maybe (L, -1);
- }
- lua_pop (L, 1);
-
- if (!task) {
- /* We need to get ev_base, config and session separately */
- lua_pushstring (L, "config");
- lua_gettable (L, -2);
- if (lua_type (L, -1) == LUA_TUSERDATA) {
- cfg = lua_check_config (L, -1);
- }
- lua_pop (L, 1);
-
- lua_pushstring (L, "session");
- lua_gettable (L, -2);
- if (lua_type (L, -1) == LUA_TUSERDATA) {
- session = lua_check_session (L, -1);
- }
- lua_pop (L, 1);
-
- lua_pushstring (L, "ev_base");
- lua_gettable (L, -2);
- if (lua_type (L, -1) == LUA_TUSERDATA) {
- ev_base = lua_check_ev_base (L, -1);
- }
- lua_pop (L, 1);
-
- if (cfg && ev_base) {
- ret = TRUE;
- }
- else if (!cfg) {
- msg_err_task_check ("config is not passed");
- }
- else {
- msg_err_task_check ("ev_base is not set");
- }
- }
- else {
- cfg = task->cfg;
- session = task->s;
- ev_base = task->event_loop;
- ret = TRUE;
- }
-
- if (pcbref) {
- lua_pushstring (L, "callback");
- lua_gettable (L, -2);
- if (lua_type (L, -1) == LUA_TFUNCTION) {
- /* This also pops function from the stack */
- cbref = luaL_ref (L, LUA_REGISTRYINDEX);
- *pcbref = cbref;
- }
- else {
- *pcbref = -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);
- }
- }
- }
- lua_pop (L, 1);
-
- lua_pushstring (L, "password");
- lua_gettable (L, -2);
- if (lua_type (L, -1) == LUA_TSTRING) {
- password = lua_tostring (L, -1);
- }
- lua_pop (L, 1);
-
- lua_pushstring (L, "dbname");
- lua_gettable (L, -2);
- if (lua_type (L, -1) == LUA_TSTRING) {
- dbname = lua_tostring (L, -1);
- }
- lua_pop (L, 1);
-
- lua_pushstring (L, "opaque_data");
- lua_gettable (L, -2);
- if (!!lua_toboolean (L, -1)) {
- flags |= LUA_REDIS_TEXTDATA;
- }
- lua_pop (L, 1);
-
- lua_pushstring (L, "no_pool");
- lua_gettable (L, -2);
- if (!!lua_toboolean (L, -1)) {
- flags |= LUA_REDIS_NO_POOL;
- }
- lua_pop (L, 1);
-
- lua_pop (L, 1); /* table */
-
- if (session && rspamd_session_blocked (session)) {
- msg_err_task_check ("Session is being destroying");
- ret = FALSE;
- }
-
- if (ret && addr != NULL) {
- ctx = g_malloc0 (sizeof (struct lua_redis_ctx));
- REF_INIT_RETAIN (ctx, lua_redis_dtor);
- if (is_async) {
- ctx->flags |= flags | LUA_REDIS_ASYNC;
- ud = &ctx->async;
- }
- else {
- ud = &ctx->async;
- ctx->replies = g_queue_new ();
- ctx->events_cleanup = g_queue_new ();
-
- }
-
- ud->s = session;
- ud->cfg = cfg;
- ud->pool = cfg->redis_pool;
- ud->event_loop = ev_base;
- ud->task = task;
-
- if (task) {
- ud->item = rspamd_symcache_get_cur_item (task);
- }
-
- ret = TRUE;
- }
- else {
- if (cbref != -1) {
- luaL_unref (L, LUA_REGISTRYINDEX, cbref);
- }
-
- msg_err_task_check ("incorrect function invocation");
- ret = FALSE;
- }
- }
-
- if (ret) {
- ud->terminated = 0;
- ud->ctx = rspamd_redis_pool_connect (ud->pool,
- dbname, password,
- rspamd_inet_address_to_string (addr->addr),
- rspamd_inet_address_get_port (addr->addr));
-
- if (ip) {
- rspamd_inet_address_free (ip);
- }
-
- if (ud->ctx == NULL || ud->ctx->err) {
- if (ud->ctx) {
- msg_err_task_check ("cannot connect to redis: %s",
- ud->ctx->errstr);
- rspamd_redis_pool_release_connection (ud->pool, ud->ctx,
- RSPAMD_REDIS_RELEASE_FATAL);
- ud->ctx = NULL;
- }
- else {
- msg_err_task_check ("cannot connect to redis: unknown error");
- }
-
- REDIS_RELEASE (ctx);
-
- return NULL;
- }
-
-
- return ctx;
- }
-
- if (ip) {
- rspamd_inet_address_free (ip);
- }
-
- return NULL;
- }
-
- /***
- * @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|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
- * @param {number} timeout timeout in seconds for request (1.0 by default)
- * @return {boolean} `true` if a request has been scheduled
- */
- static int
- lua_redis_make_request (lua_State *L)
- {
- LUA_TRACE_POINT;
- struct lua_redis_request_specific_userdata *sp_ud;
- struct lua_redis_userdata *ud;
- struct lua_redis_ctx *ctx, **pctx;
- const gchar *cmd = NULL;
- gdouble timeout = REDIS_DEFAULT_TIMEOUT;
- gint cbref = -1;
- gboolean ret = FALSE;
-
- ctx = rspamd_lua_redis_prepare_connection (L, &cbref, TRUE);
-
- if (ctx) {
- ud = &ctx->async;
- sp_ud = g_malloc0 (sizeof (*sp_ud));
- sp_ud->cbref = cbref;
- sp_ud->c = ud;
- sp_ud->ctx = ctx;
-
- lua_pushstring (L, "cmd");
- lua_gettable (L, -2);
- cmd = lua_tostring (L, -1);
- lua_pop (L, 1);
-
- lua_pushstring (L, "timeout");
- lua_gettable (L, 1);
- if (lua_type (L, -1) == LUA_TNUMBER) {
- timeout = lua_tonumber (L, -1);
- }
- lua_pop (L, 1);
- ud->timeout = timeout;
-
-
- lua_pushstring (L, "args");
- lua_gettable (L, 1);
- lua_redis_parse_args (L, -1, cmd, &sp_ud->args, &sp_ud->arglens,
- &sp_ud->nargs);
- lua_pop (L, 1);
- LL_PREPEND (ud->specific, sp_ud);
-
- ret = redisAsyncCommandArgv (ud->ctx,
- lua_redis_callback,
- sp_ud,
- sp_ud->nargs,
- (const gchar **)sp_ud->args,
- sp_ud->arglens);
-
- if (ret == REDIS_OK) {
- if (ud->s) {
- rspamd_session_add_event (ud->s,
- lua_redis_fin, sp_ud,
- M);
-
- if (ud->item) {
- rspamd_symcache_item_async_inc (ud->task, ud->item, M);
- }
- }
-
- REDIS_RETAIN (ctx); /* Cleared by fin event */
- ctx->cmds_pending ++;
- sp_ud->timeout_ev.data = sp_ud;
- ev_timer_init (&sp_ud->timeout_ev, lua_redis_timeout, timeout, 0.0);
- ev_timer_start (ud->event_loop, &sp_ud->timeout_ev);
- ret = TRUE;
- }
- else {
- msg_info ("call to redis failed: %s", ud->ctx->errstr);
- rspamd_redis_pool_release_connection (ud->pool, ud->ctx,
- RSPAMD_REDIS_RELEASE_FATAL);
- ud->ctx = NULL;
- REDIS_RELEASE (ctx);
- ret = FALSE;
- }
- }
- else {
- lua_pushboolean (L, FALSE);
- lua_pushnil (L);
-
- return 2;
- }
-
- lua_pushboolean (L, ret);
-
- if (ret) {
- pctx = lua_newuserdata (L, sizeof (ctx));
- *pctx = ctx;
- rspamd_lua_setclass (L, "rspamd{redis}", -1);
- }
- else {
- lua_pushnil (L);
- }
-
- return 2;
- }
-
- /***
- * @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 {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 {boolean + result} `true` and a result if a request has been successful
- */
- static int
- lua_redis_make_request_sync (lua_State *L)
- {
- LUA_TRACE_POINT;
- struct rspamd_lua_ip *addr = NULL;
- rspamd_inet_addr_t *ip = NULL;
- const gchar *cmd = NULL, *host;
- struct timeval tv;
- gboolean ret = FALSE;
- gdouble timeout = REDIS_DEFAULT_TIMEOUT;
- gchar **args = NULL;
- gsize *arglens = NULL;
- guint nargs = 0, flags = 0;
- redisContext *ctx;
- redisReply *r;
-
- if (lua_istable (L, 1)) {
- lua_pushvalue (L, 1);
-
- lua_pushstring (L, "cmd");
- lua_gettable (L, -2);
- cmd = lua_tostring (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);
- }
- }
- }
- 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);
-
- lua_pushstring (L, "opaque_data");
- lua_gettable (L, -2);
- if (!!lua_toboolean (L, -1)) {
- flags |= LUA_REDIS_TEXTDATA;
- }
- lua_pop (L, 1);
-
-
- if (cmd) {
- lua_pushstring (L, "args");
- lua_gettable (L, -2);
- lua_redis_parse_args (L, -1, cmd, &args, &arglens, &nargs);
- lua_pop (L, 1);
- }
-
- lua_pop (L, 1);
-
- if (addr && cmd) {
- ret = TRUE;
- }
- }
-
- if (ret) {
- double_to_tv (timeout, &tv);
-
- if (rspamd_inet_address_get_af (addr->addr) == AF_UNIX) {
- ctx = redisConnectUnixWithTimeout (
- rspamd_inet_address_to_string (addr->addr), tv);
- }
- else {
- ctx = redisConnectWithTimeout (
- rspamd_inet_address_to_string (addr->addr),
- rspamd_inet_address_get_port (addr->addr), tv);
- }
-
- if (ip) {
- rspamd_inet_address_free (ip);
- }
-
- if (ctx == NULL || ctx->err) {
- redisFree (ctx);
- lua_redis_free_args (args, arglens, nargs);
- lua_pushboolean (L, FALSE);
-
- return 1;
- }
-
- r = redisCommandArgv (ctx,
- nargs,
- (const gchar **)args,
- arglens);
-
- if (r != NULL) {
- if (r->type != REDIS_REPLY_ERROR) {
- lua_pushboolean (L, TRUE);
- lua_redis_push_reply (L, r, flags & LUA_REDIS_TEXTDATA);
- }
- else {
- lua_pushboolean (L, FALSE);
- lua_pushstring (L, r->str);
- }
-
- freeReplyObject (r);
- redisFree (ctx);
- lua_redis_free_args (args, arglens, nargs);
-
- return 2;
- }
- else {
- msg_info ("call to redis failed: %s", ctx->errstr);
- redisFree (ctx);
- lua_redis_free_args (args, arglens, nargs);
- lua_pushboolean (L, FALSE);
- }
- }
- else {
- if (ip) {
- rspamd_inet_address_free (ip);
- }
- msg_err ("bad arguments for redis request");
- lua_pushboolean (L, FALSE);
- }
-
- 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 {boolean,redis} new connection object or nil if connection failed
- */
- static int
- lua_redis_connect (lua_State *L)
- {
- LUA_TRACE_POINT;
- struct lua_redis_userdata *ud;
- struct lua_redis_ctx *ctx, **pctx;
- gdouble timeout = REDIS_DEFAULT_TIMEOUT;
-
- ctx = rspamd_lua_redis_prepare_connection (L, NULL, TRUE);
-
- if (ctx) {
- ud = &ctx->async;
-
- lua_pushstring (L, "timeout");
- lua_gettable (L, 1);
- if (lua_type (L, -1) == LUA_TNUMBER) {
- timeout = lua_tonumber (L, -1);
- }
-
- lua_pop (L, 1);
- ud->timeout = timeout;
- }
- else {
- lua_pushboolean (L, FALSE);
- lua_pushnil (L);
-
- return 2;
- }
-
- lua_pushboolean (L, TRUE);
- pctx = lua_newuserdata (L, sizeof (ctx));
- *pctx = ctx;
- rspamd_lua_setclass (L, "rspamd{redis}", -1);
-
- return 2;
- }
-
- /***
- * @function rspamd_redis.connect_sync({params})
- * Make blocking request to redis server, params is a table of key=value arguments in any order
- * @param {ip|string} host server address
- * @param {number} timeout timeout in seconds for request (1.0 by default)
- * @return {redis} redis object if a request has been successful
- */
- static int
- lua_redis_connect_sync (lua_State *L)
- {
- LUA_TRACE_POINT;
- gdouble timeout = REDIS_DEFAULT_TIMEOUT;
- struct lua_redis_ctx *ctx, **pctx;
-
- ctx = rspamd_lua_redis_prepare_connection (L, NULL, FALSE);
-
- if (ctx) {
- if (lua_istable (L, 1)) {
- lua_pushstring (L, "timeout");
- lua_gettable (L, 1);
- if (lua_type (L, -1) == LUA_TNUMBER) {
- timeout = lua_tonumber (L, -1);
- }
- lua_pop (L, 1);
- }
-
- ctx->async.timeout = timeout;
-
- lua_pushboolean (L, TRUE);
- pctx = lua_newuserdata (L, sizeof (ctx));
- *pctx = ctx;
- rspamd_lua_setclass (L, "rspamd{redis}", -1);
- }
- else {
- lua_pushboolean (L, FALSE);
- lua_pushstring (L, "bad arguments for redis request");
- return 2;
- }
-
- return 2;
- }
-
- /***
- * @method rspamd_redis:add_cmd(cmd, {args})
- * Append new cmd to redis pipeline
- * @param {string} cmd command to be sent to redis
- * @param {table} args array of strings used as redis arguments
- * @return {boolean} `true` if a request has been successful
- */
- static int
- lua_redis_add_cmd (lua_State *L)
- {
- LUA_TRACE_POINT;
- struct lua_redis_ctx *ctx = lua_check_redis (L, 1);
- struct lua_redis_request_specific_userdata *sp_ud;
- struct lua_redis_userdata *ud;
- const gchar *cmd = NULL;
- gint args_pos = 2;
- gint cbref = -1, ret;
-
- if (ctx) {
- if (ctx->flags & LUA_REDIS_TERMINATED) {
- lua_pushboolean (L, FALSE);
- lua_pushstring (L, "Connection is terminated");
-
- return 2;
- }
-
- /* Async version */
- if (lua_type (L, 2) == LUA_TSTRING) {
- /* No callback version */
- cmd = lua_tostring (L, 2);
- args_pos = 3;
- }
- else if (lua_type (L, 2) == LUA_TFUNCTION) {
- lua_pushvalue (L, 2);
- cbref = luaL_ref (L, LUA_REGISTRYINDEX);
- cmd = lua_tostring (L, 3);
- args_pos = 4;
- }
- else {
- return luaL_error (L, "invalid arguments");
- }
-
- sp_ud = g_malloc0 (sizeof (*sp_ud));
- if (IS_ASYNC (ctx)) {
- sp_ud->c = &ctx->async;
- ud = &ctx->async;
- sp_ud->cbref = cbref;
- }
- else {
- sp_ud->c = &ctx->async;
- ud = &ctx->async;
- }
- sp_ud->ctx = ctx;
-
- lua_redis_parse_args (L, args_pos, cmd, &sp_ud->args,
- &sp_ud->arglens, &sp_ud->nargs);
-
- LL_PREPEND (sp_ud->c->specific, sp_ud);
-
- if (ud->s && rspamd_session_blocked (ud->s)) {
- lua_pushboolean (L, 0);
- lua_pushstring (L, "session is terminating");
-
- return 2;
- }
-
- if (IS_ASYNC (ctx)) {
- ret = redisAsyncCommandArgv (sp_ud->c->ctx,
- lua_redis_callback,
- sp_ud,
- sp_ud->nargs,
- (const gchar **)sp_ud->args,
- sp_ud->arglens);
- }
- else {
- ret = redisAsyncCommandArgv (sp_ud->c->ctx,
- lua_redis_callback_sync,
- sp_ud,
- sp_ud->nargs,
- (const gchar **)sp_ud->args,
- sp_ud->arglens);
- }
-
- if (ret == REDIS_OK) {
- if (ud->s) {
- rspamd_session_add_event (ud->s,
- lua_redis_fin,
- sp_ud,
- M);
-
- if (ud->item) {
- rspamd_symcache_item_async_inc (ud->task, ud->item, M);
- }
- }
-
- sp_ud->timeout_ev.data = sp_ud;
-
- if (IS_ASYNC (ctx)) {
- ev_timer_init (&sp_ud->timeout_ev, lua_redis_timeout,
- sp_ud->c->timeout, 0.0);
- }
- else {
- ev_timer_init (&sp_ud->timeout_ev, lua_redis_timeout_sync,
- sp_ud->c->timeout, 0.0);
- }
-
- ev_timer_start (ud->event_loop, &sp_ud->timeout_ev);
- REDIS_RETAIN (ctx);
- ctx->cmds_pending ++;
- }
- else {
- msg_info ("call to redis failed: %s",
- sp_ud->c->ctx->errstr);
- lua_pushboolean (L, 0);
- lua_pushstring (L, sp_ud->c->ctx->errstr);
-
- return 2;
- }
- }
-
- lua_pushboolean (L, true);
-
- return 1;
- }
-
- /***
- * @method rspamd_redis:exec()
- * Executes pending commands (suitable for blocking IO only for now)
- * @return {boolean}, {table}, ...: pairs in format [bool, result] for each request pending
- */
- static int
- lua_redis_exec (lua_State *L)
- {
- LUA_TRACE_POINT;
- struct lua_redis_ctx *ctx = lua_check_redis (L, 1);
-
- if (ctx == NULL) {
- lua_error (L);
-
- return 1;
- }
-
- if (IS_ASYNC (ctx)) {
- lua_pushstring (L, "Async redis pipelining is not implemented");
- lua_error (L);
- return 0;
- }
- else {
- if (false /* !ctx->d.sync */) {
- lua_pushstring (L, "cannot exec commands when not connected");
- lua_error (L);
- return 0;
- }
- else {
- if (ctx->cmds_pending == 0 && g_queue_get_length (ctx->replies) == 0) {
- lua_pushstring (L, "No pending commands to execute");
- lua_error (L);
- }
- if (ctx->cmds_pending == 0 && g_queue_get_length (ctx->replies) > 0) {
- gint results = lua_redis_push_results (ctx, L);
- return results;
- }
- else {
- ctx->thread = lua_thread_pool_get_running_entry (ctx->async.cfg->lua_thread_pool);
- return lua_thread_yield (ctx->thread, 0);
- }
- }
- }
- }
- #else
- static int
- lua_redis_make_request (lua_State *L)
- {
- msg_warn ("rspamd is compiled with no redis support");
-
- lua_pushboolean (L, FALSE);
-
- return 1;
- }
- static int
- lua_redis_make_request_sync (lua_State *L)
- {
- msg_warn ("rspamd is compiled with no redis support");
-
- lua_pushboolean (L, FALSE);
-
- 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;
- }
- static int
- lua_redis_gc (lua_State *L)
- {
- return 0;
- }
- #endif
-
- static gint
- lua_load_redis (lua_State * L)
- {
- lua_newtable (L);
- luaL_register (L, NULL, redislib_f);
-
- return 1;
- }
- /**
- * Open redis library
- * @param L lua stack
- * @return
- */
- 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);
- }
|