[Minor] Move upstream handling into rspamd_redis_make_request

8 files changed, 20 insertions, 50 deletions

diff --git a/src/lua/global_functions.lua b/src/lua/global_functions.lua
index 879309011..31ee6c8c4 100644
--- a/src/lua/global_functions.lua
+++ b/src/lua/global_functions.lua
@@ -111,11 +111,19 @@ end
 -- command - redis command
 -- args - table of arguments
 function rspamd_redis_make_request(task, redis_params, key, is_write, callback, command, args)
+  local addr
+  local function rspamd_redis_make_request_cb(err, data)
+    if err then
+      addr:fail()
+    else
+      addr:ok()
+    end
+    callback(err, data, addr)
+  end
   if not task or not redis_params or not callback or not command then
     return false,nil,nil
   end
 
-  local addr
   local rspamd_redis = require "rspamd_redis"
 
   if key then
@@ -138,7 +146,7 @@ function rspamd_redis_make_request(task, redis_params, key, is_write, callback,
 
   local options = {
     task = task,
-    callback = callback,
+    callback = rspamd_redis_make_request_cb,
     host = addr:get_addr(),
     timeout = redis_params['timeout'],
     cmd = command,
diff --git a/src/plugins/lua/antivirus.lua b/src/plugins/lua/antivirus.lua
index 91ac61af6..0ae2170b6 100644
--- a/src/plugins/lua/antivirus.lua
+++ b/src/plugins/lua/antivirus.lua
@@ -274,10 +274,8 @@ local function fprot_check(task, rule)
           end
         else
           rspamd_logger.errx(task, 'failed to scan: %s', err)
-          upstream:fail()
         end
       else
-        upstream:ok()
 
         data = tostring(data)
         local found = (string.sub(data, 1, 1) == '1')
@@ -340,10 +338,8 @@ local function clamav_check(task, rule)
           end
         else
           rspamd_logger.errx(task, 'failed to scan: %s', err)
-          upstream:fail()
         end
       else
-        upstream:ok()
 
         data = tostring(data)
         local s = string.find(data, ' FOUND')
@@ -405,10 +401,8 @@ local function sophos_check(task, rule)
           end
         else
           rspamd_logger.errx(task, 'failed to scan: %s', err)
-          upstream:fail()
         end
       else
-        upstream:ok()
 
         data = tostring(data)
         local vname = string.match(data, 'VIRUS (%S+) ')
diff --git a/src/plugins/lua/asn.lua b/src/plugins/lua/asn.lua
index dccaf43d7..9cb109fbf 100644
--- a/src/plugins/lua/asn.lua
+++ b/src/plugins/lua/asn.lua
@@ -68,12 +68,9 @@ local function asn_check(task)
         local redis_key = options.key_prefix .. ip:to_string()
         local ret,conn,upstream
         local function redis_asn_set_cb(redis_err)
-          if not redis_err then
-            upstream:ok()
-          else
+          if redis_err then
             rspamd_logger.errx(task, 'got error %s when setting asn record on server %s',
               redis_err, upstream:get_addr())
-            upstream:fail()
           end
         end
         ret,conn,upstream = rspamd_redis_make_request(task,
diff --git a/src/plugins/lua/fann_classifier.lua b/src/plugins/lua/fann_classifier.lua
index ba41040e9..94261839b 100644
--- a/src/plugins/lua/fann_classifier.lua
+++ b/src/plugins/lua/fann_classifier.lua
@@ -98,9 +98,6 @@ local function maybe_load_fann(task, continue_cb, call_if_fail)
     local function redis_fann_check_cb(err, data)
       if err then
         rspamd_logger.errx(task, 'redis error on host %s: %s', upstream:get_addr(), err)
-        upstream:fail()
-      else
-        upstream:ok()
       end
       if not err and type(data) == 'string' then
         local version = tonumber(data)
@@ -187,13 +184,9 @@ local function create_fann()
 end
 
 local function save_fann(task, is_spam)
-  local ret, conn, upstream
   local function redis_fann_save_cb(err)
     if err then
       rspamd_logger.errx(task, "cannot save neural net to redis: %s", err)
-      upstream:fail()
-    else
-      upstream:ok()
     end
   end
 
@@ -206,7 +199,7 @@ local function save_fann(task, is_spam)
   else
     current_classify_ann.ham_learned = current_classify_ann.ham_learned + 1
   end
-  ret,conn,upstream = rspamd_redis_make_request(task,
+  local ret,conn = rspamd_redis_make_request(task,
     redis_params, -- connect params
     key, -- hash key
     true, -- is write
diff --git a/src/plugins/lua/greylist.lua b/src/plugins/lua/greylist.lua
index 36819e1d0..7a84e46f3 100644
--- a/src/plugins/lua/greylist.lua
+++ b/src/plugins/lua/greylist.lua
@@ -161,7 +161,6 @@ local function greylist_check(task)
   local body_key = data_key(task)
   local meta_key = envelope_key(task)
   local hash_key = body_key .. meta_key
-  local upstream
 
   local function redis_get_cb(err, data)
     local ret_body = false
@@ -209,14 +208,11 @@ local function greylist_check(task)
       end
     elseif err then
       rspamd_logger.errx(task, 'got error while getting greylisting keys: %1', err)
-      upstream:fail()
       return
     end
-    upstream:ok()
   end
 
-  local ret, _
-  ret,_,upstream = rspamd_redis_make_request(task,
+  local ret = rspamd_redis_make_request(task,
     redis_params, -- connect params
     hash_key, -- hash key
     false, -- is write
@@ -269,12 +265,9 @@ local function greylist_set(task)
   local hash_key = body_key .. meta_key
 
   local function redis_set_cb(err)
-    if not err then
-      upstream:ok()
-    else
+    if err then
       rspamd_logger.errx(task, 'got error %s when setting greylisting record on server %s',
           err, upstream:get_addr())
-      upstream:fail()
     end
   end
 
diff --git a/src/plugins/lua/metadata_exporter.lua b/src/plugins/lua/metadata_exporter.lua
index d2c0fa620..e95478b48 100644
--- a/src/plugins/lua/metadata_exporter.lua
+++ b/src/plugins/lua/metadata_exporter.lua
@@ -69,9 +69,6 @@ local function metadata_exporter(task)
     if err then
       rspamd_logger.errx(task, 'got error %s when publishing record on server %s',
           err, upstream:get_addr())
-      upstream:fail()
-    else
-      upstream:ok()
     end
   end
   if settings.select then
diff --git a/src/plugins/lua/mx_check.lua b/src/plugins/lua/mx_check.lua
index f5a44bc4e..c81fd0dd6 100644
--- a/src/plugins/lua/mx_check.lua
+++ b/src/plugins/lua/mx_check.lua
@@ -55,7 +55,6 @@ local function mx_check(task)
   end
 
   local valid = false
-  local ret,_,upstream
 
   local function check_results(mxes)
     if fun.all(function(_, elt) return elt.checked end, mxes) then
@@ -64,10 +63,7 @@ local function mx_check(task)
       local function redis_cache_cb(err)
         if err ~= nil then
           rspamd_logger.errx(task, 'redis_cache_cb received error: %1', err)
-          upstream:fail()
           return
-        else
-          upstream:ok()
         end
       end
       if not valid then
@@ -81,7 +77,7 @@ local function mx_check(task)
         else
           task:insert_result(settings.symbol_bad_mx, 1.0)
         end
-        ret,_,upstream = rspamd_redis_make_request(task,
+        local ret = rspamd_redis_make_request(task,
           redis_params, -- connect params
           key, -- hash key
           false, -- is write
@@ -98,7 +94,7 @@ local function mx_check(task)
           table.insert(valid_mx, k)
         end, fun.filter(function (_, elt) return elt.working end, mxes))
         task:insert_result(settings.symbol_good_mx, 1.0, valid_mx)
-        ret,_,upstream = rspamd_redis_make_request(task,
+        local ret = rspamd_redis_make_request(task,
           redis_params, -- connect params
           key, -- hash key
           false, -- is write
@@ -228,7 +224,7 @@ local function mx_check(task)
     end
 
     local key = settings.key_prefix .. mx_domain
-    ret,_,upstream = rspamd_redis_make_request(task,
+    local ret = rspamd_redis_make_request(task,
       redis_params, -- connect params
       key, -- hash key
       false, -- is write
@@ -238,7 +234,6 @@ local function mx_check(task)
     )
 
     if not ret then
-      upstream:fail()
       local r = task:get_resolver()
       r:resolve('mx', {
         name = mx_domain,
diff --git a/src/plugins/lua/ratelimit.lua b/src/plugins/lua/ratelimit.lua
index f15d8633e..1a7bdf222 100644
--- a/src/plugins/lua/ratelimit.lua
+++ b/src/plugins/lua/ratelimit.lua
@@ -207,14 +207,11 @@ end
 local function check_limits(task, args)
 
   local key = fun.foldl(function(acc, k) return acc .. k[2] end, '', args)
-  local ret,upstream
+  local ret
   --- Called when value is got from server
   local function rate_get_cb(err, data)
     if err then
       rspamd_logger.infox(task, 'got error while getting limit: %1', err)
-      upstream:fail()
-    else
-      upstream:ok()
     end
     if not data then return end
     local ntime = rspamd_util.get_time()
@@ -298,8 +295,7 @@ local function check_limits(task, args)
       fun.map(function(a) return rspamd_str_split(a[2], ":")[2] end, args)))
   end
 
-  local _
-  ret,_,upstream = rspamd_redis_make_request(task,
+  ret = rspamd_redis_make_request(task,
     redis_params, -- connect params
     key, -- hash key
     false, -- is write
@@ -318,9 +314,7 @@ local function set_limits(task, args)
   local ret, upstream
 
   local function rate_set_cb(err)
-    if not err then
-      upstream:ok()
-    else
+    if err then
       rspamd_logger.infox(task, 'got error %s when setting ratelimit record on server %s',
         err, upstream:get_addr())
     end
@@ -328,7 +322,6 @@ local function set_limits(task, args)
   local function rate_get_cb(err, data)
     if err then
       rspamd_logger.infox(task, 'got error while setting limit: %1', err)
-      upstream:fail()
     end
     if not data then return end
     local ntime = rspamd_util.get_time()