Injecting resources

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.

Scopes

A resource's ResourceScope decides its lifetime:

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 committed

Worker-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.

Pooled resources

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.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.

Injecting resources

Two equivalent ways to reach a resource from a handler.

Accessor — Resources.use

Call 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.

Declarative — @Resource parameters

With 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.

Resources that depend on resources

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.

Teardown

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.

Metrics

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.

Testing

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.