Mercurial > prosody-hg

annotate util/smqueue.lua @ 14008:21126415c8a0

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
util.smqueue: Fix syntax error in df970d2dac45 from bad merge (thanks Link Mauve)
author Matthew Wild <mwild1@gmail.com>
date Sun, 07 Dec 2025 10:17:32 +0000
parents df970d2dac45
children 1e01b91cf94d
Ignore whitespace changes - Everywhere: Within whitespace: At end of lines:
rev   line source
13994
8baaefc4db79 util.smqueue: Minor reformatting for code style, add some comments
Matthew Wild <mwild1@gmail.com>
parents: 12975
diff changeset
1 -- Represents an outgoing reliable stanza queue
8baaefc4db79 util.smqueue: Minor reformatting for code style, add some comments
Matthew Wild <mwild1@gmail.com>
parents: 12975
diff changeset
2
12975
d10957394a3c util: Prefix module imports with prosody namespace
Kim Alvefur <zash@zash.se>
parents: 12058
diff changeset
3 local queue = require("prosody.util.queue");
12055
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
4
13994
8baaefc4db79 util.smqueue: Minor reformatting for code style, add some comments
Matthew Wild <mwild1@gmail.com>
parents: 12975
diff changeset
5 -- SM queue methods
8baaefc4db79 util.smqueue: Minor reformatting for code style, add some comments
Matthew Wild <mwild1@gmail.com>
parents: 12975
diff changeset
6 local smqueue = {};
12055
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
7
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
8 function smqueue:push(v)
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
9 self._head = self._head + 1;
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
10
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
11 assert(self._queue:push(v));
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
12 end
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
13
13994
8baaefc4db79 util.smqueue: Minor reformatting for code style, add some comments
Matthew Wild <mwild1@gmail.com>
parents: 12975
diff changeset
14 -- Call when receiver has acknowledged some stanzas
8baaefc4db79 util.smqueue: Minor reformatting for code style, add some comments
Matthew Wild <mwild1@gmail.com>
parents: 12975
diff changeset
15 -- `h` (sent by the receiver) is the running count of
8baaefc4db79 util.smqueue: Minor reformatting for code style, add some comments
Matthew Wild <mwild1@gmail.com>
parents: 12975
diff changeset
16 -- successfully received stanzas
12055
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
17 function smqueue:ack(h)
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
18 if h < self._tail then
13994
8baaefc4db79 util.smqueue: Minor reformatting for code style, add some comments
Matthew Wild <mwild1@gmail.com>
parents: 12975
diff changeset
19 -- h is less than a previous h
12055
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
20 return nil, "tail"
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
21 elseif h > self._head then
13994
8baaefc4db79 util.smqueue: Minor reformatting for code style, add some comments
Matthew Wild <mwild1@gmail.com>
parents: 12975
diff changeset
22 -- h is greater than the number of stanzas we sent
12055
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
23 return nil, "head"
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
24 end
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
25
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
26 local acked = {};
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
27 self._tail = h;
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
28 local expect = self._head - self._tail;
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
29 while expect < self._queue:count() do
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
30 local v = self._queue:pop();
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
31 if not v then return nil, "pop" end
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
32 table.insert(acked, v);
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
33 end
13994
8baaefc4db79 util.smqueue: Minor reformatting for code style, add some comments
Matthew Wild <mwild1@gmail.com>
parents: 12975
diff changeset
34
14001
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
35 local c = self._next_checkpoint;
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
36 if c and h >= c then
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
37 self:_call_checkpoints(h);
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
38 end
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
39
12055
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
40 return acked
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
41 end
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
42
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
43 function smqueue:count_unacked() return self._head - self._tail end
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
44
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
45 function smqueue:count_acked() return self._tail end
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
46
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
47 function smqueue:resumable() return self._queue:count() >= (self._head - self._tail) end
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
48
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
49 function smqueue:resume() return self._queue:items() end
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
50
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
51 function smqueue:consume() return self._queue:consume() end
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
52
12058
4860da718e87 util.smqueue: Simplify compat table, fix dependent modules (thanks Martin)
Kim Alvefur <zash@zash.se>
parents: 12055
diff changeset
53 function smqueue:table()
4860da718e87 util.smqueue: Simplify compat table, fix dependent modules (thanks Martin)
Kim Alvefur <zash@zash.se>
parents: 12055
diff changeset
54 local t = {};
4860da718e87 util.smqueue: Simplify compat table, fix dependent modules (thanks Martin)
Kim Alvefur <zash@zash.se>
parents: 12055
diff changeset
55 for i, v in self:resume() do t[i] = v; end
4860da718e87 util.smqueue: Simplify compat table, fix dependent modules (thanks Martin)
Kim Alvefur <zash@zash.se>
parents: 12055
diff changeset
56 return t
12055
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
57 end
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
58
14001
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
59 -- Add a 'checkpoint' which will call a function when all the stanzas
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
60 -- currently in the queue have been successfully delivered and acknowledged
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
61 function smqueue:add_checkpoint(cb, ud)
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
62 table.insert(self._checkpoints, { self._head, cb, ud });
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
63 if not self._next_checkpoint then
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
64 self._next_checkpoint = self._head;
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
65 end
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
66 end
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
67
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
68 function smqueue:_call_checkpoints(h)
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
69 local checkpoints = self._checkpoints;
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
70
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
71 -- Pop the checkpoint info and update the next checkpoint info
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
72 local c = table.remove(checkpoints, 1);
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
73 self._next_checkpoint = checkpoints[1] and checkpoints[1][1] or nil;
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
74
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
75 -- Call the callback
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
76 local cb, ud = c[2], c[3];
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
77 cb(ud);
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
78
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
79 -- In case we passed over multiple checkpoints, call them also
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
80 if h >= self._next_checkpoint then
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
81 return self:_call_checkpoints(h);
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
82 end
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
83 end
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
84
13994
8baaefc4db79 util.smqueue: Minor reformatting for code style, add some comments
Matthew Wild <mwild1@gmail.com>
parents: 12975
diff changeset
85 local smqueue_mt = {
8baaefc4db79 util.smqueue: Minor reformatting for code style, add some comments
Matthew Wild <mwild1@gmail.com>
parents: 12975
diff changeset
86 __name = "smqueue";
8baaefc4db79 util.smqueue: Minor reformatting for code style, add some comments
Matthew Wild <mwild1@gmail.com>
parents: 12975
diff changeset
87 __index = smqueue;
8baaefc4db79 util.smqueue: Minor reformatting for code style, add some comments
Matthew Wild <mwild1@gmail.com>
parents: 12975
diff changeset
88 __len = smqueue.count_unacked;
8baaefc4db79 util.smqueue: Minor reformatting for code style, add some comments
Matthew Wild <mwild1@gmail.com>
parents: 12975
diff changeset
89 -- Return a simple object for serialization
8baaefc4db79 util.smqueue: Minor reformatting for code style, add some comments
Matthew Wild <mwild1@gmail.com>
parents: 12975
diff changeset
90 __freeze = function (q) return { head = q._head; tail = q._tail } end;
8baaefc4db79 util.smqueue: Minor reformatting for code style, add some comments
Matthew Wild <mwild1@gmail.com>
parents: 12975
diff changeset
91 };
12055
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
92
13994
8baaefc4db79 util.smqueue: Minor reformatting for code style, add some comments
Matthew Wild <mwild1@gmail.com>
parents: 12975
diff changeset
93 local lib = {};
12055
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
94
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
95 function lib.new(size)
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
96 assert(size > 0);
13994
8baaefc4db79 util.smqueue: Minor reformatting for code style, add some comments
Matthew Wild <mwild1@gmail.com>
parents: 12975
diff changeset
97 return setmetatable({
8baaefc4db79 util.smqueue: Minor reformatting for code style, add some comments
Matthew Wild <mwild1@gmail.com>
parents: 12975
diff changeset
98 _head = 0;
8baaefc4db79 util.smqueue: Minor reformatting for code style, add some comments
Matthew Wild <mwild1@gmail.com>
parents: 12975
diff changeset
99 _tail = 0;
8baaefc4db79 util.smqueue: Minor reformatting for code style, add some comments
Matthew Wild <mwild1@gmail.com>
parents: 12975
diff changeset
100 _queue = queue.new(size, true);
14001
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
101 _checkpoints = {};
13994
8baaefc4db79 util.smqueue: Minor reformatting for code style, add some comments
Matthew Wild <mwild1@gmail.com>
parents: 12975
diff changeset
102 }, smqueue_mt)
12055
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
103 end
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
104
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
105 return lib