Merge branch 'release/0.08'

README.md

@@ -19,7 +19,7 @@ This module should be considered experimental.
 Requirements
 ============
 
-* Redis >= 2.8.x
+* Redis >= 2.8.x < 3.2.x (newer Redis versions require a patch to qless-core)
 * OpenResty >= 1.9.x
 * [lua-resty-redis-connector](https://github.com/pintsized/lua-resty-redis-connector) >= 0.03
 
@@ -100,8 +100,16 @@ local _M = {}
 
 function _M.perform(job)
     -- job is an instance of Qless_Job and provides access to
-    -- job.data (which is a Lua table), a means to cancel the 
+    -- job.data (which is a Lua table), a means to cancel the
     -- job (job:cancel()), and more.
+
+    -- return "nil, err_type, err_msg" to indicate an unexpected failure
+
+    if not job.data then
+        return nil, "job-error", "data missing"
+    end
+
+    -- Do work
 end
 
 return _M

dist.ini

@@ -0,0 +1,10 @@
+name=lua-resty-qless
+abstract=Lua binding to Qless (Queue / Pipeline management) for OpenResty
+author=James Hurst
+is_original=yes
+license=2bsd
+lib_dir=lib
+doc_dir=lib
+repo_link=https://github.com/pintsized/lua-resty-qless
+main_module=lib/resty/qless.lua
+requires=luajit, openresty/lua-resty-redis, pintsized/lua-resty-redis-connector >= 0.03

lib/resty/qless.lua

@@ -238,7 +238,7 @@ end
 
 
 local _M = {
-    _VERSION = '0.07',
+    _VERSION = '0.08',
 }
 
 local mt = { __index = _M }

lib/resty/qless/job.lua

@@ -10,7 +10,7 @@ local cjson_decode = cjson.decode
 
 
 local _M = {
-    _VERSION = '0.01',
+    _VERSION = '0.08',
 }
 
 local mt = {
@@ -162,7 +162,7 @@ function _M.fail(self, group, message)
     local res, err = self.client:call("fail",
         self.jid,
         self.client.worker_name,
-        group, message or "[no message]",
+        group or "[unknown group]", message or "[no message]",
         cjson_encode(self.data))
 
     if not res then

lib/resty/qless/luascript.lua

@@ -12,7 +12,7 @@ local io_open = io.open
 
 
 local _M = {
-    _VERSION = '0.01',
+    _VERSION = '0.08',
 }
 
 local mt = { __index = _M }

lib/resty/qless/queue.lua

@@ -48,7 +48,7 @@ end
 
 
 local _M = {
-    _VERSION = '0.01',
+    _VERSION = '0.08',
 }
 
 

lib/resty/qless/recurring_job.lua

@@ -10,7 +10,7 @@ local cjson_decode = cjson.decode
 
 
 local _M = {
-    _VERSION = '0.01',
+    _VERSION = '0.08',
 }
 
 local mt = {

lib/resty/qless/reserver/ordered.lua

@@ -4,7 +4,7 @@ local ngx_ERR = ngx.ERR
 local ngx_INFO = ngx.INFO
 
 local _M = {
-    _VERSION = '0.01',
+    _VERSION = '0.08',
 }
 
 local mt = { __index = _M }

lib/resty/qless/reserver/round_robin.lua

@@ -4,7 +4,7 @@ local ngx_ERR = ngx.ERR
 local ngx_INFO = ngx.INFO
 
 local _M = {
-    _VERSION = '0.01',
+    _VERSION = '0.08',
 }
 
 local mt = { __index = _M }

lib/resty/qless/reserver/shuffled_round_robin.lua

@@ -5,7 +5,7 @@ local math_random = math.random
 local math_randomseed = math.randomseed
 
 local _M = {
-    _VERSION = '0.01',
+    _VERSION = '0.08',
 }
 
 local mt = { __index = _M }

lib/resty/qless/worker.lua

@@ -17,7 +17,7 @@ local co_yield = coroutine.yield
 
 
 local _M = {
-    _VERSION = '0.01',
+    _VERSION = '0.08',
 }
 
 local mt = { __index = _M }
@@ -73,11 +73,16 @@ function _M.start(self, options)
             repeat
                 local job = reserver:reserve()
                 if job then
-                    local res, err_type, err = self:perform(job)
-                    if res == true then
-                        job:complete()
-                    else
+                    local ok, err_type, err = self:perform(job)
+                    if not ok and err_type then
+                        -- err_type, err indicates the job "raised an exception"
                         job:fail(err_type, err)
+                        ngx_log(ngx_ERR, "Got ", err_type, " failure from ", job:description(), " \n", err)
+                    else
+                        -- Complete the job, unless its status has been changed already
+                        if not job.state_changed then
+                            job:complete()
+                        end
                     end
                 end
                 co_yield() -- The scheduler will resume us.

lua-resty-qless-0.07-0.rockspec → lua-resty-qless-0.08-0.rockspec

@@ -1,8 +1,8 @@
 package = "lua-resty-qless"
-version = "0.07-0"
+version = "0.08-0"
 source  = {
   url = "git://github.com/pintsized/lua-resty-qless",
-  tag = "v0.07"
+  tag = "v0.08"
 }
 description = {
   summary    = "Lua binding to Qless (Queue / Pipeline management) for OpenResty",

t/05-worker.t

@@ -1,9 +1,7 @@
-# vim:set ft= ts=4 sw=4 et:
-
 use Test::Nginx::Socket;
 use Cwd qw(cwd);
 
-plan tests => repeat_each() * (blocks() * 3) + 1;
+plan tests => repeat_each() * (blocks() * 3) + 2;
 
 my $pwd = cwd();
 
@@ -27,18 +25,26 @@ our $HttpConfig = qq{
 
         function sum.perform(job)
             local data = job.data
-            if not data or #data == 0 then
+
+            if data.cancel then
                 job:cancel()
-                return nil
+                return
+            end
+
+            if not data or not data.numbers or #data.numbers == 0 then
+                return nil, "job-error", "no data provided"
             end
 
             local sum = 0
-            for _,v in ipairs(data) do
+            for _,v in ipairs(data.numbers) do
                 sum = sum + v
             end
 
             ngx.log(ngx.NOTICE, "Sum: ", sum)
-            return true
+
+            if data.autocomplete then
+                job:complete()
+            end
         end
 
         package.loaded["testtasks.sum"] = sum
@@ -55,7 +61,7 @@ our $HttpConfig = qq{
             concurrency = 4,
             reserver = "ordered",
             queues = { "queue_14" },
-        }) 
+        })
 
 
         local worker_mw = Qless_Worker.new(redis_params)
@@ -86,7 +92,7 @@ __DATA__
             local qless = require "resty.qless"
             local q = qless.new(redis_params)
 
-            local jid = q.queues["queue_14"]:put("testtasks.sum", { 1, 2, 3, 4 })
+            local jid = q.queues["queue_14"]:put("testtasks.sum", { numbers = { 1, 2, 3, 4 } })
             ngx.sleep(1)
 
             local job = q.jobs:get(jid)
@@ -109,7 +115,7 @@ complete
             local qless = require "resty.qless"
             local q = qless.new(redis_params)
 
-            local jid = q.queues["queue_15"]:put("testtasks.sum", { 1, 2, 3, 4 })
+            local jid = q.queues["queue_15"]:put("testtasks.sum", { numbers = { 1, 2, 3, 4 } })
             ngx.sleep(1)
 
             local job = q.jobs:get(jid)
@@ -126,15 +132,17 @@ qr/Middleware stop/,
 qr/Middleware start/]
 
 
-=== TEST 3: Test a job can cancel itself if data is bad
+=== TEST 3: Test a job can cancel itself
 --- http_config eval: $::HttpConfig
 --- config
     location = /1 {
         content_by_lua '
             local qless = require "resty.qless"
             local q = qless.new(redis_params)
 
-            local jid = q.queues["queue_14"]:put("testtasks.sum")
+            local jid = q.queues["queue_14"]:put("testtasks.sum", {
+                cancel = true
+            })
             ngx.sleep(1)
 
             local job = q.jobs:get(jid)
@@ -149,3 +157,60 @@ qr/Middleware start/]
 GET /1
 --- response_body
 canceled
+--- no_error_log
+[error]
+
+
+=== TEST 3b: Test a job is failed and logs the error if data is bad
+--- http_config eval: $::HttpConfig
+--- config
+    location = /1 {
+        content_by_lua '
+            local qless = require "resty.qless"
+            local q = qless.new(redis_params)
+
+            local jid = q.queues["queue_14"]:put("testtasks.sum")
+            ngx.sleep(1)
+
+            local job = q.jobs:get(jid)
+            if job then
+                ngx.say(job.state)
+            else
+                ngx.say("canceled")
+            end
+        ';
+    }
+--- request
+GET /1
+--- response_body
+failed
+--- error_log eval
+[qr/Got job-error failure from testtasks\.sum \([a-f0-9]{32} \/ queue_14 \/ running\)/]
+
+
+=== TEST 4: Test a job can complete itself without tripping up the worker
+--- http_config eval: $::HttpConfig
+--- config
+    location = /1 {
+        content_by_lua '
+            local qless = require "resty.qless"
+            local q = qless.new(redis_params)
+
+            local jid = q.queues["queue_14"]:put("testtasks.sum", {
+                numbers = { 1, 2, 3, 4},
+                autocomplete = true
+            })
+            ngx.sleep(1)
+
+            local job = q.jobs:get(jid)
+            if job then
+                ngx.say(job.state)
+            end
+        ';
+    }
+--- request
+GET /1
+--- response_body
+complete
+--- no_error_log
+[error]