Mercurial > prosody-hg

annotate util/smqueue.lua @ 14001:df970d2dac45

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
author Matthew Wild <mwild1@gmail.com>
date Sat, 06 Dec 2025 15:57:36 +0000
parents 8baaefc4db79
children 21126415c8a0
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
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
69 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
70 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
71
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
72 -- 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
73 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
74 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
75
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
76 -- 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
77 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
78 cb(ud);
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
79
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
80 -- 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
81 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
82 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
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 end
df970d2dac45 util.smqueue: Add "checkpoints" API to run code on successful stanza delivery
Matthew Wild <mwild1@gmail.com>
parents: 13994
diff changeset
85
13994
8baaefc4db79 util.smqueue: Minor reformatting for code style, add some comments
Matthew Wild <mwild1@gmail.com>
parents: 12975
diff changeset
86 local smqueue_mt = {
8baaefc4db79 util.smqueue: Minor reformatting for code style, add some comments
Matthew Wild <mwild1@gmail.com>
parents: 12975
diff changeset
87 __name = "smqueue";
8baaefc4db79 util.smqueue: Minor reformatting for code style, add some comments
Matthew Wild <mwild1@gmail.com>
parents: 12975
diff changeset
88 __index = smqueue;
8baaefc4db79 util.smqueue: Minor reformatting for code style, add some comments
Matthew Wild <mwild1@gmail.com>
parents: 12975
diff changeset
89 __len = smqueue.count_unacked;
8baaefc4db79 util.smqueue: Minor reformatting for code style, add some comments
Matthew Wild <mwild1@gmail.com>
parents: 12975
diff changeset
90 -- 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
91 __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
92 };
12055
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
93
13994
8baaefc4db79 util.smqueue: Minor reformatting for code style, add some comments
Matthew Wild <mwild1@gmail.com>
parents: 12975
diff changeset
94 local lib = {};
12055
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
95
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
96 function lib.new(size)
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
97 assert(size > 0);
13994
8baaefc4db79 util.smqueue: Minor reformatting for code style, add some comments
Matthew Wild <mwild1@gmail.com>
parents: 12975
diff changeset
98 return setmetatable({
8baaefc4db79 util.smqueue: Minor reformatting for code style, add some comments
Matthew Wild <mwild1@gmail.com>
parents: 12975
diff changeset
99 _head = 0;
8baaefc4db79 util.smqueue: Minor reformatting for code style, add some comments
Matthew Wild <mwild1@gmail.com>
parents: 12975
diff changeset
100 _tail = 0;
8baaefc4db79 util.smqueue: Minor reformatting for code style, add some comments
Matthew Wild <mwild1@gmail.com>
parents: 12975
diff changeset
101 _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
102 _checkpoints = {};
13994
8baaefc4db79 util.smqueue: Minor reformatting for code style, add some comments
Matthew Wild <mwild1@gmail.com>
parents: 12975
diff changeset
103 }, smqueue_mt)
12055
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
104 end
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
105
daced16154fa util.smqueue: Abstract queue with acknowledgements and overflow
Kim Alvefur <zash@zash.se>
parents:
diff changeset
106 return lib