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);

			end;

		})

	end

	local function new(func, name)

		local event_log = {};

		local spy_func = spy.new(func);

		return async.runner(spy_func, mock_watchers(event_log)), event_log, spy_func;

	end

	describe("#runner", function()

		it("should work", function()

			local r, l = new(function (item) assert(type(item) == "number") end);

			r:run(1);

			r:run(2);

		end);

		it("should be ready after creation", function ()

			local r = async.runner(function (item) end);

			assert.equal(r.state, "ready");

		end);

		it("should do nothing if the queue is empty", function ()

			local did_run;

			local r = async.runner(function (item) 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 = async.runner(function (item) 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 s = spy(function (item) last_item = item; end);

			local r = async.runner(s);

			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 s = spy(function (item) last_item = item; end);

			local r = async.runner(s);

			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);

		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);

			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);

			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);

			it("should continue to process work items", function ()

				local wait, done, last_item;

				local runner_func = spy.new(function (item)

					if item == "error" then

						error("test error");

					end

					last_item = item;

				end);

				local runner = async.runner(runner_func, mock{

					ready = function () end;

					waiting = function () end;

					error = function () end;

				});

				runner:enqueue("one");

				runner:enqueue("error");

				runner:enqueue("two");

				runner:run();

				assert.equal(r.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_func = spy.new(function (item)

					if item == "wait-error" then

						wait, done = async.waiter();

						wait();

						error("test error");

					end

					last_item = item;

				end);

				local runner = async.runner(runner_func, mock{

					ready = function () end;

					waiting = function () end;

					error = function () end;

				});

				runner:enqueue("one");

				runner:enqueue("wait-error");

				runner:enqueue("two");

				runner:run();

				done();

				assert.equal(r.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, l = 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, l = 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, l = 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, l = 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);