Iterate_with and process_while

For more flexible functionality, added the ability
to create a custom iterator that will be created at
the selected index (iterate_with). You can also pass a
predicate that will stop the fullscan process,
if required(process_while).

Needed for: #50

Author: ArtDu

README.md

@@ -50,6 +50,10 @@ Run a scheduled task to check and process (expire) tuples in a given space.
      The index value may be a single value, if the index consists of one field, a tuple with the index key parts, or a function which returns such value.
      If omitted or nil, all tuples will be checked.
     * `tuples_per_iteration` - Number of tuples to check in one batch (iteration). Default is 1024.
+    * `process_while` - Function to call before checking each tuple.
+     If it returns false, the current tuple scan task finishes.
+    * `iterate_with` - Function which returns an iterator object which provides tuples to check, considering the start_key, process_while and other options.
+     There's a default function which can be overriden with this parameter.
     * `on_full_scan_start` - Function to call before starting a tuple scan.
     * `on_full_scan_complete` - Function to call after completing a full scan.
     * `on_full_scan_success` - Function to call after successfully completing a full scan.

expirationd.lua

@@ -48,6 +48,8 @@ local constants = {
     default_on_full_scan = function() end,
     -- default function for start_key
     start_key = function() return nil end,
+    -- default function for process_while
+    process_while = function() return true end,
     -- default iterating over the loop will go in ascending index
     iterator_type = "ALL",
 }
@@ -97,7 +99,7 @@ local function default_do_worker_iteration(task)
     local space_len = task.vinyl_assumed_space_len
     local checked_tuples_count = 0
     local vinyl_checked_tuples_count = 0
-    for _, tuple in task.index:pairs(task.start_key(), {iterator = task.iterator_type}) do
+    for _, tuple in task:iterate_with() do
         checked_tuples_count = checked_tuples_count + 1
         vinyl_checked_tuples_count = vinyl_checked_tuples_count + 1
         expiration_process(task, tuple)
@@ -226,6 +228,7 @@ local function create_task(name)
         process_expired_tuple = nil,
         args                  = nil,
         index                 = nil,
+        iterate_with          = nil,
         iteration_delay                = constants.max_delay,
         full_scan_delay                = constants.max_delay,
         tuples_per_iteration           = constants.default_tuples_per_iteration,
@@ -237,6 +240,7 @@ local function create_task(name)
         on_full_scan_start             = constants.default_on_full_scan,
         on_full_scan_complete          = constants.default_on_full_scan,
         start_key                      = constants.start_key,
+        process_while                  = constants.process_while,
         iterator_type                  = constants.iterator_type,
     }, { __index = Task_methods })
     return task
@@ -262,6 +266,16 @@ local function default_tuple_drop(space_id, args, tuple)
 end
 
 
+-- default iterate_with function
+local function default_iterate_with(task)
+    return task.index:pairs(task.start_key(), { iterator = task.iterator_type })
+       :take_while(
+            function()
+                return task:process_while()
+            end
+        )
+end
+
 -- ========================================================================= --
 -- Expiration daemon management functions
 -- ========================================================================= --
@@ -289,6 +303,11 @@ end
 --                                or a function which returns such value;
 --                                if omitted or nil, all tuples will be checked.
 --     * tuples_per_iteration  -- Number of tuples to check in one batch (iteration); default is 1024.
+--     * process_while         -- Function to call before checking each tuple;
+--                                if it returns false, the task will stop until next full scan.
+--     * iterate_with          -- Function which returns an iterator object which provides tuples to check,
+--                                considering the start_key, process_while and other options.
+--                                There's a default function which can be overriden with this parameter.
 --     * on_full_scan_start    -- Function to call before starting a full scan iteration.
 --     * on_full_scan_complete -- Function to call after completing a full scan iteration.
 --     * on_full_scan_success  -- Function to call after successfully completing a full scan iteration.
