diff options
author | Kim Alvefur <zash@zash.se> | 2021-12-14 19:58:53 +0100 |
---|---|---|
committer | Kim Alvefur <zash@zash.se> | 2021-12-14 19:58:53 +0100 |
commit | 93a9ed262065d9966643ad5ae50b58142a1c6823 (patch) | |
tree | 0fbcc8578d3aefac8a5021d36a86c5ae2aa78950 /util | |
parent | bdce99b7dd4411a8717798874a320bd2fc191b57 (diff) | |
download | prosody-93a9ed262065d9966643ad5ae50b58142a1c6823.tar.gz prosody-93a9ed262065d9966643ad5ae50b58142a1c6823.zip |
util.smqueue: Abstract queue with acknowledgements and overflow
Meant to be used in mod_smacks for XEP-0198
Meant to have a larger virtual size than actual number of items stored,
on the theory that in most cases, the excess will be acked before needed
for a resumption event.
Diffstat (limited to 'util')
-rw-r--r-- | util/smqueue.lua | 61 |
1 files changed, 61 insertions, 0 deletions
diff --git a/util/smqueue.lua b/util/smqueue.lua new file mode 100644 index 00000000..d24c35ae --- /dev/null +++ b/util/smqueue.lua @@ -0,0 +1,61 @@ +local queue = require("util.queue"); + +local lib = { smqueue = {} } + +local smqueue = lib.smqueue; + +function smqueue:push(v) + self._head = self._head + 1; + + assert(self._queue:push(v)); +end + +function smqueue:ack(h) + if h < self._tail then + return nil, "tail" + elseif h > self._head then + return nil, "head" + end + + local acked = {}; + self._tail = h; + local expect = self._head - self._tail; + while expect < self._queue:count() do + local v = self._queue:pop(); + if not v then return nil, "pop" end + table.insert(acked, v); + end + return acked +end + +function smqueue:count_unacked() return self._head - self._tail end + +function smqueue:count_acked() return self._tail end + +function smqueue:resumable() return self._queue:count() >= (self._head - self._tail) end + +function smqueue:resume() return self._queue:items() end + +function smqueue:consume() return self._queue:consume() end + +local compat_mt = {} + +function compat_mt:__index(i) + if i < self._queue._tail then return nil end + return self._queue._queue._items[(i + self._queue._tail) % self._queue._queue.size] +end + +function compat_mt:__len() return self._queue:count_unacked() end + +function smqueue:table() return setmetatable({ _queue = self }, compat_mt) end + +local function freeze(q) return { head = q._head; tail = q._tail } end + +local queue_mt = { __name = "smqueue"; __index = smqueue; __len = smqueue.count_unacked; __freeze = freeze } + +function lib.new(size) + assert(size > 0); + return setmetatable({ _head = 0; _tail = 0; _queue = queue.new(size, true) }, queue_mt) +end + +return lib |