diff options
| -rw-r--r-- | spec/util_smqueue_spec.lua | 55 | ||||
| -rw-r--r-- | teal-src/util/smqueue.tl | 107 | ||||
| -rw-r--r-- | util/smqueue.lua | 61 |
3 files changed, 223 insertions, 0 deletions
diff --git a/spec/util_smqueue_spec.lua b/spec/util_smqueue_spec.lua new file mode 100644 index 00000000..991a656b --- /dev/null +++ b/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); diff --git a/teal-src/util/smqueue.tl b/teal-src/util/smqueue.tl new file mode 100644 index 00000000..0bbceb8e --- /dev/null +++ b/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; diff --git a/util/smqueue.lua b/util/smqueue.lua new file mode 100644 index 00000000..d24c35ae --- /dev/null +++ b/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 |