File

util/smqueue.lua @ 12301:4f1fe6eb1ddb

mod_c2s,mod_s2s: Wait for sessions to close before proceeding with shutdown steps Ensures unavailable presence and other outgoing stanzas are sent. Waiting for c2s sessions to close first before proceeding to disable and close s2s ensures that unavailable presence can go out, even if it requires dialback to complete first.
author Kim Alvefur <zash@zash.se>
date Thu, 17 Feb 2022 03:49:47 +0100
parent 12058:4860da718e87
child 12975:d10957394a3c
line wrap: on
line source

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

function smqueue:table()
	local t = {};
	for i, v in self:resume() do t[i] = v; end
	return t
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