… (weaverryan)This PR was merged into the 4.3-dev branch.Discussion----------[Messenger] Worker events + global retry functionality| Q             | A| ------------- | ---| Branch?       | master| Bug fix?      | no| New feature?  | yes| BC breaks?    | yes, on Messenger only| Deprecations? | no| Tests pass?   | NEEDED| Fixed tickets |#29132,#27008,#27215 and part of#30540| License       | MIT| Doc PR        | TODOThis is an alternative to#29132 and#27008. There are several big things:1) The `messenger:consume` does not die if a handler has an error2) Events are dispatched before, after and on error a message being handled3) Logic is moved out of Amqp and into the Worker so that we can have some consistent features, like error handling.4) A generic retry system was added, which works with Amqp and future transports should support. It will work out of the box for users. Retrying works by putting the received `Envelope` back into the bus, but with the `ReceivedStamp` removed. The retry functionality has an integration test for AMQP.5) Added a new `MessageDecodingFailedException` that transport Serializers should throw if `decode()` fails. It allows us to reject a message in this situation, as allowing it to fail but remain on the queue causes it to be retried forever.6) A new `DelayStamp` was added, which is the first of (later) more stamps for configuring the transport layer (see#30558).BC breaks are documented in the CHANGELOG.Thanks!Commits-------a989384 Adding global retry support, events & more to messenger transport

…nnect() does not work (sroze)This PR was merged into the 4.3-dev branch.Discussion----------[Messenger] Ensure an exception is thrown when the AMQP connect() does not work| Q             | A| ------------- | ---| Branch?       | master| Bug fix?      | yes| New feature?  | no| BC breaks?    | no| Deprecations? | no| Tests pass?   | yes| Fixed tickets |#30557| License       | MIT| Doc PR        | øThis `connectionCredentials` instance escaped the renaming in#30557.Commits-------46b9476 Ensure an exception is thrown when the AMQP connect() does not work

…d (weaverryan)This PR was merged into the 4.3-dev branch.Discussion----------Dispatching two events when a message is sent & handled| Q             | A| ------------- | ---| Branch?       | master| Bug fix?      | no| New feature?  | yes| BC breaks?    | no| Deprecations? | no| Tests pass?   | yes| Fixed tickets | none| License       | MIT| Doc PR        | TODOAlternative to#30646. This uses a more generic system, so you could do anything when a message is sent. The main use-case is when a message is dispatched by a 3rd party.I didn't try to add *exhaustive* events everywhere: I added an event for a very specific use-case:When a message is dispatched by a 3rd party, being able to add stamps (e.g. `DelayStamp` or a future `AmqpRoutingKeyStamp` before the message is sent. Example:```phpclass MailerMessageSendToTransportEventSubscriber implements EventSubscriberInterface{    public function onSendMessage(SendMessageToTransportsEvent $event)    {        $envelope = $event->getEnvelope();        if (!$envelope->getMessage() instanceof SomeMailerMessage) {            return;        }        $event->setEnvelope($envelope->with(new AmpqRoutingKeyStamp('mailer-route')));    }    public static function getSubscribedEvents()    {        return [SendMessageToTransportsEvent::class => 'onSendMessage'];    }}```Along with#30557, we will now have the following events, regarding async messages:* Event when a message is sent to transports (this PR)* Event when a message is received from transport, but before handling it* Event when a message is received from transport and after handling itCommits-------a7ad1b4 Dispatching two events when a message is sent & handled

This PR was merged into the 4.3-dev branch.Discussion----------[Messenger] Add a Doctrine transport| Q             | A| ------------- | ---| Branch?       | master| Bug fix?      | no| New feature?  | yes| BC breaks?    | no| Deprecations? | no| Tests pass?   | yes| Fixed tickets || License       | MIT| Doc PR        |symfony/symfony-docs#10616| DoctrineBundle PR |doctrine/DoctrineBundle#868As discussed with@sroze at PHPForum in Paris I've worked on adding a Doctrine transport to the Messenger component.Actually `AMQP` is the only supported transport and it could be a good thing to support multiple transports. Having a Doctrine transport could help users to start using the component IMHO (Almost all projects use a database).# How it worksThe code is splitted betwwen this PR and the one on the DoctrineBundle :doctrine/DoctrineBundle#868## ConfigurationTo configure a Doctrine transport the dsn MUST have the format `doctrine://<entity_manager_name>` where `<entity_manager_name>` is the name of the entity manager (usually `default`)```yml        # config/packages/messenger.yaml        framework:            messenger:                transports:                    my_transport: "doctrine://default?queue=important"```## Table schemaDispatched messages are stored into a database table with the following schema:| Column       | Type     | Options                  | Description                                                       ||--------------|----------|--------------------------|-------------------------------------------------------------------|| id           | bigint   | AUTO_INCREMENT, NOT NULL | Primary key                                                       || body         | text     | NOT NULL                 | Body of the message                                               || headers      | text     | NOT NULL                 | Headers of the message                                            || queue      | varchar(32)     | NOT NULL                 | Headers of the message                                            || created_at   | datetime | NOT NULL                 | When the message was inserted onto the table. (automatically set) || available_at       | datetime   | NOT NULL                 | When the message is available to be handled                      || delivered_at | datetime | NULL                     | When the message was delivered to a worker                        |## Message dispatchingWhen dispatching a message a new row is inserted into the table. See `Symfony\Component\Messenger\Transport\Doctrine::publish`## Message consumingThe message is retrieved by the `Symfony\Component\Messenger\Transport\Doctrine\DoctrineReceiver`. It calls the `Symfony\Component\Messenger\Transport\Doctrine::get` method to get the next message to handle.### Getting the next message* Start a transaction* Lock the table to get the first message to handle (The lock is done with the `SELECT ... FOR UPDATE` query)* Update the message in database to update the delivered_at columns* Commit the transaction### Handling the messageThe retrieved message is then passed to the handler. If the message is correctly handled the receiver call the `Symfony\Component\Messenger\Transport\Doctrine::ack` which delete the message from the table.If an error occured the receiver call the `Symfony\Component\Messenger\Transport\Doctrine::nack` method which update the message to set the delivered_at column to `null`.## Message requeueingIt may happen that a message is stuck in `delivered` state but the handler does not really handle the message (Database connection error, server crash, ...). To requeue messages the `DoctrineReceiver` call the `Symfony\Component\Messenger\Transport\Doctrine::requeueMessages`. This method update all the message with a  `delivered_at` not null since more than the "redeliver timeout" (default to 3600 seconds)# TODO- [x] Add tests- [x] Create DOC PR- [x] PR on doctrine-bundle for transport factory- [x] Add a `available_at` column- [x] Add a `queue` column- [x] Implement the retry functionnality : See#30557- [x] Rebase after#29476Commits-------88d008c [Messenger] Add a Doctrine transport