util.pubsub: Add support for limiting number of items to retrieve

Hopefully this will eventually be upgraded to RSM, which is why the
argument is called 'resultspec' and is a table.

spec/util_pubsub_spec.lua

@@ -528,4 +528,61 @@ describe("util.pubsub", function ()
 		end);
 
 	end)
+
+	describe("max_items", function ()
+		it("works", function ()
+			local service = pubsub.new { };
+
+			local ok = service:create("node", true)
+			assert.truthy(ok);
+
+			for i = 1, 20 do
+				assert.truthy(service:publish("node", true, "item"..tostring(i), "data"..tostring(i)));
+			end
+
+			do
+				local ok, items = service:get_items("node", true, nil, { max = 3 });
+				assert.truthy(ok, items);
+				assert.equal(3, #items);
+				assert.same({
+						"item20",
+						"item19",
+						"item18",
+						item20 = "data20",
+						item19 = "data19",
+						item18 = "data18",
+					}, items, "items should be ordered by oldest first");
+			end
+
+			do
+				local ok, items = service:get_items("node", true, nil, { max = 10 });
+				assert.truthy(ok, items);
+				assert.equal(10, #items);
+				assert.same({
+						"item20",
+						"item19",
+						"item18",
+						"item17",
+						"item16",
+						"item15",
+						"item14",
+						"item13",
+						"item12",
+						"item11",
+						item20 = "data20",
+						item19 = "data19",
+						item18 = "data18",
+						item17 = "data17",
+						item16 = "data16",
+						item15 = "data15",
+						item14 = "data14",
+						item13 = "data13",
+						item12 = "data12",
+						item11 = "data11",
+					}, items, "items should be ordered by oldest first");
+			end
+
+		end);
+
+	end)
 end);

util/pubsub.lua

@@ -642,7 +642,7 @@ function service:purge(node, actor, notify) --> ok, err
 	return true
 end
 
-function service:get_items(node, actor, ids) --> (true, { id, [id] = node }) or (false, err)
+function service:get_items(node, actor, ids, resultspec) --> (true, { id, [id] = node }) or (false, err)
 	-- Access checking
 	if not self:may(node, actor, "get_items") then
 		return false, "forbidden";
@@ -660,18 +660,23 @@ function service:get_items(node, actor, ids) --> (true, { id, [id] = node }) or
 		ids = { ids };
 	end
 	local data = {};
+	local limit = resultspec and resultspec.max;
 	if ids then
 		for _, key in ipairs(ids) do
 			local value = self.data[node]:get(key);
 			if value then
 				data[#data+1] = key;
 				data[key] = value;
+				-- Limits and ids seem like a problematic combination.
+				if limit and #data >= limit then break end
 			end
 		end
 	else
 		for key, value in self.data[node]:items() do
 			data[#data+1] = key;
 			data[key] = value;
+			if limit and #data >= limit then break
+			end
 		end
 	end
 	return true, data;