Add TimerTask

CHANGELOG.md

@@ -21,6 +21,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 - ENV variables to make APIcast listen on HTTPS port [PR #622](https://github.com/3scale/apicast/pull/622) 
 - New `ssl_certificate` phase allows policies to provide certificate to terminate HTTPS connection [PR #622](https://github.com/3scale/apicast/pull/622).
 - Configurable `auth_type` for the token introspection policy [PR #755](https://github.com/3scale/apicast/pull/755)
+- `TimerTask` module to execute recurrent tasks that can be cancelled [PR #782](https://github.com/3scale/apicast/pull/782)
 
 ### Changed
 

gateway/Roverfile.lock

@@ -9,6 +9,7 @@ lua-resty-env 0.4.0-1||production
 lua-resty-execvp 0.1.1-1||production
 lua-resty-http 0.12-0||production
 lua-resty-iputils 0.3.0-1||development
+lua-resty-jit-uuid 0.0.7-1||production
 lua-resty-jwt 0.2.0-0||production
 lua-resty-repl 0.0.6-0||development
 lua-resty-url 0.2.0-1||production
@@ -24,4 +25,4 @@ net-url 0.9-1||testing
 nginx-lua-prometheus 0.20171117-4||production
 penlight 1.5.4-1||production,development,testing
 router 2.1-0||production
-say 1.3-1||testing
+say 1.3-1||testing

gateway/apicast-scm-1.rockspec

@@ -21,6 +21,7 @@ dependencies = {
    'argparse',
    'penlight',
    'nginx-lua-prometheus',
+   'lua-resty-jit-uuid',
 }
 build = {
    type = "make",

gateway/src/apicast/executor.lua

@@ -10,7 +10,7 @@ local PolicyChain = require('apicast.policy_chain')
 local Policy = require('apicast.policy')
 local linked_list = require('apicast.linked_list')
 local prometheus = require('apicast.prometheus')
-
+local uuid = require('resty.jit-uuid')
 
 local setmetatable = setmetatable
 local ipairs = ipairs
@@ -85,6 +85,10 @@ do
 
     local init_worker = _M.init_worker
     function _M:init_worker()
+        -- Need to seed the UUID in init_worker.
+        -- Ref: https://github.com/thibaultcha/lua-resty-jit-uuid/blob/c4c0004da0c4c4cdd23644a5472ea5c0d18decbb/README.md#usage
+        uuid.seed()
+
         local executed = {}
 
         for _,policy in init_worker(self) do

gateway/src/resty/concurrent/timer_task.lua

@@ -0,0 +1,94 @@
+local uuid = require 'resty.jit-uuid'
+local setmetatable = setmetatable
+local getmetatable = getmetatable
+local newproxy = newproxy
+local unpack = unpack
+
+local _M = {}
+
+local default_interval_seconds = 60
+
+_M.active_tasks = {}
+
+function _M.register_task(id)
+  _M.active_tasks[id] = true
+end
+
+function _M.unregister_task(id)
+  _M.active_tasks[id] = nil
+end
+
+function _M.task_is_active(id)
+  return _M.active_tasks[id] or false
+end
+
+local function generate_id()
+  return uuid.generate_v4()
+end
+
+local function mt(id)
+  -- Using 'newproxy' is needed to be able to define __gc().
+  -- With __gc we can make sure that the task will stop scheduling more work
+  -- after it has been garbage collected.
+  local proxy = newproxy(true)
+  local res_mt = getmetatable(proxy)
+  res_mt.__gc = function() _M.unregister_task(id) end
+  res_mt.__index = _M
+  return res_mt
+end
+
+--- Initialize a TimerTask.
+-- @tparam function task The function to be run periodically
+-- @tparam[opt] table opts
+-- @tfield ?table args Arguments to the function
+-- @tfield ?number interval Interval in seconds (defaults to 60)
+function _M.new(task, opts)
+  local options = opts or {}
+
+  local id = generate_id()
+
+  local self = setmetatable({}, mt(id))
+  self.task = task
+  self.args = options.args
+  self.interval = options.interval or default_interval_seconds
+  self.id = id
+
+  _M.register_task(id)
+
+  return self
+end
+
+local run_periodic, schedule_next, timer_execute
+
+run_periodic = function(self)
+  if not _M.task_is_active(self.id) then return end
+
+  self.task(unpack(self.args))
+
+  schedule_next(self)
+end
+
+-- Note: ngx.timer.at always sends "premature" as the first param.
+-- "premature" is boolean value indicating whether it is a premature timer
+-- expiration.
+timer_execute = function(_, self)
+  run_periodic(self)
+end
+
+schedule_next = function(self)
+  local ok, err = ngx.timer.at(self.interval, timer_execute, self)
+
+  if not ok then
+    ngx.log(ngx.ERR, "failed to schedule timer task: ", err)
+  end
+end
+
+function _M:execute()
+  run_periodic(self)
+end
+
+function _M:cancel()
+  _M.unregister_task(self.id)
+end
+
+return _M

spec/resty/concurrent/timer_task_spec.lua

@@ -0,0 +1,135 @@
+local TimerTask = require('resty.concurrent.timer_task')
+local ngx_timer = ngx.timer
+
+describe('TimerTask', function()
+  local test_task = function() end
+
+  before_each(function()
+    TimerTask.active_tasks = {}
+  end)
+
+  describe('.register_task', function()
+    it('adds an ID to the list of active tasks', function()
+      local id = '1'
+
+      TimerTask.register_task(id)
+
+      assert.is_true(TimerTask.task_is_active(id))
+    end)
+  end)
+
+  describe('.unregister_task', function()
+    local id = '1'
+
+    setup(function()
+      TimerTask.register_task(id)
+    end)
+
+    it('removes an ID to the list of active tasks', function()
+      TimerTask.unregister_task(id)
+      assert.is_false(TimerTask.task_is_active(id))
+    end)
+  end)
+
+  describe(':new', function()
+    it('adds the task to the list of active tasks', function()
+      local task = TimerTask.new(test_task)
+
+      assert.is_true(TimerTask.task_is_active(task.id))
+    end)
+
+    it('sets a default interval of 60s when not specified', function()
+      local task = TimerTask.new(test_task)
+
+      assert.equals(60, task.interval)
+    end)
+
+    it('allows to set a running interval', function()
+      local interval = 10
+
+      local task = TimerTask.new(test_task, { interval = interval })
+
+      assert.equals(interval, task.interval)
+    end)
+
+    it('allows to set arguments for the task', function()
+      local args = { '1', '2' }
+
+      local task = TimerTask.new(test_task, { args = args })
+
+      assert.same(args, task.args)
+    end)
+  end)
+
+  describe(':cancel', function()
+    it('removes the task from the list of active tasks', function()
+      local task = TimerTask.new(test_task)
+
+      task:cancel()
+
+      assert.is_false(TimerTask.task_is_active(task.id))
+    end)
+  end)
+
+  describe(':execute', function()
+    local func = test_task
+    local ngx_timer_stub
+
+    local args = { '1', '2', '3' }
+    local interval = 10
+
+    before_each(function()
+      ngx_timer_stub = stub(ngx_timer, 'at')
+    end)
+
+    describe('when the task is active', function()
+      it('runs the task', function()
+        local timer_task = TimerTask.new(func, { args = args, interval = interval })
+        local func_stub = stub(timer_task, 'task')
+
+        timer_task:execute()
+
+        assert.stub(func_stub).was_called_with(unpack(args))
+      end)
+
+      it('schedules the next one', function()
+        local timer_task = TimerTask.new(func, { args = args, interval = interval })
+
+        timer_task:execute()
+
+        assert.stub(ngx_timer_stub).was_called()
+      end)
+    end)
+
+    describe('when the task is not active', function()
+      it('does not run the task', function()
+        local timer_task = TimerTask.new(func, { args = args, interval = interval })
+        local func_stub = stub(timer_task, 'task')
+        timer_task:cancel()
+
+        timer_task:execute()
+
+        assert.stub(func_stub).was_not_called()
+      end)
+
+      it('does not schedule another task', function()
+        local timer_task = TimerTask.new(func, { args = args, interval = interval })
+        timer_task:cancel()
+
+        timer_task:execute()
+
+        assert.stub(ngx_timer_stub).was_not_called()
+      end)
+    end)
+  end)
+
+  it('cancels itself when it is garbage collected', function()
+    local timer_task = TimerTask.new(test_task)
+    local id = timer_task.id
+
+    timer_task = nil
+    collectgarbage()
+
+    assert.is_false(TimerTask.task_is_active(id))
+  end)
+end)