👾 Pipeline

A ordered pipeline for Luau. Register callbacks as phases, control execution order, run them sync or async, and cancel gracefully or abruptly at any point.

Installation

Install via Wally by adding Pipeline to your wally.toml:

[dependencies]
Pipeline = "sqzee1/pipeline@1.0.0"

Basic Usage

local Pipeline = require(path.to.Pipeline)

local pipeline = Pipeline.new(function()
    print("first")
end, function()
    print("second")
end)

pipeline:run()

Adding phases with hooks

pipeline:addPhase(function()
    print("does the work")
end, {
    OnEnter = function()
        print("about to start")
    end,
    OnExit = function()
        print("just finished")
    end,
})

Running async

pipeline:run(true) -- doesn't block the calling thread

Passing arguments

Anything after async in run is forwarded to every phase's OnEnter, Callback, and OnExit:

pipeline:run(false, player, 10)
-- every OnEnter/Callback/OnExit in this run receives (player, 10)

Global hooks

pipeline:onComplete(function(...)
    print("finished every phase")
end)

pipeline:onStopped(function(...)
    print("halted early via stop()")
end)

onComplete fires when every phase finishes normally; onStopped fires instead if stop() was called during the run. Note that onStopped — not onComplete — fires even when stop() was called during the last phase, so nothing was actually skipped.

Neither fires if a phase errors — see Errors.

Stopping

pipeline:stop() -- lets the current phase (and its OnExit) finish, then halts
pipeline:stop(true) -- cuts immediately, skips the rest of the current phase

Abruptness only escalates. Once a run has been stopped with stop(true), a later stop() or clear() without abrupt will not downgrade it back to a graceful stop.

Clearing

pipeline:clear() -- waits for the current phase to finish, then wipes all phases
pipeline:clear(true) -- cuts immediately, then wipes all phases

Methods

Method	Description
`Pipeline.new(...: Callback)`	Creates a pipeline. Any callbacks passed in are registered as phases, in order
`pipeline:addPhase(callback, options?)`	Registers a phase after the last one. `options?` = `{ OnEnter?, OnExit? }`
`pipeline:insertBefore(callback, target?, options?)`	Inserts a phase before `target`'s phase. Appends at the end if `target` isn't found or omitted
`pipeline:insertAfter(callback, target?, options?)`	Inserts a phase after `target`'s phase. Appends at the end if `target` isn't found or omitted
`pipeline:remove(target)`	Removes the phase registered with `target`
`pipeline:getCurrentPhase()`	Returns the phase currently running, or `nil`
`pipeline:isRunning()`	Returns whether the pipeline is currently executing
`pipeline:shuffle()`	Shuffles the registered phases into random order
`pipeline:run(async?, ...)`	Runs all phases in order, forwarding `...` to every `OnEnter`/`Callback`/`OnExit`. Blocks the caller unless `async` is `true`
`pipeline:stop(abrupt?)`	Halts execution. Waits for the current phase to finish unless `abrupt` is `true`. No-op if no run is in progress
`pipeline:clear(abrupt?)`	Stops (same `abrupt` rules as above) and removes every registered phase. Wipes immediately if no run is in progress
`pipeline:onComplete(callback?)`	Sets the callback fired when a run finishes every phase without being stopped
`pipeline:onStopped(callback?)`	Sets the callback fired when `stop()` halts a run before it finishes

target in insertBefore/insertAfter/remove is the callback function used when the phase was registered. If the same callback was registered more than once, only the first matching phase is targeted.

Editing the phase list from inside a running phase (addPhase, remove, insertBefore, insertAfter) is safe: the run iterates a snapshot taken when it started, so those edits take effect on the next run rather than shifting the phases still queued in this one.

getCurrentPhase() returns the pipeline's own phase table, not a copy. Treat it as read-only; mutating it corrupts the pipeline.

Calling run() on a pipeline that is already running warns and does nothing.

stop() only affects a run that is already in progress; calling it beforehand does not arm a stop for the next run().

Types

local Pipeline = require(path.to.Pipeline)

local p: Pipeline.Pipeline = Pipeline.new()

Exported: Pipeline, Phase, Options, Callback.

Options

addPhase's options table accepts:

pipeline:addPhase(callback, {
    OnEnter = function() end, -- runs right before the phase's callback
    OnExit = function() end,  -- runs right after the phase's callback
})

Both fields are optional.

Errors

If any OnEnter, Callback, or OnExit raises, the run stops there: no later phase runs, and neither onComplete nor onStopped fires.

The pipeline always resets its internal state first, so it is never left stuck as "running" and can be run again:

pipeline:addPhase(function()
    error("boom")
end)

local ok, err = pcall(function()
    pipeline:run()
end)

print(ok, err)              --> false, "ServerScriptService.Server:2: boom"
print(pipeline:isRunning()) --> false

The error is re-raised with level 0, so the message you get back is exactly the one the phase raised — position info comes from the original error() call, not from inside Pipeline.

A pending clear() still applies before the error is re-raised.

Where the error surfaces depends on how the run was started:

Sync (pipeline:run()) — re-raised out of run(), so pcall around the call catches it, as above.
Async (pipeline:run(true)) — run() has already returned by then, so there is no caller to catch it. The error surfaces as an unhandled error in the spawned thread. Wrap the phase body yourself if an async run must not fail loudly.

Made by sqzee1

pipeline

By @sqzee1