local async = require "util.async";
describe("util.async", function()
local debug = false;
local print = print;
if debug then
require "util.logger".add_simple_sink(print);
else
print = function () end
end
local function mock_watchers(event_log)
local function generic_logging_watcher(name)
return function (...)
table.insert(event_log, { name = name, n = select("#", ...)-1, select(2, ...) });
end;
end;
return setmetatable(mock{
ready = generic_logging_watcher("ready");
waiting = generic_logging_watcher("waiting");
error = generic_logging_watcher("error");
}, {
__index = function (_, event)
-- Unexpected watcher called
assert(false, "unexpected watcher called: "..event);
end;
})
end
local function new(func)
local event_log = {};
local spy_func = spy.new(func);
return async.runner(spy_func, mock_watchers(event_log)), spy_func, event_log;
end
describe("#runner", function()
it("should work", function()
local r = new(function (item) assert(type(item) == "number") end);
r:run(1);
r:run(2);
end);
it("should be ready after creation", function ()
local r = new(function () end);
assert.equal(r.state, "ready");
end);
it("should do nothing if the queue is empty", function ()
local did_run;
local r = new(function () did_run = true end);
r:run();
assert.equal(r.state, "ready");
assert.is_nil(did_run);
r:run("hello");
assert.is_true(did_run);
end);
it("should support queuing work items without running", function ()
local did_run;
local r = new(function () did_run = true end);
r:enqueue("hello");
assert.equal(r.state, "ready");
assert.is_nil(did_run);
r:run();
assert.is_true(did_run);
end);
it("should support queuing multiple work items", function ()
local last_item;
local r, s = new(function (item) last_item = item; end);
r:enqueue("hello");
r:enqueue("there");
r:enqueue("world");
assert.equal(r.state, "ready");
r:run();
assert.equal(r.state, "ready");
assert.spy(s).was.called(3);
assert.equal(last_item, "world");
end);
it("should support all simple data types", function ()
local last_item;
local r, s = new(function (item) last_item = item; end);
local values = { {}, 123, "hello", true, false };
for i = 1, #values do
r:enqueue(values[i]);
end
assert.equal(r.state, "ready");
r:run();
assert.equal(r.state, "ready");
assert.spy(s).was.called(#values);
for i = 1, #values do
assert.spy(s).was.called_with(values[i]);
end
assert.equal(last_item, values[#values]);
end);
it("should work with no parameters", function ()
local item = "fail";
local r = async.runner();
local f = spy.new(function () item = "success"; end);
r:run(f);
assert.spy(f).was.called();
assert.equal(item, "success");
end);
it("supports a default error handler", function ()
local item = "fail";
local r = async.runner();
local f = spy.new(function () error("test error"); end);
assert.error_matches(function ()
r:run(f);
end, "test error");
assert.spy(f).was.called();
assert.equal(item, "fail");
end);
describe("#errors", function ()
describe("should notify", function ()
local last_processed_item, last_error;
local r;
r = async.runner(function (item)
if item == "error" then
error({ e = "test error" });
end
last_processed_item = item;
end, mock{
ready = function () end;
waiting = function () end;
error = function (runner, err)
assert.equal(r, runner);
last_error = err;
end;
});
-- Simple item, no error
r:run("hello");
assert.equal(r.state, "ready");
assert.equal(last_processed_item, "hello");
assert.spy(r.watchers.ready).was_not.called();
assert.spy(r.watchers.error).was_not.called();
-- Trigger an error inside the runner
assert.equal(last_error, nil);
r:run("error");
test("the correct watcher functions", function ()
-- Only the error watcher should have been called
assert.spy(r.watchers.ready).was_not.called();
assert.spy(r.watchers.waiting).was_not.called();
assert.spy(r.watchers.error).was.called(1);
end);
test("with the correct error", function ()
-- The error watcher state should be correct, to
-- demonstrate the error was passed correctly
assert.is_table(last_error);
assert.equal(last_error.e, "test error");
last_error = nil;
end);
assert.equal(r.state, "ready");
assert.equal(last_processed_item, "hello");
end);
do
local last_processed_item, last_error;
local r;
local wait, done;
r = async.runner(function (item)
if item == "error" then
error({ e = "test error" });
elseif item == "wait" then
wait, done = async.waiter();
wait();
error({ e = "post wait error" });
end
last_processed_item = item;
end, mock({
ready = function () end;
waiting = function () end;
error = function (runner, err)
assert.equal(r, runner);
last_error = err;
end;
}));
randomize(false); --luacheck: ignore 113/randomize
it("should not be fatal to the runner", function ()
r:run("world");
assert.equal(r.state, "ready");
assert.spy(r.watchers.ready).was_not.called();
assert.equal(last_processed_item, "world");
end);
it("should work despite a #waiter", function ()
-- This test covers an important case where a runner
-- throws an error while being executed outside of the
-- main loop. This happens when it was blocked ('waiting'),
-- and then released (via a call to done()).
last_error = nil;
r:run("wait");
assert.equal(r.state, "waiting");
assert.spy(r.watchers.waiting).was.called(1);
done();
-- At this point an error happens (state goes error->ready)
assert.equal(r.state, "ready");
assert.spy(r.watchers.error).was.called(1);
assert.spy(r.watchers.ready).was.called(1);
assert.is_table(last_error);
assert.equal(last_error.e, "post wait error");
last_error = nil;
r:run("hello again");
assert.spy(r.watchers.ready).was.called(1);
assert.spy(r.watchers.waiting).was.called(1);
assert.spy(r.watchers.error).was.called(1);
assert.equal(r.state, "ready");
assert.equal(last_processed_item, "hello again");
end);
end
it("should continue to process work items", function ()
local last_item;
local runner, runner_func = new(function (item)
if item == "error" then
error("test error");
end
last_item = item;
end);
runner:enqueue("one");
runner:enqueue("error");
runner:enqueue("two");
runner:run();
assert.equal(runner.state, "ready");
assert.spy(runner_func).was.called(3);
assert.spy(runner.watchers.error).was.called(1);
assert.spy(runner.watchers.ready).was.called(0);
assert.spy(runner.watchers.waiting).was.called(0);
assert.equal(last_item, "two");
end);
it("should continue to process work items during resume", function ()
local wait, done, last_item;
local runner, runner_func = new(function (item)
if item == "wait-error" then
wait, done = async.waiter();
wait();
error("test error");
end
last_item = item;
end);
runner:enqueue("one");
runner:enqueue("wait-error");
runner:enqueue("two");
runner:run();
done();
assert.equal(runner.state, "ready");
assert.spy(runner_func).was.called(3);
assert.spy(runner.watchers.error).was.called(1);
assert.spy(runner.watchers.waiting).was.called(1);
assert.spy(runner.watchers.ready).was.called(1);
assert.equal(last_item, "two");
end);
end);
end);
describe("#waiter", function()
it("should error outside of async context", function ()
assert.has_error(function ()
async.waiter();
end);
end);
it("should work", function ()
local wait, done;
local r = new(function (item)
assert(type(item) == "number")
if item == 3 then
wait, done = async.waiter();
wait();
end
end);
r:run(1);
assert(r.state == "ready");
r:run(2);
assert(r.state == "ready");
r:run(3);
assert(r.state == "waiting");
done();
assert(r.state == "ready");
--for k, v in ipairs(l) do print(k,v) end
end);
it("should work", function ()
--------------------
local wait, done;
local last_item = 0;
local r = new(function (item)
assert(type(item) == "number")
assert(item == last_item + 1);
last_item = item;
if item == 3 then
wait, done = async.waiter();
wait();
end
end);
r:run(1);
assert(r.state == "ready");
r:run(2);
assert(r.state == "ready");
r:run(3);
assert(r.state == "waiting");
r:run(4);
assert(r.state == "waiting");
done();
assert(r.state == "ready");
--for k, v in ipairs(l) do print(k,v) end
end);
it("should work", function ()
--------------------
local wait, done;
local last_item = 0;
local r = new(function (item)
assert(type(item) == "number")
assert((item == last_item + 1) or item == 3);
last_item = item;
if item == 3 then
wait, done = async.waiter();
wait();
end
end);
r:run(1);
assert(r.state == "ready");
r:run(2);
assert(r.state == "ready");
r:run(3);
assert(r.state == "waiting");
r:run(3);
assert(r.state == "waiting");
r:run(3);
assert(r.state == "waiting");
r:run(4);
assert(r.state == "waiting");
for i = 1, 3 do
done();
if i < 3 then
assert(r.state == "waiting");
end
end
assert(r.state == "ready");
--for k, v in ipairs(l) do print(k,v) end
end);
it("should work", function ()
--------------------
local wait, done;
local last_item = 0;
local r = new(function (item)
assert(type(item) == "number")
assert((item == last_item + 1) or item == 3);
last_item = item;
if item == 3 then
wait, done = async.waiter();
wait();
end
end);
r:run(1);
assert(r.state == "ready");
r:run(2);
assert(r.state == "ready");
r:run(3);
assert(r.state == "waiting");
r:run(3);
assert(r.state == "waiting");
for i = 1, 2 do
done();
if i < 2 then
assert(r.state == "waiting");
end
end
assert(r.state == "ready");
r:run(4);
assert(r.state == "ready");
assert(r.state == "ready");
--for k, v in ipairs(l) do print(k,v) end
end);
it("should work with multiple runners in parallel", function ()
-- Now with multiple runners
--------------------
local wait1, done1;
local last_item1 = 0;
local r1 = new(function (item)
assert(type(item) == "number")
assert((item == last_item1 + 1) or item == 3);
last_item1 = item;
if item == 3 then
wait1, done1 = async.waiter();
wait1();
end
end, "r1");
local wait2, done2;
local last_item2 = 0;
local r2 = new(function (item)
assert(type(item) == "number")
assert((item == last_item2 + 1) or item == 3);
last_item2 = item;
if item == 3 then
wait2, done2 = async.waiter();
wait2();
end
end, "r2");
r1:run(1);
assert(r1.state == "ready");
r1:run(2);
assert(r1.state == "ready");
r1:run(3);
assert(r1.state == "waiting");
r1:run(3);
assert(r1.state == "waiting");
r2:run(1);
assert(r1.state == "waiting");
assert(r2.state == "ready");
r2:run(2);
assert(r1.state == "waiting");
assert(r2.state == "ready");
r2:run(3);
assert(r1.state == "waiting");
assert(r2.state == "waiting");
done2();
r2:run(3);
assert(r1.state == "waiting");
assert(r2.state == "waiting");
done2();
r2:run(4);
assert(r1.state == "waiting");
assert(r2.state == "ready");
for i = 1, 2 do
done1();
if i < 2 then
assert(r1.state == "waiting");
end
end
assert(r1.state == "ready");
r1:run(4);
assert(r1.state == "ready");
assert(r1.state == "ready");
--for k, v in ipairs(l1) do print(k,v) end
end);
it("should work work with multiple runners in parallel", function ()
--------------------
local wait1, done1;
local last_item1 = 0;
local r1 = new(function (item)
print("r1 processing ", item);
assert(type(item) == "number")
assert((item == last_item1 + 1) or item == 3);
last_item1 = item;
if item == 3 then
wait1, done1 = async.waiter();
wait1();
end
end, "r1");
local wait2, done2;
local last_item2 = 0;
local r2 = new(function (item)
print("r2 processing ", item);
assert.is_number(item);
assert((item == last_item2 + 1) or item == 3);
last_item2 = item;
if item == 3 then
wait2, done2 = async.waiter();
wait2();
end
end, "r2");
r1:run(1);
assert.equal(r1.state, "ready");
r1:run(2);
assert.equal(r1.state, "ready");
r1:run(5);
assert.equal(r1.state, "ready");
r1:run(3);
assert.equal(r1.state, "waiting");
r1:run(5); -- Will error, when we get to it
assert.equal(r1.state, "waiting");
done1();
assert.equal(r1.state, "ready");
r1:run(3);
assert.equal(r1.state, "waiting");
r2:run(1);
assert.equal(r1.state, "waiting");
assert.equal(r2.state, "ready");
r2:run(2);
assert.equal(r1.state, "waiting");
assert.equal(r2.state, "ready");
r2:run(3);
assert.equal(r1.state, "waiting");
assert.equal(r2.state, "waiting");
done2();
assert.equal(r1.state, "waiting");
assert.equal(r2.state, "ready");
r2:run(3);
assert.equal(r1.state, "waiting");
assert.equal(r2.state, "waiting");
done2();
assert.equal(r1.state, "waiting");
assert.equal(r2.state, "ready");
r2:run(4);
assert.equal(r1.state, "waiting");
assert.equal(r2.state, "ready");
done1();
assert.equal(r1.state, "ready");
r1:run(4);
assert.equal(r1.state, "ready");
assert.equal(r1.state, "ready");
end);
it("should support multiple done() calls", function ()
local processed_item;
local wait, done;
local r, rf = new(function (item)
wait, done = async.waiter(4);
wait();
processed_item = item;
end);
r:run("test");
for _ = 1, 3 do
done();
assert.equal(r.state, "waiting");
assert.is_nil(processed_item);
end
done();
assert.equal(r.state, "ready");
assert.equal(processed_item, "test");
assert.spy(r.watchers.error).was_not.called();
end);
it("should not allow done() to be called more than specified", function ()
local processed_item;
local wait, done;
local r, rf = new(function (item)
wait, done = async.waiter(4);
wait();
processed_item = item;
end);
r:run("test");
for _ = 1, 4 do
done();
end
assert.has_error(done);
assert.equal(r.state, "ready");
assert.equal(processed_item, "test");
assert.spy(r.watchers.error).was_not.called();
end);
it("should allow done() to be called before wait()", function ()
local processed_item;
local r, rf = new(function (item)
local wait, done = async.waiter();
done();
wait();
processed_item = item;
end);
r:run("test");
assert.equal(processed_item, "test");
assert.equal(r.state, "ready");
-- Since the observable state did not change,
-- the watchers should not have been called
assert.spy(r.watchers.waiting).was_not.called();
assert.spy(r.watchers.ready).was_not.called();
end);
end);
describe("#ready()", function ()
it("should return false outside an async context", function ()
assert.falsy(async.ready());
end);
it("should return true inside an async context", function ()
local r = new(function ()
assert.truthy(async.ready());
end);
r:run(true);
assert.spy(r.func).was.called();
assert.spy(r.watchers.error).was_not.called();
end);
end);
end);