@@ -375,6 +394,22 @@ local function expirationd_run_task(name, space_id, is_tuple_expired, options)
     -- check valid of iterator_type and start key
     task.index:pairs( task.start_key(), { iterator = task.iterator_type })
 
+    -- check process_while
+    if options.process_while ~= nil then
+        if type(options.process_while) ~= "function" then
+            error("Invalid type of process_while, expected function")
+        end
+        task.process_while = options.process_while
+    end
+
+    -- check iterate_with
+    if options.iterate_with ~= nil then
+        if type(options.iterate_with) ~= "function" then
+            error("Invalid type of iterate_with, expected function")
+        end
+    end
+    task.iterate_with = options.iterate_with or default_iterate_with
+
     -- check expire and process after expiration handler's arguments
     task.args = options.args
 

test/unit/iterate_with_test.lua

@@ -0,0 +1,26 @@
+local expirationd = require("expirationd")
+local t = require("luatest")
+local g = t.group("iterate_with")
+
+local helpers = require("test.helper")
+
+g.before_all(function()
+    helpers.init_spaces(g)
+end)
+
+g.after_each(function()
+    helpers.truncate_spaces(g)
+end)
+
+function g.test_passing()
+    local task = expirationd.start("clean_all", g.tree.id, helpers.is_expired_true,
+            { iterate_with = helpers.is_expired_true })
+    -- default process_while always return false, iterations never stopped by this function
+    t.assert_equals(task.iterate_with, helpers.is_expired_true)
+    task:kill()
+
+    -- errors
+    t.assert_error_msg_content_equals("Invalid type of iterate_with, expected function",
+            expirationd.start, "clean_all", g.tree.id, helpers.is_expired_true,
+            { iterate_with = "" })
+end

test/unit/process_while_test.lua

@@ -0,0 +1,70 @@
+local expirationd = require("expirationd")
+local t = require("luatest")
+local g = t.group("process_while")
+
+local helpers = require("test.helper")
+
+g.before_all(function()
+    helpers.init_spaces(g)
+end)
+
+g.after_each(function()
+    helpers.truncate_spaces(g)
+end)
+
+function g.test_passing()
+    local task = expirationd.start("clean_all", g.tree.id, helpers.is_expired_true)
+    -- default process_while always return false, iterations never stopped by this function
+    t.assert_equals(task.process_while(), true)
+    task:kill()
+
+    local function process_while()
+        return false
+    end
+
+    task = expirationd.start("clean_all", g.tree.id, helpers.is_expired_true,
+            {process_while = process_while})
+    t.assert_equals(task.process_while(), false)
+    task:kill()
+
+    -- errors
+    t.assert_error_msg_content_equals("Invalid type of process_while, expected function",
+            expirationd.start, "clean_all", g.tree.id, helpers.is_expired_true,
+            { process_while = "" })
+end
+
+local function process_while(task)
+    if task.checked_tuples_count >= 1 then return false end
+    return true
+end
+
+function g.test_tree_index()
+    for _, space in pairs({g.tree, g.vinyl}) do
+        helpers.iteration_result = {}
+        space:insert({1, "3"})
+        space:insert({2, "2"})
+        space:insert({3, "1"})
+        local task = expirationd.start("clean_all", space.id, helpers.is_expired_debug,
+                {process_while = process_while})
+        -- wait for tuples expired
+        helpers.retrying({}, function()
+            t.assert_equals(helpers.iteration_result, {{1, "3"}})
+        end)
+        task:kill()
+    end
+end
+
+function g.test_hash_index()
+    helpers.iteration_result = {}
+    g.hash:insert({1, "3"})
+    g.hash:insert({2, "2"})
+    g.hash:insert({3, "1"})
+
+    local task = expirationd.start("clean_all", g.hash.id, helpers.is_expired_debug,
+            {process_while = process_while})
+    -- wait for tuples expired
+    helpers.retrying({}, function()
+        t.assert_equals(helpers.iteration_result, {{3, "1"}})
+    end)
+    task:kill()
+end