Mercurial > prosody-hg

comparison util/smqueue.lua @ 14017:1e01b91cf94d

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
util.smqueue: Backport changes to Teal source
author Kim Alvefur <zash@zash.se>
date Sun, 07 Dec 2025 15:13:09 +0100
parents 21126415c8a0
children 73903352de86
comparison
equal deleted inserted replaced
14016:1df1782ccdfe 14017:1e01b91cf94d
1 -- Represents an outgoing reliable stanza queue 1 -- This file is generated from teal-src/prosody/util/smqueue.tl
2
3 local queue = require("prosody.util.queue"); 2 local queue = require("prosody.util.queue");
4 3
5 -- SM queue methods 4 local lib = { smqueue = {} }
6 local smqueue = {}; 5
6 local smqueue = lib.smqueue;
7 7
8 function smqueue:push(v) 8 function smqueue:push(v)
9 self._head = self._head + 1; 9 self._head = self._head + 1;
10 10
11 assert(self._queue:push(v)); 11 assert(self._queue:push(v));
12 end 12 end
13 13
14 -- Call when receiver has acknowledged some stanzas
15 -- `h` (sent by the receiver) is the running count of
16 -- successfully received stanzas
17 function smqueue:ack(h) 14 function smqueue:ack(h)
18 if h < self._tail then 15 if h < self._tail then
19 -- h is less than a previous h 16
20 return nil, "tail" 17 return nil, "tail"
21 elseif h > self._head then 18 elseif h > self._head then
22 -- h is greater than the number of stanzas we sent 19
23 return nil, "head" 20 return nil, "head"
24 end 21 end
25 22
26 local acked = {}; 23 local acked = {};
27 self._tail = h; 24 self._tail = h;
28 local expect = self._head - self._tail; 25 local expect = self._head - self._tail;
29 while expect < self._queue:count() do 26 while expect < self._queue:count() do
30 local v = self._queue:pop(); 27 local v = self._queue:pop();
31 if not v then return nil, "pop" end 28 if not v then
29 return nil, "pop"
30 end
32 table.insert(acked, v); 31 table.insert(acked, v);
33 end 32 end
34 33
35 local c = self._next_checkpoint; 34 local c = self._next_checkpoint;
36 if c and h >= c then 35 if c and h >= c then
37 self:_call_checkpoints(h); 36 self:_call_checkpoints(h);
38 end 37 end
39
40 return acked 38 return acked
41 end 39 end
42 40
43 function smqueue:count_unacked() return self._head - self._tail end 41 function smqueue:count_unacked()
42 return self._head - self._tail
43 end
44 44
45 function smqueue:count_acked() return self._tail end 45 function smqueue:count_acked()
46 return self._tail
47 end
46 48
47 function smqueue:resumable() return self._queue:count() >= (self._head - self._tail) end 49 function smqueue:resumable()
50 return self._queue:count() >= (self._head - self._tail)
51 end
48 52
49 function smqueue:resume() return self._queue:items() end 53 function smqueue:resume()
54 return self._queue:items()
55 end
50 56
51 function smqueue:consume() return self._queue:consume() end 57 function smqueue:consume()
58 return self._queue:consume()
59 end
52 60
53 function smqueue:table() 61 function smqueue:table()
54 local t = {}; 62 local t = {};
55 for i, v in self:resume() do t[i] = v; end 63 for i, v in self:resume() do
64 t[i] = v;
65 end
56 return t 66 return t
57 end 67 end
58 68
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) 69 function smqueue:add_checkpoint(cb, ud)
62 table.insert(self._checkpoints, { self._head, cb, ud }); 70 table.insert(self._checkpoints, { self._head; cb; ud });
63 if not self._next_checkpoint then 71 if not self._next_checkpoint then
64 self._next_checkpoint = self._head; 72 self._next_checkpoint = self._head;
65 end 73 end
66 end 74 end
67 75
68 function smqueue:_call_checkpoints(h) 76 function smqueue:_call_checkpoints(h)
69 local checkpoints = self._checkpoints; 77 local checkpoints = self._checkpoints;
70 78
71 -- Pop the checkpoint info and update the next checkpoint info
72 local c = table.remove(checkpoints, 1); 79 local c = table.remove(checkpoints, 1);
73 self._next_checkpoint = checkpoints[1] and checkpoints[1][1] or nil; 80 self._next_checkpoint = checkpoints[1] and checkpoints[1][1] or nil;
74 81
75 -- Call the callback
76 local cb, ud = c[2], c[3]; 82 local cb, ud = c[2], c[3];
77 cb(ud); 83 cb(ud);
78 84
79 -- In case we passed over multiple checkpoints, call them also
80 if h >= self._next_checkpoint then 85 if h >= self._next_checkpoint then
81 return self:_call_checkpoints(h); 86 return self:_call_checkpoints(h)
82 end 87 end
83 end 88 end
84 89
85 local smqueue_mt = { 90 local function freeze(q)
86 __name = "smqueue"; 91 return { head = q._head; tail = q._tail }
87 __index = smqueue; 92 end
88 __len = smqueue.count_unacked;
89 -- Return a simple object for serialization
90 __freeze = function (q) return { head = q._head; tail = q._tail } end;
91 };
92 93
93 local lib = {}; 94 local smqueue_mt = { __name = "smqueue"; __index = smqueue; __len = smqueue.count_unacked; __freeze = freeze }
94 95
95 function lib.new(size) 96 function lib.new(size)
96 assert(size > 0); 97 assert(size > 0);
97 return setmetatable({ 98 return setmetatable({ _head = 0; _tail = 0; _queue = queue.new(size, true); _checkpoints = {} }, smqueue_mt)
98 _head = 0;
99 _tail = 0;
100 _queue = queue.new(size, true);
101 _checkpoints = {};
102 }, smqueue_mt)
103 end 99 end
104 100
105 return lib 101 return lib