Getting Started

This repo is a read-only subtree split of the Symfony bundle from the enqueue monorepo (php-enqueue/enqueue-dev). Everything in the Enqueue\* namespaces, the enqueue: config tree, and all upstream docs apply as-is; only the vendor name on this mirror is different. Treat issues/PRs as upstream work, and pin the version you install.

Install and let Flex wire it up:

composer require micro-module/enqueue-bundle
# Flex adds ENQUEUE_DSN to .env and config/packages/enqueue.yaml

Minimal configuration — one transport, one client:
```
# config/packages/enqueue.yaml
enqueue:
    default:
        transport: '%env(ENQUEUE_DSN)%'   # amqp://guest:guest@localhost:5672/%2f
        client: ~
```
transport accepts a plain DSN string or a map (dsn, factory_class, factory_service); client: ~ enables the client layer (topic/command routing, processors, the ProducerInterface).

Declare broker topology once per environment, then consume:

bin/console enqueue:setup              # creates exchanges/queues for the router + processors
bin/console enqueue:consume -vvv       # long-running worker

Where to look first: bin/console list enqueue (the commands you actually have), bin/console debug:container enqueue (service ids per transport), and the upstream docs at php-enqueue.github.io.

First use case — fire-and-forget async work from a controller:

use Enqueue\Client\ProducerInterface;

public function checkout(ProducerInterface $producer): Response
{
    $producer->sendEvent('order.created', ['id' => 42]); // JSON-encoded for you
    return $this->json(['ok' => true]);
}

namespace App\Processor;

use Enqueue\Client\TopicSubscriberInterface;
use Interop\Queue\Context;
use Interop\Queue\Message;
use Interop\Queue\Processor;

class OrderCreatedProcessor implements Processor, TopicSubscriberInterface
{
    public function __construct(private MailerInterface $mailer) {}

    public function process(Message $message, Context $context)
    {
        $data = json_decode($message->getBody(), true, 512, JSON_THROW_ON_ERROR);
        // ... do the slow work
        return self::ACK;
    }

    public static function getSubscribedTopics(): array
    {
        return ['order.created'];
    }
}

With autoconfigure: true the interface is enough; otherwise tag the service with enqueue.processor and topic/command attributes.

Implementation Patterns

Events vs commands. sendEvent(topic, body) fans out to every processor subscribed to that topic (via the router exchange). sendCommand(command, body) goes to exactly one processor queue — use it when a message must be handled once. Pick per call site; the distinction drives your whole topology.

Processor DI and returns. Processors are ordinary services, so constructor-inject repositories, mailers, HTTP clients. Return self::ACK, self::REJECT, self::REQUEUE, or a Result (Enqueue\Consumption\Result::reject('reason')). Returning nothing is not "ack".

Consumption middleware. Everything interesting in a worker is an extension: limits (LimitConsumedMessagesExtension, LimitConsumerTimeExtension, MemoryLimitExtension), SignalExtension for graceful SIGTERM, DelayRedeliveredMessageExtension for backoff, and a services-resetter extension so Doctrine connections and Symfony services don't go stale. Tag your own with enqueue.consumption.extension plus a priority to run inside the pipeline; inspect what's registered with debug:container --tag=enqueue.consumption.extension.

Multiple transports. Add more keys under enqueue: (default, audit, …). Each gets its own context/client services; run a dedicated worker per transport for the non-default ones. Useful for isolating bulk/low-priority work from interactive work.

Async events instead of ad-hoc topics. With enqueue/async-event-dispatcher, a Symfony listener can be declared async and will be serialized into a message and executed inside the consumer — same API as EventDispatcher, different execution model.

RPC (enqueue/rpc, optional companion) when you need a reply: an RpcClient sends a message with a reply queue and blocks with a timeout. It's a separate package with its own server command; only reach for it when "fire a request and wait" is genuinely the interaction.

Testing. Use a filesystem or null transport in test (file:///tmp/enqueue, null://) so tests never touch the broker, and enable client: { traceable_producer: true } to record what was produced (getTracedEvents() / getTracedCommands()), so you can assert on messages rather than on side effects.

Long-running workers in production. Supervisor or systemd, --memory-limit as a backstop, and a restart on every deploy. The consumer runs a snapshot of the container; new code does not appear until restart.

Gotchas and Tips

enqueue:setup is not optional. On AMQP, publishing to an undeclared exchange fails or silently drops; on SQS/SNS, setup creates topics/queues and may require IAM permissions. Re-run after changing processor subscriptions, and wire --setup into deploy (and into consumer startup in CI/dev).
The router queue must be consumed. If you pass explicit --queue flags to enqueue:consume, you can leave out the router queue and watch events arrive at the exchange and never reach a processor. Symptom: commands work, events don't.
Two message classes. You produce with Enqueue\Client\Message, but process() receives Interop\Queue\Message — getBody() returns the JSON string, not your array. Decode explicitly, or you'll debug "why is my array a string".
At-least-once, not exactly-once. Ack after the work, make processors idempotent, and use delayed redelivery instead of immediate REQUEUE hot loops. An uncaught exception is not a retry policy — handle it and decide.
Publishing inside a Doctrine transaction. The message is sent before commit; the consumer can read a row that doesn't exist yet. Send in postFlush/an outbox, or accept the race knowingly.
Payload limits and content types. SQS caps at 256 KB; large payloads belong in object storage with a pointer in the message. Set contentType when the body isn't JSON-encoded by the client.
Stale consumers on deploy. Classic "fixed it but the bug persists" — the old worker is still running. Restart workers in the deploy hook and use the signal extension for graceful shutdown.
Autoconfigure is required for interface-based auto-tagging. If your processor only implements TopicSubscriberInterface but its service is defined explicitly without autoconfigure, it will be invisible. Verify with debug:container --tag=enqueue.processor.
Extension points: a custom broker via a TransportFactoryInterface (enqueue.transport.factory tag), a custom routing layer via a client driver factory (enqueue.client.driver.factory tag), and custom middleware via enqueue.consumption.extension. Everything else is configuration.
Debugging: bin/console enqueue:produce to inject a message without the app, -vvv on the worker to see receipt/ack, and broker-side tools (rabbitmqctl list_queues) to confirm the topology actually exists.
Project-level caveat: enqueue development has slowed and the client/router layer is a heavier abstraction than Symfony Messenger. For greenfield async work, compare against Messenger before committing; this bundle earns its keep when you need multi-broker portability (RabbitMQ / SQS / Redis / AMQP / filesystem) behind one API, or the event/command + RPC semantics it provides out of the box.

Enqueue Bundle Laravel Package

Getting Started

Implementation Patterns

Gotchas and Tips