refactor parallel code

milo/plugins/importTask.lua

@@ -1,7 +1,6 @@
 local itemDB = require('core.itemDB')
 local Milo   = require('milo')
-
-local parallel = _G.parallel
+local Tasks  = require('milo.taskRunner')
 
 local ImportTask = {
 	name = 'importer',
@@ -13,73 +12,67 @@ local function filter(a)
 end
 
 function ImportTask:cycle(context)
-	local tasks = { }
+	local tasks = Tasks({
+		errorMsg = 'IMPORT error: '
+	})
 
 	for node in context.storage:filterActive('machine', filter) do
-		table.insert(tasks, function()
-			local s, m = pcall(function()
-				for _, entry in pairs(node.imports) do
+		tasks:add(function()
+			for _, entry in pairs(node.imports) do
 
-					local function itemMatchesFilter(item)
-						if not entry.ignoreDamage and not entry.ignoreNbtHash then
-							local key = itemDB:makeKey(item)
-							return entry.filter[key]
-						end
-
-						for key in pairs(entry.filter) do
-							local v = itemDB:splitKey(key)
-							if item.name == v.name and
-								(entry.ignoreDamage or item.damage == v.damage) and
-								(entry.ignoreNbtHash or item.nbtHash == v.nbtHash) then
-								return true
-							end
-						end
+				local function itemMatchesFilter(item)
+					if not entry.ignoreDamage and not entry.ignoreNbtHash then
+						local key = itemDB:makeKey(item)
+						return entry.filter[key]
 					end
 
-					local function matchesFilter(item)
-						if not entry.filter then
+					for key in pairs(entry.filter) do
+						local v = itemDB:splitKey(key)
+						if item.name == v.name and
+							(entry.ignoreDamage or item.damage == v.damage) and
+							(entry.ignoreNbtHash or item.nbtHash == v.nbtHash) then
 							return true
 						end
-
-						if entry.blacklist then
-							return not itemMatchesFilter(item)
-						end
-
-						return itemMatchesFilter(item)
-					end
-
-					local list = node.adapter.list()
-
-					local function importSlot(slotNo)
-						local item = itemDB:get(list[slotNo], function()
-							return node.adapter.getItemMeta(slotNo)
-						end)
-						if item and matchesFilter(item) then
-							context.storage:import(node, slotNo, item.count, item)
-						end
-					end
-
-					if type(entry.slot) == 'number' then
-						if list[entry.slot] then
-							importSlot(entry.slot)
-						end
-					else
-						for i in pairs(list) do
-							importSlot(i)
-						end
 					end
 				end
-			end)
 
-			if not s and m then
-				_G._debug('IMPORTER error: ' .. m)
+				local function matchesFilter(item)
+					if not entry.filter then
+						return true
+					end
+
+					if entry.blacklist then
+						return not itemMatchesFilter(item)
+					end
+
+					return itemMatchesFilter(item)
+				end
+
+				local list = node.adapter.list()
+
+				local function importSlot(slotNo)
+					local item = itemDB:get(list[slotNo], function()
+						return node.adapter.getItemMeta(slotNo)
+					end)
+					if item and matchesFilter(item) then
+						context.storage:import(node, slotNo, item.count, item)
+					end
+				end
+
+				if type(entry.slot) == 'number' then
+					if list[entry.slot] then
+						importSlot(entry.slot)
+					end
+				else
+					for i in pairs(list) do
+						importSlot(i)
+					end
+				end
 			end
 		end)
 	end
 
-	if #tasks > 0 then
-		parallel.waitForAll(table.unpack(tasks))
-	end
+	tasks:run()
 end
 
 Milo:registerTask(ImportTask)