README

A terminal UI to drive Symfony Messenger workers while you develop: list the transports of your application, start and scale their workers, watch messages flow in real time, and send failed messages back once fixed.

Built on top of the Symfony TUI component.

If it saves you time, you can sponsor my open source work.

Warning

This is a development tool, and the Symfony TUI component it relies on is experimental. Do not enable the bundle in production. It is tested with Symfony 8.1, and with the development version of Symfony 8.2.

Features

Transports in a sidebar, with their running workers and waiting messages. Open a transport to reach its actions:
- Message classes routed to it, from the routing configuration and the #[AsMessage] attribute of handled messages;
- Start workers, as many as you want (one by default);
- Concurrency: how many messages each worker handles at the same time, with Symfony 8.2;
- Options: DSN, transport options, retry strategy, failure transport, serializer. Credentials are masked.
Live view of the selected transport:
- incoming messages per second, as a sparkline;
- each worker, with the messages it handles and for how long, or why it exited;
- the messages waiting in the transport, when it can tell;
- the last handled, retried and failed messages, with their error.
Failed messages: workers do not consume failure transports. Open one to list its messages, then edit a message and send it back to the transport it failed on.
Output of the workers: what they write, like logs and dump(), with the time and the worker.
Stats in the footer: processed, failed and retried messages, throughput, memory used by the workers and their peak.

Requirements

PHP 8.4.1 or higher; with the pcntl extension, workers finish their current messages when stopped
Symfony 8.1 or higher, with Messenger enabled; Symfony 8.2 and amphp/parallel to handle several messages at the same time in each worker
A Unix-like system (Linux, macOS, WSL)

Installation

composer require --dev welcomattic/messenger-central

With Symfony Flex, the bundle is enabled by itself in the dev and test environments. To enable it in dev only, or without Flex, set it in config/bundles.php:

return [
    // ...
    Welcomattic\MessengerCentral\MessengerCentralBundle::class => ['dev' => true],
];

The command refuses to run when the kernel runs without debug, like in production; --force runs it anyway.

Usage

bin/console messenger:central

Key	Action
`↑` `↓`	Move in the sidebar
`⏎`	Open a transport, or run the action under the cursor
`←`	Close the opened transport
`s`	Start or stop the workers of the selected transport
`+` `-`	Scale the workers up or down (before or after starting)
`c`	Type the concurrency of the workers (Symfony 8.2)
`a`, `Esc`	Live view
`l`	Output of the workers: logs, dumps, errors
`m`	Message classes
`o`	Options
`?`	Every key
`q`, `Ctrl+C`	Quit; asks first when workers run, `q` again kills them while they stop

Key	On a failure transport
`[` `]`	Turn the pages of failed messages
`r`	Send the failed message back as is
`⏎`	Edit the failed message, `Ctrl+S` sends it back, `Esc` cancels
`d`	Remove the failed message
`e`	Show the errors of the failed message in full
`R`, `D`	Send every failed message back, remove them all

How it works

Workers are regular messenger:consume <transport> processes, started through your application console. Each one runs with the MESSENGER_CENTRAL_WORKER environment variable, which turns on a listener of the worker events: it writes one JSON line per event (message received, handled, failed, heartbeat...) on the standard output, read by the dashboard. Outside the dashboard, the listener does nothing.

Workers also get a unique MESSENGER_CONSUMER_NAME environment variable, made of the id of the dashboard process and of the worker. Use it as the consumer name of Redis transports, as the Symfony documentation recommends, so that scaled workers do not share their pending messages:

framework:
    messenger:
        transports:
            async:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                options:
                    consumer: '%env(MESSENGER_CONSUMER_NAME)%'

Give the variable a default value in .env, for the processes the dashboard does not start (a worker started by hand, a web request...):

MESSENGER_CONSUMER_NAME=consumer

Waiting messages are counted for transports that implement MessageCountAwareInterface (Doctrine, Redis, AMQP, SQS, Beanstalkd), and listed for those implementing ListableReceiverInterface (Doctrine). A probe process started by the dashboard, the hidden messenger:central:probe command, inspects them: a broker that does not answer never freezes the dashboard, and services are reset after each inspection, as messenger:consume does.

