Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

batch processor name and error return type fix #1927

Merged
merged 2 commits into from
Jul 29, 2020
Merged

batch processor name and error return type fix #1927

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
418ed10
batch processor name and error return fix
sshniro Jul 29, 2020
228d8c2
Adding EOL
sshniro Jul 29, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 10 additions & 10 deletions apisix/utils/batch-processor.lua
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,9 +21,9 @@ local ipairs = ipairs
local table = table
local now = ngx.now
local type = type
local Batch_Processor = {}
local Batch_Processor_mt = {
__index = Batch_Processor
local batch_processor = {}
local batch_processor_mt = {
__index = batch_processor
}
local execute_func
local create_buffer_timer
Expand Down Expand Up @@ -109,17 +109,17 @@ function create_buffer_timer(self)
end


function Batch_Processor:new(func, config)
function batch_processor:new(func, config)
local ok, err = core.schema.check(schema, config)
if not ok then
return err
return nil, err
end

if not(type(func) == "function") then
return nil, "Invalid argument, arg #1 must be a function"
end

local batch_processor = {
local processor = {
func = func,
buffer_duration = config.buffer_duration,
inactive_timeout = config.inactive_timeout,
Expand All @@ -134,11 +134,11 @@ function Batch_Processor:new(func, config)
last_entry_t = 0
}

return setmetatable(batch_processor, Batch_Processor_mt)
return setmetatable(processor, batch_processor_mt)
end


function Batch_Processor:push(entry)
function batch_processor:push(entry)
-- if the batch size is one then immediately send for processing
if self.batch_max_size == 1 then
local batch = { entries = { entry }, retry_count = 0 }
Expand Down Expand Up @@ -166,7 +166,7 @@ function Batch_Processor:push(entry)
end


function Batch_Processor:process_buffer()
function batch_processor:process_buffer()
-- If entries are present in the buffer move the entries to processing
if #self.entry_buffer.entries > 0 then
core.log.debug("tranferring buffer entries to processing pipe line, ",
Expand All @@ -182,4 +182,4 @@ function Batch_Processor:process_buffer()
end


return Batch_Processor
return batch_processor