Flux: event-driven CC

Flux.lua

--[[
    Flux by AlexDevs (c) 2025

    This software is licensed under the MIT license.
]] --

---Assert a value type of arguments
---@private
---@param value any Value to check against.
---@param index number|string Position of the argument in the function. Starts at 1.
---@param ... string Types allowed.
---@return any value
local function expect(value, index, ...)
    local valType = type(value)
    local expected = { ... }

    for _, expType in ipairs(expected) do
        if valType == expType then
            return value
        end
    end

    error(string.format("bad argument #%s (expected %s, got %s)", index, table.concat(expected, ", "), valType), 3)
end

---@alias Task table

---@class FluxOptions
---@field localEventsOnly boolean? Whether to only emit local events (signals via `emit()`) to event listeners (`on`, `once`). Defaults to false.

--- Create a new Flux event instance
---@param options FluxOptions?
---@return Flux Flux instance
local function create(options)
    options = options or {}
    local Flux = {}

    local tasks = {}
    local running = false
    local signalQueue = {}
    local localEventsOnly = options.localEventsOnly

    ---@private
    ---Create a task waiting to run
    ---@param func function Task function
    ---@param isListener boolean? Whether it's an event listener
    ---@param ... any? Task init arguments
    ---@return Task Task Newly created task
    local function spawnTask(func, isListener, ...)
        local pid = #tasks + 1

        local task = {
            pid = pid,
            thread = coroutine.create(func),
            args = table.pack(...),
            initialized = false,
            filter = nil,
            isListener = isListener or false
        }

        tasks[pid] = task

        return task
    end

    ---@private
    local function killTask(pid)
        expect(pid, 1, "number")

        local task = tasks[pid]
        tasks[pid] = nil

        return task ~= nil
    end

    ---@private
    local function awaitEvent(eventName, callback)
        return spawnTask(function()
            callback(select(2, coroutine.yield(eventName)))
        end, true)
    end

    ---@private
    local function createListener(eventName, callback)
        while true do
            local event = table.pack(select(2, coroutine.yield(eventName)))
            spawnTask(callback, false, table.unpack(event))
        end
    end

    ---@private
    local function delayExec(callback, delay, ...)
        sleep(delay)
        spawnTask(callback, false, ...)
    end

    ---@private
    local function resumeTask(task, isSignal, ...)
        if localEventsOnly and task.isListener and not isSignal then
            return
        end
        local ok, par = coroutine.resume(task.thread, ...)
        if ok then
            task.filter = par
        else
            local message = string.format("%s\n%s", tostring(par), debug.traceback(task.thread))
            error(message, 0)
        end
    end

    ---@private
    local function resume(isSignal, eventName, ...)
        for _, task in pairs(tasks) do
            if task.initialized then
                if task.filter == nil or task.filter == eventName then
                    resumeTask(task, isSignal, eventName, ...)
                end
            else
                task.initialized = true
                resumeTask(task, true, table.unpack(task.args))
            end

            if coroutine.status(task.thread) == "dead" then
                tasks[task.pid] = nil
            end
        end
    end

    ---@private
    local function countTasks()
        local count = 0
        for _ in pairs(tasks) do
            count = count + 1
        end
        return count
    end

    ---@private
    local function signal(...)
        table.insert(signalQueue, table.pack(...))
    end

    ---@private
    local function intervalExec(callback, delay, ...)
        while true do
            delayExec(callback, delay, ...)
        end
    end

    ---Start the loop event. This function exits once the task list is empty.
    ---@param pullEvent function Event polling function. Uses `os.pullEvent` by default, use `os.pullEventRaw` to handle `terminate` events.
    function Flux.run(pullEvent)
        pullEvent = pullEvent or os.pullEvent
        expect(pullEvent, 1, "function")

        if running then
            error("already running", 2)
        end

        running = true

        local eventPars = { n = 0 }
        while running do
            resume(false, table.unpack(eventPars))

            for i = 1, #signalQueue do
                local sig = table.remove(signalQueue, 1)
                resume(true, table.unpack(sig))
            end

            -- process new signals in the next cycle
            if #signalQueue > 0 then
                os.queueEvent("flux_signal")
            end

            if countTasks() == 0 then
                break
            end

            eventPars = table.pack(pullEvent())
        end

        running = false
    end

    ---Stop the execution of the loop event
    ---@param graceful boolean Signal the `interrupt` event listener. Immediately stops if false.
    function Flux.stop(graceful)
        -- nil counts as true
        if graceful == false then
            running = false
            return
        end

        spawnTask(function()
            for pid, task in pairs(tasks) do
                if task.filter ~= "interrupt" then
                    tasks[pid] = nil
                end
            end
        end)

        signal("interrupt")
    end

    ---Checks whether the loop is running
    ---@return boolean running Loop is running
    function Flux.isRunning()
        return running
    end

    ---Listen to an event
    ---@param eventName string Name of the event
    ---@param callback function
    ---@return Task Task
    function Flux.on(eventName, callback)
        expect(eventName, 1, "string")
        expect(callback, 2, "function")

        return spawnTask(createListener, true, eventName, callback)
    end

    ---Listen to an event a single time
    ---@param eventName string Name of the event
    ---@param callback function
    ---@return Task Task
    function Flux.once(eventName, callback)
        return awaitEvent(eventName, callback)
    end

    ---Emit a local signal to all tasks.
    ---@param eventName string Name of the signal event
    ---@param ... any? Parameters
    function Flux.emit(eventName, ...)
        expect(eventName, 1, "string")
        signal(eventName, ...)
    end

    ---Used to clear event listeners, intervals and timeouts, but can also kill tasks.
    ---@param task Task Task to kill
    ---@return boolean success Whether it successfully killed the task
    function Flux.clear(task)
        expect(task, 1, "table")
        expect(task.pid, "pid", "number")
        return killTask(task.pid)
    end

    ---Start a loop that calls `callback` every `delay` seconds. Waits `delay` seconds before the first call.
    ---@async
    ---@param callback function Callback
    ---@param delay number Seconds to delay between calls.
    ---@param ... any? Callback arguments
    ---@return Task Task
    function Flux.interval(callback, delay, ...)
        expect(callback, 1, "function")
        expect(delay, 2, "number")

        return spawnTask(intervalExec, false, callback, delay, ...)
    end

    ---Wait `delay` seconds before calling `callback`.
    ---@param callback function Callback
    ---@param delay number Seconds to wait
    ---@param ... any? Callback arguments
    ---@return Task Task
    function Flux.timeout(callback, delay, ...)
        expect(callback, 1, "function")
        expect(delay, 2, "number")

        return spawnTask(delayExec, false, callback, delay, ...)
    end

    ---Spawn a background task. This is effectively multitasking.
    ---@param func function Task function
    ---@param ... any? Function arguments
    ---@return Task Task
    function Flux.spawnTask(func, ...)
        expect(func, 1, "function")
        return spawnTask(func, false, ...)
    end

    Flux.debug = {}

    ---Dump in terminal the state of tasks
    function Flux.debug.dumpTasks()
        textutils.tabulate({ "PID", "FILTER", "INIT?", "TASK STATUS", "EV LSTNR" })
        for pid, task in pairs(tasks) do
            textutils.tabulate(colors.lightGray,
                { task.pid, task.filter or "*", tostring(task.initialized), coroutine.status(task.thread), tostring(task
                .isListener) })
        end
    end

    return Flux
