Added group_id and until_group() to flow (#19)

* Added `group_id` and `until_group()` to flow

* Moved `ensure_hash(group_id)` outside the loop

examples/flow/flow.script

@@ -88,6 +88,47 @@ function init(self)
 
 		-- play a sprite animation and wait until it is done
 		flow.play_animation("#sprite", "green_walk_once")
+		print("flow.play_animation() is done")
+
+		-- start two parallel flows and store them to the table "foo"
+		print("Parallel flows stored to the table 'foo' are starting")
+		local foo = {}
+
+		-- start a first parallel flow
+		foo[1] = flow.start(function()
+			flow.delay(math.random())
+			print("Parallel flow foo 1 completed")
+		end)
+
+		-- start a second parallel flow
+		foo[2] = flow.start(function()
+			flow.delay(math.random())
+			print("Parallel flow foo 2 completed")
+		end)
+
+		-- wait until all the flows in the table "foo" have completed
+		flow.until_flows(foo)
+		print("All the parallel stored to the table completed")
+
+		-- start two parallel flows with the same group_id "bar"
+		print("Parallel flows with group_id 'bar' are starting")
+
+		-- start a parallel flow with the group_id "bar"
+		flow.parallel_group("bar", function()
+			flow.delay(math.random())
+			print("Parallel flow bar 1 completed")
+		end)
+
+		-- start an another parallel flow with the group_id "bar"
+		flow.parallel_group("bar", function()
+			flow.delay(math.random())
+			print("Parallel flow bar 2 completed")
+		end)
+
+		-- wait until all the flows with the group_id "bar" have completed
+		flow.until_group("bar")
+		print("All the parallel flows with group_id 'bar' completed")
+
 		print("DONE")
 	end)
 

ludobits/m/flow.lua

@@ -109,7 +109,7 @@ end
 
 local id_counter = 0
 
-local function create_or_get(co)
+local function create_or_get(co, group_id)
 	assert(co, "You must provide a coroutine")
 	if not instances[co] then
 		id_counter = id_counter + 1
@@ -118,6 +118,7 @@ local function create_or_get(co)
 			url = msg.url(),
 			state = READY,
 			co = co,
+			group_id = group_id and ensure_hash(group_id),
 			timer_id = timer.delay(0, true, function(self, handle, time_elapsed)
 				if not instances[co] then
 					timer.cancel(handle)
@@ -137,12 +138,15 @@ end
 -- @param fn The function to run within the flow
 -- @param options Key value pairs. Allowed keys:
 --		parallel = true if running flow shouldn't wait for this flow
+--		group_id = hash("group") if you want to use it in until_group() later
 -- @param on_error Function to call if something goes wrong while
 -- running the flow
 -- @return The created flow instance
 function M.start(fn, options, on_error)
 	assert(fn, "You must provide a function")
-	local created_instance = create_or_get(coroutine.create(fn))
+	local group_id = options and ensure_hash(options.group_id)
+
+	local created_instance = create_or_get(coroutine.create(fn), group_id)
 	created_instance.on_error = on_error
 
 	local parallel = options and options.parallel
@@ -162,6 +166,10 @@ function M.parallel(fn, on_error)
 	return M.start(fn, { parallel = true }, on_error)
 end
 
+function M.parallel_group(group_id, fn, on_error)
+	return M.start(fn, { parallel = true, group_id = group_id }, on_error)
+end
+
 
 --- Stop a created flow before it has completed
 -- @param instance This can be either the returned value from
@@ -504,13 +512,31 @@ function M.until_flows(flows)
 	end
 
 	-- several coroutines
-	for k, v in pairs(flows) do
+	for _, instance in pairs(flows) do
 		M.until_true(function()
-			return coroutine.status(v.co) == "dead"
+			return coroutine.status(instance.co) == "dead"
 		end)
 	end
 end
 
+--- Wait until other flow coroutines with a specific group_id were finished
+-- @param group_id identifier of the flows group
+function M.until_group(group_id)
+	assert(group_id)
+	local group_id = ensure_hash(group_id)
+	local group = {}
+
+	for _, instance in pairs(instances) do
+		if instance.group_id == group_id then
+			table.insert(group, instance)
+		end
+	end
+
+	if #group > 0 then
+		M.until_flows(group)
+	end
+end
+
 function M.ray_cast()
 	print("flow.ray_cast() is deprecated. Use synchronous ray casts released in Defold 1.2.150 instead!")
 end

test/test_flow.lua

@@ -198,5 +198,31 @@ return function()
 			wait_seconds(0.2)
 			assert(flow_finished)
 		end)
+
+		it("should be possible to pause until flows with the specific group_id complete", function()
+			local flow_finished = false
+			local instance = flow.start(function()
+				flow.parallel_group("foo", function()
+					flow.delay(0.1)
+				end)
+
+				flow.parallel_group("foo", function()
+					flow.delay(0.2)
+				end)
+
+				flow.parallel_group("foo", function()
+					flow.delay(0.3)
+				end)
+
+				flow.until_group("foo")
+				flow_finished = true
+			end)
+
+			wait_seconds(0.2)
+			assert(not flow_finished)
+
+			wait_seconds(0.2)
+			assert(flow_finished)
+		end)
 	end)
 end