/ 
PAGI-FastAPI-Queue-Driver-Redis-v0.0.1
/ PAGI::FastAPI::Queue::Driver::Redis

NAME

PAGI::FastAPI::Queue::Driver::Redis - Redis Storage Driver for PAGI::FastAPI::Queue

VERSION

Version v0.0.1

SYNOPSIS

use PAGI::FastAPI::Queue;
use PAGI::FastAPI::Depends qw(Depends);

my $queue = PAGI::FastAPI::Queue->new(
    driver  => 'PAGI::FastAPI::Queue::Driver::Redis',
    options => {
        host   => '127.0.0.1',
        port   => 6379,
        prefix => 'myapp:queue:',
    },
);

$app->post('/jobs',
    dependencies => [ Depends($queue->dep, key => 'queue') ],
    handler      => async sub ($c) {
        await $c->stash->{queue}->push('emails', { to => 'a@b.com' });
        return { queued => 1 };
    },
);

# Release the connection on shutdown (optional; not part of the
# PAGI::FastAPI::Queue::Driver contract, so reach the driver directly)
$app->on_shutdown(async sub {
    # e.g. if you kept a reference to the driver instance yourself
    # $driver->disconnect;
});

DESCRIPTION

PAGI::FastAPI::Queue::Driver::Redis is a storage driver plugin for PAGI::FastAPI::Queue. It stores each topic as a Redis list (RPUSH to enqueue, LPOP to dequeue), so queued items are shared across every worker process and host pointed at the same Redis instance and prefix, unlike the built-in PAGI::FastAPI::Queue::Driver::Memory, which is scoped to a single process.

Payloads (any Perl scalar - including hashrefs/arrayrefs) are serialised with JSON::PP before being stored, and deserialised on pop, so the round-trip behaviour matches PAGI::FastAPI::Queue::Driver::Memory for any JSON-representable value.

Why Async::Redis and not Net::Async::Redis?

PAGI::FastAPI documents Future::IO as the event-loop-agnostic target for anything loop-driven, with IO::Async (used by the reference PAGI::Server) treated as an implementation detail applications shouldn't depend on directly (see "EVENT LOOPS: FUTURE::IO IS THE GOAL, IO::ASYNC IS AN IMPLEMENTATION DETAIL" in PAGI::FastAPI). Async::Redis is built directly on Future::IO rather than tying itself to IO::Async, so this driver keeps working regardless of which backend a given PAGI server ends up using, with no reactor-bridging required.

Aggregate size() and SCAN

Unlike PAGI::FastAPI::Queue::Driver::Memory, Redis has no O(1) way to enumerate "every list under this driver". size() with no $topic therefore walks the keyspace with Redis's cursor-based SCAN command (never the blocking KEYS command), matching on "$prefix*", and sums LLEN across whatever keys it finds. This is a best-effort snapshot, not an atomic one: concurrent pushes/pops elsewhere during the scan can make the total slightly stale. Prefer size($topic) when you only care about one topic; it's a single LLEN call.

METHODS

Inherits push, pop, and size from PAGI::FastAPI::Queue::Driver.

new(%options)

  • redis - (Optional) A pre-built, Async::Redis-compatible client instance to use instead of constructing one. Useful for sharing a single connection across multiple drivers, for supplying your own pre-configured client (TLS, auth, custom timeouts), or for injecting a test double. When given, host/port/uri/username/password/ database are ignored.

  • host - (Optional) Redis host. Defaults to '127.0.0.1'.

  • port - (Optional) Redis port. Defaults to 6379.

  • uri - (Optional) A full redis:// connection URI, forwarded to Async::Redis->new as-is. When given together with host/ port, uri takes precedence in Async::Redis itself.

  • username / password - (Optional) Redis 6+ ACL / AUTH credentials.

  • database - (Optional) Redis logical database index to SELECT.

  • prefix - (Optional) Key namespace prefix. Defaults to 'pagi:queue:'. Each topic is stored under "$prefix$topic". Use a distinct prefix per application when sharing one Redis instance across multiple apps.

  • scan_count - (Optional) COUNT hint passed to each SCAN call made by aggregate size(). Defaults to 100. Higher values mean fewer round-trips but larger batches per call.

The underlying client connects lazily: no network I/O happens in new() or ADJUST, only on the first push/pop/size call, and only once per driver instance.

disconnect()

Not part of the PAGI::FastAPI::Queue::Driver contract and not exposed through PAGI::FastAPI::Queue. Closes the underlying client connection. Call it directly on the driver instance (not through PAGI::FastAPI::Queue, which does not expose its backend) from an $app->on_shutdown hook if you want a clean shutdown. Synchronous (does not return a Future). Safe to skip; the OS reclaims the socket on process exit regardless.

DEPENDENCIES

Requires Async::Redis (not installed automatically as a hard PREREQ_PM dependency solely because redis can also be supplied pre-built by the caller; if you don't pass redis, install Async::Redis yourself).

SEE ALSO

PAGI::FastAPI::Queue, PAGI::FastAPI::Queue::Driver, PAGI::FastAPI::Queue::Driver::Memory, Async::Redis

AUTHOR

Mohammad Sajid Anwar, <mohammad.anwar at yahoo.com>

REPOSITORY

https://github.com/manwar/PAGI-FastAPI-Queue-Driver-Redis

BUGS

Please report any bugs or feature requests through the web interface at https://github.com/manwar/PAGI-FastAPI-Queue-Driver-Redis/issues. I will be notified and then you'll automatically be notified of progress on your bug as I make changes.

SUPPORT

You can find documentation for this module with the perldoc command.

perldoc PAGI::FastAPI::Queue::Driver::Redis

You can also look for information at:

LICENSE AND COPYRIGHT

Copyright (C) 2026 Mohammad Sajid Anwar.

This program is free software; you can redistribute it and/or modify it under the terms of the Artistic License (2.0).