Takes jobs from a queue and runs them.
Add one to your supervision tree alongside a client:
children = [
{Zizq, name: MyApp.Zizq, url: "http://localhost:7890"},
{Zizq.Worker,
client: MyApp.Zizq,
queues: ["emails"],
concurrency: 25,
handler: &MyApp.handle_job/1}
]The handler
A one-argument function receiving a Zizq.Job. What it returns
decides what happens to the job:
| Returns | Outcome |
|---|---|
:ok or {:ok, term} | acknowledged as complete |
{:error, reason} | failed; retried per the backoff policy |
{:cancel, reason} | killed now, whatever retries remain |
{:snooze, milliseconds} | retried after that long, ignoring backoff |
{:snooze, %DateTime{}} | retried at that time, ignoring backoff |
| raises, exits, or is killed | failed, with the exception and stacktrace |
Snooze durations are milliseconds, as every duration in this
client is, so {:snooze, :timer.minutes(5)} says plainly what
{:snooze, 300_000} means. Note that if you are coming from Oban, Oban's
equivalent is in seconds; take care if porting jobs from Oban to Zizq.
Anything else is acknowledged as complete and logged as a warning
naming what was returned. Failing the job instead would be worse —
the handler most likely did its work and ended on the wrong value, e.g.
due to a Logger call, so failing would re-run a side effect that already
happened. Return :ok explicitly to keep the log quiet.
Isolation
Each job runs in its own supervised task. A handler that raises, exits, or is killed outright cannot take the worker down — the crash is reported to the server as a failure and the next job starts. A slow handler cannot block the stream from being read either.
Shutdown
On shutdown the worker stops taking new work, waits for running jobs
to finish, flushes outstanding acknowledgements, and only then closes
the connection — in that order, so the server keeps the in-flight
jobs until the acknowledgements arrive. Anything still running when
:drain_timeout expires is abandoned and redelivered later.
Summary
Functions
Start a new worker.
Functions
@spec start_link(keyword()) :: Supervisor.on_start()
Start a new worker.
Options
:handler- Required. What to run for each job: a function taking aZizq.Job, or aZizq.Routerto dispatch by type. A router is compiled to a function once, when the worker starts.:name(atom/0) - Name for this worker. Defaults to the module.:queues(list ofString.t/0) - Queues to take from. Empty means every queue. The default value is[].:concurrency(pos_integer/0) - How many jobs may run at once. The default value is10.:prefetch(pos_integer/0) - Maximum unacknowledged jobs the server will send. Defaults to twice:concurrency, so a replacement job is already waiting when one finishes rather than costing a round trip.:worker_id(String.t/0) - Identifies this worker in the server's logs. Assigned by the server if omitted.:drain_timeout(pos_integer/0) - Milliseconds to allow on shutdown for running jobs to finish and acknowledgements to be flushed.This is the whole budget, not a per-step one: stopping the worker will not take materially longer than this, so it can be set from whatever deadline a deployment or orchestrator imposes. Jobs still running when it expires are abandoned locally, and redelivered by the server upon worker disconnection.
The default value is
30000.