Mercurial > prosody-hg

comparison 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
comparison
equal deleted inserted replaced
14000:071779f7645c 14001:df970d2dac45
30 local v = self._queue:pop(); 30 local v = self._queue:pop();
31 if not v then return nil, "pop" end 31 if not v then return nil, "pop" end
32 table.insert(acked, v); 32 table.insert(acked, v);
33 end 33 end
34 34
35 local c = self._next_checkpoint;
36 if c and h >= c then
37 self:_call_checkpoints(h);
38 end
39
35 return acked 40 return acked
36 end 41 end
37 42
38 function smqueue:count_unacked() return self._head - self._tail end 43 function smqueue:count_unacked() return self._head - self._tail end
39 44
49 local t = {}; 54 local t = {};
50 for i, v in self:resume() do t[i] = v; end 55 for i, v in self:resume() do t[i] = v; end
51 return t 56 return t
52 end 57 end
53 58
59 -- Add a 'checkpoint' which will call a function when all the stanzas
60 -- currently in the queue have been successfully delivered and acknowledged
61 function smqueue:add_checkpoint(cb, ud)
62 table.insert(self._checkpoints, { self._head, cb, ud });
63 if not self._next_checkpoint then
64 self._next_checkpoint = self._head;
65 end
66 end
67 };
68
69 function smqueue:_call_checkpoints(h)
70 local checkpoints = self._checkpoints;
71
72 -- Pop the checkpoint info and update the next checkpoint info
73 local c = table.remove(checkpoints, 1);
74 self._next_checkpoint = checkpoints[1] and checkpoints[1][1] or nil;
75
76 -- Call the callback
77 local cb, ud = c[2], c[3];
78 cb(ud);
79
80 -- In case we passed over multiple checkpoints, call them also
81 if h >= self._next_checkpoint then
82 return self:_call_checkpoints(h);
83 end
84 end
85
54 local smqueue_mt = { 86 local smqueue_mt = {
55 __name = "smqueue"; 87 __name = "smqueue";
56 __index = smqueue; 88 __index = smqueue;
57 __len = smqueue.count_unacked; 89 __len = smqueue.count_unacked;
58 -- Return a simple object for serialization 90 -- Return a simple object for serialization
65 assert(size > 0); 97 assert(size > 0);
66 return setmetatable({ 98 return setmetatable({
67 _head = 0; 99 _head = 0;
68 _tail = 0; 100 _tail = 0;
69 _queue = queue.new(size, true); 101 _queue = queue.new(size, true);
102 _checkpoints = {};
70 }, smqueue_mt) 103 }, smqueue_mt)
71 end 104 end
72 105
73 return lib 106 return lib