Dependency Injection
Register external dependencies once and inject them into tasks by scope.
Register external dependencies once and inject them into tasks by scope.
Tasks often need a database pool, an HTTP client, or a cloud SDK. Rather than constructing those inside every handler, register them once as resources and let the worker build, share, and tear them down with the right lifetime.
try (FlexiQ queue = FlexiQ.builder().sqlite("tasks.db").open()) {
queue.resource("db", ResourceScope.WORKER,
ctx -> openPool(System.getenv("DATABASE_URL")),
pool -> pool.close());
try (Worker worker = queue.worker()
.handle(SYNC_USER, userId -> {
DataSource db = Resources.use("db");
return markSynced(db, userId);
})
.start()) {
queue.enqueue(SYNC_USER, "u_123");
}
}The pool is created once when the worker first needs it, shared across every job, and closed when the worker stops. No connection is ever serialized into the queue.
A resource's ResourceScope decides its lifetime:
| Scope | Built | Lifetime | Use for |
|---|---|---|---|
WORKER (default) | Lazily, on first use | The worker — a shared singleton | Connection pools, HTTP clients, SDK clients |
THREAD | Lazily, once per worker thread | Until worker shutdown | Non-thread-safe clients that are cheap to keep per thread |
TASK | Once per task invocation | That invocation — disposed when it ends | Per-job transactions, request-scoped clients |
REQUEST | Fresh on every use() | Disposed with the task; never cached | Values that must not be shared even within one job |
POOLED | Lazily, up to poolSize at once | Checked out per task, returned at task end | Expensive clients that must stay bounded but be reused |
Scope names and their meanings are shared across SDKs. Two are platform-bound:
THREAD has no Node equivalent (a Node worker runs its tasks on one event loop),
and REQUEST has no Python equivalent (Python resources arrive by injection,
resolved once per task, so there is no second resolve to build for).
queue.resource("tx", ResourceScope.TASK,
ctx -> beginTransaction(),
tx -> tx.rollbackIfOpen()); // no-op if already committedWorker-scoped resources are built at most once per worker even under
concurrency — a concurrent first use does not double-build. A factory that
throws is not cached, so the next job retries it. Registering the same name
twice throws ResourceException, as does a dependency cycle between factories.
A POOLED resource sits between WORKER and TASK: instances are reused like
worker resources but bound to a single task at a time, with a hard cap on how
many exist at once. Each task checks out one instance — reusing an idle one,
building a new one only when none is free — and returns it to the pool when the
task finishes. A task that cannot get an instance within acquireTimeout fails
with ResourceException:
queue.resource("ftp",
new PoolConfig(4, 1, Duration.ofSeconds(10), Duration.ofMinutes(5)),
ctx -> connectFtp(),
conn -> conn.close());PoolConfig field | Default | Meaning |
|---|---|---|
poolSize | required | Max instances checked out concurrently. Tasks wait when exhausted. |
poolMin | 0 | Instances prewarmed when the worker starts. 0 means fully lazy. |
acquireTimeout | 10s | How long a checkout waits before failing the task. |
maxLifetime | unlimited | An idle instance older than this is disposed and rebuilt instead of reused. |
PoolConfig.of(poolSize) gives the defaults; derive variations with
withPoolMin, withAcquireTimeout, and withMaxLifetime. The disposer runs
when the pool retires an instance (worker shutdown or maxLifetime expiry) —
not when a task returns it. Pool capacity is per worker, never shared across
workers.
Two equivalent ways to reach a resource from a handler.
Resources.useCall Resources.use(name) anywhere inside a running handler. It resolves
against the current task's scope:
queue.worker().handle(REPORT, request -> {
DataSource db = Resources.use("db");
Cache cache = Resources.use("cache");
// ...
return buildReport(db, cache, request);
});Calling Resources.use outside a task throws ResourceException — it is only
available while a handler runs.
@Resource parametersWith the annotation processor, list resources as extra parameters on a
@TaskHandler method. The generated companion resolves each by name and passes
it positionally — no runtime reflection:
public class EmailHandlers {
@TaskHandler("send_email")
void send(EmailPayload payload, @Resource("db") Database db) {
// db is injected, not part of the payload
}
}
// The processor generates EmailHandlersTasks with a typed Task constant
// per handler and a bind method:
Worker worker = queue.worker()
.apply(builder -> EmailHandlersTasks.bind(builder, new EmailHandlers()))
.start();
// Producers still enqueue only the real payload:
queue.enqueue(EmailHandlersTasks.SEND_EMAIL, payload);@Resource parameters must come after the payload parameter; each name is
resolved from the worker's resource runtime, exactly like Resources.use.
A factory receives a ResourceContext whose use resolves another resource,
so you can compose them:
queue.resource("config", ctx -> loadConfig());
queue.resource("db", ctx -> {
Config config = ctx.use("config");
return openPool(config.databaseUrl());
});A factory may only depend on same-or-longer-lived resources: WORKER and
POOLED factories may use only worker resources (a pooled instance outlives
the task that built it), a THREAD factory may use worker or thread resources,
and TASK/REQUEST factories may use any scope. Reaching for anything
shorter-lived throws ResourceException.
Pass a disposer to release a resource when its scope ends — worker and thread resources when the worker stops, task and request resources when the task finishes. Disposal runs in reverse order of construction (LIFO), so a resource is always torn down before anything it depended on:
queue.resource("db", ResourceScope.WORKER, ctx -> openPool(), pool -> pool.close());Disposal errors are logged, never thrown — they cannot fail an already-settled job.
queue.resourceMetrics() returns per-resource lifecycle counters — how many
instances were built, disposed, and are currently live:
Map<String, ResourceStat> metrics = queue.resourceMetrics();
metrics.get("db"); // ResourceStat[created=1, disposed=0, active=1]
metrics.get("tx"); // ResourceStat[created=12, disposed=12, active=0]A worker-scoped resource shows active = 1 while the worker runs and 0 after
it closes; a task-scoped resource's created/disposed climb per job with
active near zero.
Register a stub factory to swap a real dependency in tests — handlers resolve
by name, so they never know the difference. The test-support module's
InMemoryFlexiQ gives you a queue with no native backend at all:
try (FlexiQ queue = InMemoryFlexiQ.open()) {
AtomicInteger built = new AtomicInteger();
queue.resource("db", ctx -> {
built.incrementAndGet();
return new FakeDatabase();
});
// ... run the task through a worker ...
assertEquals(1, built.get());
assertEquals(1, queue.resourceMetrics().get("db").created());
}See testing for the full worker test setup.