Rsqueue Bundle Laravel Package

Getting Started

Install and register. The bundle targets the Symfony 2.8 / 3.x era, so pin it accordingly (composer require mmoreram/rsqueue-bundle), then register new Mmoreram\RSQueueBundle\RSQueueBundle() in AppKernel. You also need a Redis instance reachable from every producer and worker.

Configure a queue. The bundle's DependencyInjection/Configuration.php is the source of truth for the rs_queue tree; a typical declaration is:

# app/config/config.yml
rs_queue:
    host: 127.0.0.1
    port: 6379
    database: 0
    queues:
        thumbnails:
            type: queue            # queue | priority_queue | exchange | rpc
            serializer: default    # default | json

A named queue in this block becomes a container service, so you typically declare one entry per job type rather than one global queue.

First use case — fire-and-forget work. Start with a plain queue for slow, non-critical side effects (image thumbnails, outbound email, search reindexing). Produce from a controller or service, consume from a long-running console command, and run that command under Supervisor. That is the 80% path; the priority/RPC/exchange variants exist for the other 20%.

Where to look first.

1. README.md for the canonical produce/consume snippets.
2. DependencyInjection/Configuration.php — exact config keys, defaults, and which keys apply per queue type.
3. Resources/config/*.yml — real service IDs (the README's IDs lag behind the code).
4. Command/ — the base consumer/producer/admin command classes you will extend.
5. The underlying mmoreram/rsqueue library for Redis-level semantics (list vs. sorted-set vs. pub/sub).

Smoke test without writing code. The bundle ships producer/admin commands; use them to push a test payload and watch it arrive, before wiring any application code.

Implementation Patterns

Producing from application code. Obtain the queue from the bundle's registry/manager service and call produce() with an array payload:

$queue = $this->getContainer()->get('rsqueue')->getQueue('thumbnails');
$queue->produce(['image' => $path, 'width' => 200]);

Wrap this in a thin domain service (ThumbnailDispatcher) instead of calling getContainer() from controllers. It keeps queue names in one place and gives you a seam to swap in a fake during tests.

Consuming by extending the base command. The intended pattern is one command class per queue type:

class ThumbnailConsumerCommand extends ConsumerCommand
{
    protected function configure()
    {
        $this->setName('app:consume:thumbnails');
        parent::configure();   // adds timeout/interval/verbosity options
    }

    protected function getQueue()
    {
        return 'thumbnails';
    }

    protected function process(OutputInterface $output, $payload)
    {
        // $payload is the array you produced
        $output->writeln('Processing ' . $payload['image']);
    }
}

Put the real work in a service the command delegates to; the command should only handle deserialization, logging, and exceptions.

Pick the queue type by semantics, not by preference.

Scale with one worker process per queue. Each app:consume:* command is a single consumer loop; run several instances of the same command to parallelize a queue, and separate commands for separate queues. This is the simplest scaling model the bundle offers and the one most teams stop at.

Design handlers to be idempotent. Delivery is at-least-once: a worker can pop a job, crash, and have it re-queued by the recover command. Make process() safe to run twice (upsert, dedupe keys, idempotency tokens).

Recover stuck jobs on a schedule. Anything popped but not completed sits in an "in progress" structure. Run the bundle's recover command from cron (every few minutes) so crashed workers don't strand jobs. Pair it with --timeout on consumers so a hung job eventually becomes recoverable rather than pinning a worker forever.

Compose with Symfony lifecycle. For work that must not block the response but should be queued immediately, dispatch on kernel.terminate in Symfony 2/3 rather than inside the controller action.

Cross-language or multi-service setups. Switch serializer to json and keep the payload a flat array; the default PHP serializer ties you to PHP on both ends and to the same class definitions.

Gotchas and Tips

• Maintenance reality. Last release 2016, zero dependents. Expect PHP 5.6/7.x and Symfony 2.8/3.x to be the sweet spot. PHP 8 and Symfony 4+ will require a fork (service visibility, getContainer() removal, Predis version conflicts). Budget for vendoring or porting before committing.
• Service IDs drift. README examples are older than the code. Always confirm IDs in Resources/config/ and the config keys in Configuration.php before writing integration code.
• Serializer mismatch is silent data loss. Producer and consumer must agree on serializer and payload shape. Rotating serializers leaves in-flight jobs unreadable — drain the queue first, or version payloads with a v field.
• exchange is not a job queue. Pub/sub drops messages when no subscriber is connected and has no recovery. Never route work that must complete through it.
• Redis is the queue. Point the bundle at a dedicated Redis database/instance, not a cache you FLUSHDB casually. Decide persistence (appendonly yes) deliberately; a restart without it loses queued jobs.
• Unbounded growth. Nothing trims a queue that fills faster than it drains. Alert on LLEN / ZCARD per queue key and on consumer lag (oldest job age).
• Long-running worker staleness. Like any daemonized Symfony command from this era, the container and any injected entity managers are captured at boot. Recycle workers periodically (Supervisor max_time/stopwaitsecs, or a bounded loop in execute()) rather than trusting them to run for weeks.
• Tune --timeout and --interval. A blocking pop with no timeout prevents graceful shutdown; a zero interval with an empty queue hammers Redis. Defaults are usually wrong for production.
• rpc blocks a PHP process on both sides. Without a timeout the producer hangs indefinitely if no consumer is up. Reserve it for genuinely synchronous calls; prefer async + a result key for anything slow.
• Debug with redis-cli, not the application. MONITOR to see pushes/pops, LRANGE <key> 0 -1 for list queues, ZRANGE <key> 0 -1 WITHSCORES for priority queues. This tells you instantly whether the producer or the consumer is the broken half.
• Namespace your keys per environment. Queue names map directly to Redis keys; without a prefix, staging workers will happily eat production jobs.
• Extension points worth using. A custom serializer class (implement the library's serializer interface) for encryption or schema validation; a shared abstract consumer base for metrics/tracing around process(); decorating the queue service to add logging on produce(). These are cheap and keep the bundle's surface area small.
• Test the producer, not the worker, in unit tests. Assert produce() was called with the right payload via a fake queue; test process() directly with a payload array. Spin up real Redis only for an integration suite, since the loop is hard to assert on.