util.async: Add sleep() method with configurable scheduling backend

No scheduler set by default, so it will error (we plan to initialize it in
util.startup).

We wanted to avoid a hard dependency on util.timer (which in turn depends on
network backends, etc.), and we didn't add timer.sleep() because we didn't
want to add a hard dependency on util.async for things that don't need it.

spec/util_async_spec.lua

@@ -615,4 +615,58 @@ describe("util.async", function()
 			assert.spy(r.watchers.error).was_not.called();
 		end);
 	end);
+
+	describe("#sleep()", function ()
+		after_each(function ()
+			-- Restore to default
+			async.set_schedule_function(nil);
+		end);
+
+		it("should fail if no scheduler configured", function ()
+			local r = new(function ()
+				async.sleep(5);
+			end);
+			r:run(true);
+			assert.spy(r.watchers.error).was.called();
+
+			-- Set dummy scheduler
+			async.set_schedule_function(function () end);
+
+			local r2 = new(function ()
+				async.sleep(5);
+			end);
+			r2:run(true);
+			assert.spy(r2.watchers.error).was_not.called();
+		end);
+		it("should work", function ()
+			local queue = {};
+			local add_task = spy.new(function (t, f)
+				table.insert(queue, { t, f });
+			end);
+			async.set_schedule_function(add_task);
+
+			local processed_item;
+			local r = new(function (item)
+				async.sleep(5);
+				processed_item = item;
+			end);
+			r:run("test");
+
+			-- Nothing happened, because the runner is sleeping
+			assert.is_nil(processed_item);
+			assert.equal(r.state, "waiting");
+			assert.spy(add_task).was_called(1);
+			assert.spy(add_task).was_called_with(match.is_number(), match.is_function());
+			assert.spy(r.watchers.waiting).was.called();
+			assert.spy(r.watchers.ready).was_not.called();
+
+			-- Pretend the timer has triggered, call the handler
+			queue[1][2]();
+
+			assert.equal(processed_item, "test");
+			assert.equal(r.state, "ready");
+
+			assert.spy(r.watchers.ready).was.called();
+		end);
+	end);
 end);

util/async.lua

@@ -11,6 +11,9 @@ local function checkthread()
 	return thread;
 end
 
+-- Configurable functions
+local schedule_task = nil; -- schedule_task(seconds, callback)
+
 local function runner_from_thread(thread)
 	local level = 0;
 	-- Find the 'level' of the top-most function (0 == current level, 1 == caller, ...)
@@ -118,6 +121,15 @@ local function guarder()
 	end;
 end
 
+local function sleep(seconds)
+	if not schedule_task then
+		error("async.sleep() is not available - configure schedule function");
+	end
+	local wait, done = waiter();
+	schedule_task(seconds, done);
+	wait();
+end
+
 local runner_mt = {};
 runner_mt.__index = runner_mt;
 
@@ -272,4 +284,7 @@ return {
 	runner = runner;
 	wait = wait_for; -- COMPAT w/trunk pre-0.12
 	wait_for = wait_for;
+	sleep = sleep;
+
+	set_schedule_function = function (new_schedule_function) schedule_task = new_schedule_function; end;
 };