[Messenger] Add a Doctrine transport

[vincenttouzet]

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#868

As 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 works

The code is splitted betwwen this PR and the one on the DoctrineBundle : doctrine/DoctrineBundle#868

Configuration

To 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)

        # config/packages/messenger.yaml
        framework:
            messenger:
                transports:
                    my_transport: "doctrine://default?queue=important"

Table schema

Dispatched 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 dispatching

When dispatching a message a new row is inserted into the table. See Symfony\Component\Messenger\Transport\Doctrine::publish

Message consuming

The 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 message

The 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 requeueing

It 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

• Add tests
• Create DOC PR
• PR on doctrine-bundle for transport factory
• Add a available_at column
• Add a queue column
• Implement the retry functionnality : See [Messenger] Worker events + global retry functionality #30557
• Rebase after [Messenger] Add a command to setup transports #29476

[Taluu]

Looks good, but overall missing typehints / return hints as it's a feature for 4.3+, so php 7.1+. I'm guessing the missing tests are because it's a wip :}

[sroze]

Can you add @experimental in 4.3 on all these classes please? Then, a few remaining comments. After this, it should be ready to go!

[sroze]

Let's do this then. It's going to be YYYY-MM-DDTHH:mm:ss.sssZ, with the DateTime as UTC. I suggest to use a function rather than having the ->format(...) call copy/pasted.

[sroze]

@vincenttouzet mind the conflict with master. Once you've removed the id_strategy, can you squash and rebase as well?

[vincenttouzet]

@sroze I've rebase and squash my commits :)

[sroze]

@vincenttouzet but you kept the id_strategy 😉

[vincenttouzet]

That was just the doc in the configuration (part of another commit).

[sroze]

Type::JSON does not exists on early Doctrine versions as the low test suite is highlighting. Therefore, I suggest to move it back to STRING (like the body) and use json_encode. Whoever wants to have the table with a JSON field can create it manually. (FYI @weaverryan)

[vincenttouzet]

Yes i saw that. I was trying to require a minimum version for Doctrine but the Json type is available since v2.6.0 (tagged 2017-07-23 01:02) ... I've put back the json_encode / json_decode with the string type then

[sroze]

Thank you @vincenttouzet.

[vincenttouzet]

@sroze while updating the Doc PR I realized that I've keep the loop_sleep option but this is not used. I can make another PR to remove it

[sroze]

@vincenttouzet sure.

[vincenttouzet]

@sroze Hum there is also a problem with the DoctrineTransportFactory:

Fatal error: Declaration of Symfony\Component\Messenger\Transport\Doctrine\DoctrineTransportFactory::createTransport(string $dsn, array $options): Symfony\Component\Messenger\Transport\TransportInterface must be compatible with Symfony\Component\Messenger\Transport\TransportFactoryInterface::createTransport(string $dsn, array $options, Symfony\Component\Messenger\Transport\Serialization\SerializerInterface $serializer): Symfony\Component\Messenger\Transport\TransportInterface in /var/www/sources/symfony/src/Symfony/Component/Messenger/Transport/Doctrine/DoctrineTransportFactory.php on line 26

it worked before merging 🤔 . I can fix this in the same MR if you want

[sroze]

I fixed it in #30799.

[vincenttouzet]

@sroze Ok. The PR to remove the unused option is created : #30803