end

return create

FluxMeta.lua

--[[
    This is a definition file for Flux
]]

---@meta Flux

---@class Flux
local Flux = {
    debug = {}
}
---Start the loop event. This function exits once the task list is empty.
---@param pullEvent function Event polling function. Uses `os.pullEvent` by default, use `os.pullEventRaw` to handle `terminate` events.
function Flux.run(pullEvent)
end

---Stop the execution of the loop event
---@param graceful boolean Signal the `interrupt` event listener. Immediately stops if false.
function Flux.stop(graceful)
end

---Checks whether the loop is running
---@return boolean running Loop is running
function Flux.isRunning()
end

---Listen to an event
---@param eventName string Name of the event
---@param callback function
---@return Task Task
function Flux.on(eventName, callback)
end

---Listen to an event a single time
---@param eventName string Name of the event
---@param callback function
---@return Task Task
function Flux.once(eventName, callback)
end

---Emit a local signal to all tasks.
---@param eventName string Name of the signal event
---@param ... any? Parameters
function Flux.emit(eventName, ...)
end

---Used to clear event listeners, intervals and timeouts, but can also kill tasks.
---@param task Task Task to kill
---@return boolean success Whether it successfully killed the task
function Flux.clear(task)
end

---Start a loop that calls `callback` every `delay` seconds. Waits `delay` seconds before the first call.
---@async
---@param callback function Callback
---@param delay number Seconds to delay between calls.
---@param ... any? Callback arguments
---@return Task Task
function Flux.interval(callback, delay, ...)
end

---Wait `delay` seconds before calling `callback`.
---@param callback function Callback
---@param delay number Seconds to wait
---@param ... any? Callback arguments
---@return Task Task
function Flux.timeout(callback, delay, ...)
end

---Spawn a background task. This is effectively multitasking.
---@param func function Task function
---@param ... any? Function arguments
---@return Task Task
function Flux.spawnTask(func, ...)
end

---Dump in terminal the state of tasks
function Flux.debug.dumpTasks()
end