bug #30805 [Messenger] bug fixes in Doctrine Transport (vincenttouzet)

This PR was merged into the 4.3-dev branch.

Discussion
----------

[Messenger] bug fixes in Doctrine Transport

| Q             | A
| ------------- | ---
| Branch?       | master
| Bug fix?      | yes
| New feature?  | no
| BC breaks?    | no
| Deprecations? | no
| Tests pass?   | yes
| Fixed tickets |
| License       | MIT
| Doc PR        |

Just tested the new Doctrine transport and I've see 3 bugs so far :
- [x] The message is not return by the transport
- [x] The headers column must be of type TEXT and not just STRING
- [ ] When using the PhpSerializer the message is truncated (PR: https://github.com/symfony/symfony/pull/30814)

The body in database looks like this :
```
O:36:"Symfony\Component\Messenger\Envelope":2:{s:44:"
```

The body given by the serializer is the following :
```
O:36:"Symfony\Component\Messenger\Envelope":2:{s:44:"Symfony\Component\Messenger\Envelopestamps";a:3:{s:49:"Symfony\Component\Messenger\Stamp\SerializerStamp";a:1:{i:0;O:49:"Symfony\Component\Messenger\Stamp\SerializerStamp":1:{s:58:"Symfony\Component\Messenger\Stamp\SerializerStampcontext";a:0:{}}}s:46:"Symfony\Component\Messenger\Stamp\BusNameStamp";a:1:{i:0;O:46:"Symfony\Component\Messenger\Stamp\BusNameStamp":1:{s:55:"Symfony\Component\Messenger\Stamp\BusNameStampbusName";s:21:"messenger.bus.default";}}s:43:"Symfony\Component\Messenger\Stamp\SentStamp";a:1:{i:0;O:43:"Symfony\Component\Messenger\Stamp\SentStamp":2:{s:56:"Symfony\Component\Messenger\Stamp\SentStampsenderClass";s:64:"Symfony\Component\Messenger\Transport\Doctrine\DoctrineTransport";s:56:"Symfony\Component\Messenger\Stamp\SentStampsenderAlias";s:16:"environment.stop";}}}s:45:"Symfony\Component\Messenger\Envelopemessage";O:34:"App\Message\EnvironmentStopMessage":1:{s:51:"App\Message\AbstractEnvironmentMessageenvironment";O:22:"App\Entity\Environment":5:{s:26:"App\Entity\Environmentid";s:36:"3bade252-b7a9-4188-82bd-3e68129e0da7";s:37:"App\Entity\EnvironmentrepositoryUrl";s:6:"string";s:30:"App\Entity\Environmentbranch";s:6:"string";s:33:"App\Entity\EnvironmenthostNames";a:1:{i:0;N;}s:27:"App\Entity\Environmentenv";a:2:{s:7:"APP_ENV";s:4:"prod";s:7:"APP_VAR";s:13:"example value";}}}}
```

Commits
-------

27466498d0 [Messenger] Fix get in Doctrine Transport

src/Symfony/Component/Messenger/Tests/Transport/Doctrine/DoctrineTransportTest.php

@@ -0,0 +1,60 @@
+<?php
+
+/*
+* This file is part of the symfony project.
+*
+* (c) Vincent Touzet <vincent.touzet@dotsafe.fr>
+*
+* For the full copyright and license information, please view the LICENSE
+* file that was distributed with this source code.
+*/
+
+namespace Symfony\Component\Messenger\Tests\Transport\Doctrine;
+
+use PHPUnit\Framework\TestCase;
+use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Tests\Fixtures\DummyMessage;
+use Symfony\Component\Messenger\Transport\Doctrine\Connection;
+use Symfony\Component\Messenger\Transport\Doctrine\DoctrineTransport;
+use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
+use Symfony\Component\Messenger\Transport\TransportInterface;
+
+class DoctrineTransportTest extends TestCase
+{
+    public function testItIsATransport()
+    {
+        $transport = $this->getTransport();
+
+        $this->assertInstanceOf(TransportInterface::class, $transport);
+    }
+
+    public function testReceivesMessages()
+    {
+        $transport = $this->getTransport(
+            $serializer = $this->getMockBuilder(SerializerInterface::class)->getMock(),
+            $connection = $this->getMockBuilder(Connection::class)->disableOriginalConstructor()->getMock()
+        );
+
+        $decodedMessage = new DummyMessage('Decoded.');
+
+        $doctrineEnvelope = [
+            'id' => '5',
+            'body' => 'body',
+            'headers' => ['my' => 'header'],
+        ];
+
+        $serializer->method('decode')->with(['body' => 'body', 'headers' => ['my' => 'header']])->willReturn(new Envelope($decodedMessage));
+        $connection->method('get')->willReturn($doctrineEnvelope);
+
+        $envelopes = iterator_to_array($transport->get());
+        $this->assertSame($decodedMessage, $envelopes[0]->getMessage());
+    }
+
+    private function getTransport(SerializerInterface $serializer = null, Connection $connection = null)
+    {
+        $serializer = $serializer ?: $this->getMockBuilder(SerializerInterface::class)->getMock();
+        $connection = $connection ?: $this->getMockBuilder(Connection::class)->disableOriginalConstructor()->getMock();
+
+        return new DoctrineTransport($connection, $serializer);
+    }
+}

src/Symfony/Component/Messenger/Transport/Doctrine/Connection.php

@@ -230,7 +230,7 @@ class Connection
             ->setNotnull(true);
         $table->addColumn('body', Type::TEXT)
             ->setNotnull(true);
-        $table->addColumn('headers', Type::STRING)
+        $table->addColumn('headers', Type::TEXT)
             ->setNotnull(true);
         $table->addColumn('queue_name', Type::STRING)
             ->setNotnull(true);

src/Symfony/Component/Messenger/Transport/Doctrine/DoctrineTransport.php

@@ -39,7 +39,7 @@ class DoctrineTransport implements TransportInterface, SetupableTransportInterfa
      */
     public function get(): iterable
     {
-        ($this->receiver ?? $this->getReceiver())->get();
+        return ($this->receiver ?? $this->getReceiver())->get();
     }
 
     /**