util.smqueue: Abstract queue with acknowledgements and overflow

Meant to be used in mod_smacks for XEP-0198

Meant to have a larger virtual size than actual number of items stored,
on the theory that in most cases, the excess will be acked before needed
for a resumption event.

spec/util_smqueue_spec.lua

@@ -0,0 +1,55 @@
+describe("util.smqueue", function()
+
+	local smqueue
+	setup(function() smqueue = require "util.smqueue"; end)
+
+	describe("#new()", function()
+		it("should work", function()
+			local q = smqueue.new(10);
+			assert.truthy(q);
+		end)
+	end)
+
+	describe("#push()", function()
+		it("should allow pushing many items", function()
+			local q = smqueue.new(10);
+			for i = 1, 20 do q:push(i); end
+			assert.equal(20, q:count_unacked());
+		end)
+	end)
+
+	describe("#resumable()", function()
+		it("returns true while the queue is small", function()
+			local q = smqueue.new(10);
+			for i = 1, 10 do q:push(i); end
+			assert.truthy(q:resumable());
+			q:push(11);
+			assert.falsy(q:resumable());
+		end)
+	end)
+
+	describe("#ack", function()
+		it("allows removing items", function()
+			local q = smqueue.new(10);
+			for i = 1, 10 do q:push(i); end
+			assert.same({ 1; 2; 3 }, q:ack(3));
+			assert.same({ 4; 5; 6 }, q:ack(6));
+			assert.falsy(q:ack(3), "can't go backwards")
+			assert.falsy(q:ack(100), "can't ack too many")
+			for i = 11, 20 do q:push(i); end
+			assert.same({ 11; 12 }, q:ack(12), "items are dropped");
+		end)
+	end)
+
+	describe("#resume", function()
+		it("iterates over current items", function()
+			local q = smqueue.new(10);
+			for i = 1, 12 do q:push(i); end
+			assert.same({ 3; 4; 5; 6 }, q:ack(6));
+			assert.truthy(q:resumable());
+			local resume = {}
+			for _, i in q:resume() do resume[i] = true end
+			assert.same({ [7] = true; [8] = true; [9] = true; [10] = true; [11] = true; [12] = true }, resume);
+		end)
+	end)
+end);

teal-src/util/smqueue.tl

@@ -0,0 +1,107 @@
+local queue = require "util.queue";
+
+local record lib
+	-- T would typically be util.stanza
+	record smqueue<T>
+		_queue : queue.queue<T>
+		_head : integer
+		_tail : integer
+
+		enum ack_errors
+			"tail"
+			"head"
+			"pop"
+		end
+		push : function (smqueue, T)
+		ack : function (smqueue, integer) : { T }, ack_errors
+		resumable : function (smqueue) : boolean
+		type consume_iter = function (smqueue<T>) : T
+		consume : function (smqueue<T>) : consume_iter
+
+		table : function (smqueue<T>) : { T }
+	end
+	new : function <T>(integer) : smqueue<T>
+end
+
+local type smqueue = lib.smqueue;
+
+function smqueue:push(v)
+	self._head = self._head + 1;
+	-- Wraps instead of errors
+	assert(self._queue:push(v));
+end
+
+function smqueue:ack(h : integer) : { any }, smqueue.ack_errors
+	if h < self._tail then
+		return nil, "tail";
+	elseif h > self._head then
+		return nil, "head";
+	end
+	-- TODO optimize? cache table fields
+	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() : integer
+	return self._head - self._tail;
+end
+
+function smqueue:count_acked() : integer
+	return self._tail;
+end
+
+function smqueue:resumable() : boolean
+	return self._queue:count() >= (self._head - self._tail);
+end
+
+function smqueue:resume() : queue.queue.iterator, any, integer
+	return self._queue:items();
+end
+
+function smqueue:consume() : queue.queue.consume_iter
+	return self._queue:consume()
+end
+
+-- Compatibility wrapper, meant to look like a plain ol' array
+local record compat_mt
+	_queue : smqueue<any>
+end
+
+function compat_mt:__index(i : integer) : any
+	if i < self._queue._tail then return nil end
+	return self._queue._queue._items[(i + self._queue._tail) % self._queue._queue.size];
+end
+
+function compat_mt:__len() : integer
+	return self._queue:count_unacked()
+end
+
+function smqueue:table() : { any }
+	return setmetatable({ _queue = self }, compat_mt);
+end
+
+local function freeze(q : smqueue<any>) : { string:integer }
+	return { head = q._head, tail = q._tail }
+end
+
+local queue_mt = {
+	--
+	__name = "smqueue";
+	__index = smqueue;
+	__len = smqueue.count_unacked;
+	__freeze = freeze;
+}
+
+function lib.new<T>(size : integer) : queue.queue<T>
+	assert(size>0);
+	return setmetatable({ _head = 0; _tail = 0; _queue = queue.new(size, true) }, queue_mt);
+end
+
+return lib;

util/smqueue.lua

@@ -0,0 +1,61 @@
+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
+
+local compat_mt = {}
+
+function compat_mt:__index(i)
+	if i < self._queue._tail then return nil end
+	return self._queue._queue._items[(i + self._queue._tail) % self._queue._queue.size]
+end
+
+function compat_mt:__len() return self._queue:count_unacked() end
+
+function smqueue:table() return setmetatable({ _queue = self }, compat_mt) 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