Workers that stop by themselves, because of --time-limit, --memory-limit, --limit or messenger:stop-workers, are started again at once while their transport runs. A worker that crashes is started again after 1, 2, 4… up to 30 seconds, so that one that cannot start does not loop.

When you stop workers or quit, they get SIGTERM and finish their current messages; the ones still running after the stop timeout (10 seconds by default) are killed. Quitting asks first when workers run, then shows them stopping: q again kills them at once. SIGTERM and SIGINT sent to the dashboard quit the same way. If the dashboard itself is killed, its workers notice it and stop after their current messages.

Failed messages

Workers cannot be started on a failure transport. Open it instead: its failed messages are listed under it, 20 per page ([ and ] turn the pages), with the transport they failed on and their last error. On a message:

r sends it back as is to the transport it failed on, which removes it from the failure transport;
⏎ edits it first: its properties show up as a JSON document, Ctrl+S sends the message back with your changes, Esc cancels;
d removes it for good;
e shows its errors in full, with the stack trace of the last one.

On the failure transport, R sends every failed message back, and D removes them all. Removing and sending everything back ask for a confirmation.

{
    "email": "jane@example.com",
    "status": "pending",
    "placedAt": "2026-10-13T10:42:01.000000+02:00",
    "address": {
        "street": "1 rue de la Paix",
        "city": "Paris"
    }
}

The document does not depend on the serializer of the transport: it holds the properties of the message, private and read-only ones included, and of the objects it contains. Enums are written with their value (or the name of their case), dates as strings.
Edited objects are copied without running their constructor, but the set hooks of the properties you change do run. An object that did not exist, in a property that was null, is built through its constructor.
Values that JSON cannot hold, like closures or instances of PHP classes other than dates, are left out and kept as they are. A message that is itself an instance of a PHP class, like a message that could not be decoded, cannot be edited.
The message keeps its stamps (bus, retries and errors so far), but not the ones about its stay in the failure transport, so that it goes back there if it fails again.
Listing and editing failed messages needs a failure transport that lists its messages (ListableReceiverInterface, like Doctrine).

Concurrency

With Symfony 8.2 and amphp/parallel installed, press c to type how many messages each worker of a transport handles at the same time: the dashboard starts its workers with messenger:consume --concurrency=N. Changing it restarts the running workers. A transport has at most 32 messages handled at the same time (workers × concurrency), as each one is a process that boots your application.

The child processes that handle the messages are started by Symfony through the console of your application (bin/console), whatever the console_path of the dashboard. The memory shown is the one of the workers, without their child processes.

Configuration

Everything is optional:

# config/packages/dev/messenger_central.yaml
messenger_central:
    # The console used to start the workers
    console_path: '%kernel.project_dir%/bin/console'
    # Extra options given to messenger:consume (but --concurrency, see below)
    consume_options: ['--time-limit=3600', '--memory-limit=256M']
    # Messages handled at the same time by each worker, Symfony 8.2 only
    concurrency: 1
    # Seconds given to the workers to finish their messages when stopped
    stop_timeout: 10

Limitations

Only the workers started by the dashboard are shown.
In-memory transports live in the memory of each process: the dashboard cannot see their messages.
Failed messages are sent back to their transport without going through the bus: its middlewares do not run until a worker handles the message.
Message classes are found from the routing configuration and from the messages that have a handler. A message routed with #[AsMessage] but handled by another application is not listed.

Contributing

composer install
composer qa   # mago fmt --dry-run, mago lint, mago analyze, phpunit

Configuration trees are indented by nesting level, which the formatter would flatten: a project rule of mago lint (in tools/mago) checks it, and mago lint --fix fixes it. The builder methods that open a level are read from the installed Symfony Config component, so new node types need no change.

License

MIT, see LICENSE.

welcomattic / messenger-central

Package info

Security

Fund package maintenance!

Statistics