Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly
Appearance settings

Commit f0b0993

Browse filesBrowse files
vincenttouzetvincenttouzet
committed
[Messenger] Add a Doctrine transport
1 parent edcd627 commit f0b0993
Copy full SHA for f0b0993

File tree

9 files changed

+418
-0
lines changed
Filter options

9 files changed

+418
-0
lines changed

‎src/Symfony/Bundle/FrameworkBundle/DependencyInjection/FrameworkExtension.php

Copy file name to clipboardExpand all lines: src/Symfony/Bundle/FrameworkBundle/DependencyInjection/FrameworkExtension.php
+1Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1499,6 +1499,7 @@ private function registerMessengerConfiguration(array $config, ContainerBuilder
14991499
if (empty($config['transports'])) {
15001500
$container->removeDefinition('messenger.transport.symfony_serializer');
15011501
$container->removeDefinition('messenger.transport.amqp.factory');
1502+
$container->removeDefinition('messenger.transport.doctrine.factory');
15021503
} else {
15031504
if ('messenger.transport.symfony_serializer' === $config['serializer']['id']) {
15041505
if (!$this->isConfigEnabled($container, $serializerConfig)) {

‎src/Symfony/Bundle/FrameworkBundle/Resources/config/messenger.xml

Copy file name to clipboardExpand all lines: src/Symfony/Bundle/FrameworkBundle/Resources/config/messenger.xml
+8Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -61,5 +61,13 @@
6161

6262
<tag name="messenger.transport_factory" />
6363
</service>
64+
65+
<service id="messenger.transport.doctrine.factory" class="Symfony\Component\Messenger\Transport\Doctrine\DoctrineTransportFactory">
66+
<argument type="service" id="doctrine" on-invalid="ignore"/>
67+
<argument type="service" id="messenger.transport.serializer" />
68+
<argument>%kernel.debug%</argument>
69+
70+
<tag name="messenger.transport_factory" />
71+
</service>
6472
</services>
6573
</container>

‎src/Symfony/Component/Messenger/CHANGELOG.md

Copy file name to clipboardExpand all lines: src/Symfony/Component/Messenger/CHANGELOG.md
+5Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,11 @@
11
CHANGELOG
22
=========
33

4+
4.3.0
5+
-----
6+
7+
* Add a `Doctrine` transport using `doctrine://<connection_name>` DSN
8+
49
4.2.0
510
-----
611

‎src/Symfony/Component/Messenger/Tests/Transport/Doctrine/DoctrineTransportFactoryTest.php

Copy file name to clipboard
+79Lines changed: 79 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,79 @@
1+
<?php
2+
3+
/*
4+
* This file is part of the Symfony package.
5+
*
6+
* (c) Fabien Potencier <fabien@symfony.com>
7+
*
8+
* For the full copyright and license information, please view the LICENSE
9+
* file that was distributed with this source code.
10+
*/
11+
12+
namespace Symfony\Component\Messenger\Tests\Transport\Doctrine;
13+
14+
use PHPUnit\Framework\TestCase;
15+
use Symfony\Bridge\Doctrine\RegistryInterface;
16+
use Symfony\Component\Messenger\Transport\Doctrine\Connection;
17+
use Symfony\Component\Messenger\Transport\Doctrine\DoctrineTransport;
18+
use Symfony\Component\Messenger\Transport\Doctrine\DoctrineTransportFactory;
19+
use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
20+
21+
class DoctrineTransportFactoryTest extends TestCase
22+
{
23+
24+
public function testSupports()
25+
{
26+
$factory = new DoctrineTransportFactory(
27+
$this->getMockBuilder(RegistryInterface::class)->getMock(),
28+
null,
29+
false
30+
);
31+
32+
$this->assertTrue($factory->supports('doctrine://default', array()));
33+
$this->assertFalse($factory->supports('amqp://localhost', array()));
34+
}
35+
36+
public function testCreateTransport()
37+
{
38+
$connection = $this->getMockBuilder(\Doctrine\DBAL\Connection::class)
39+
->disableOriginalConstructor()
40+
->getMock();
41+
$registry = $this->getMockBuilder(RegistryInterface::class)->getMock();
42+
$registry->expects($this->once())
43+
->method('getConnection')
44+
->willReturn($connection);
45+
46+
$factory = new DoctrineTransportFactory(
47+
$registry,
48+
null,
49+
false
50+
);
51+
52+
$this->assertEquals(
53+
new DoctrineTransport(new Connection('messenger_messages', $connection, false), null),
54+
$factory->createTransport('doctrine://default', array())
55+
);
56+
}
57+
58+
public function testCreateTransportWithCustomTableName()
59+
{
60+
$connection = $this->getMockBuilder(\Doctrine\DBAL\Connection::class)
61+
->disableOriginalConstructor()
62+
->getMock();
63+
$registry = $this->getMockBuilder(RegistryInterface::class)->getMock();
64+
$registry->expects($this->once())
65+
->method('getConnection')
66+
->willReturn($connection);
67+
68+
$factory = new DoctrineTransportFactory(
69+
$registry,
70+
null,
71+
false
72+
);
73+
74+
$this->assertEquals(
75+
new DoctrineTransport(new Connection('custom_messages', $connection, false), null),
76+
$factory->createTransport('doctrine://default', array('table_name' => 'custom_messages'))
77+
);
78+
}
79+
}

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

Copy file name to clipboard
+105Lines changed: 105 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,105 @@
1+
<?php
2+
3+
/*
4+
* This file is part of the Symfony package.
5+
*
6+
* (c) Fabien Potencier <fabien@symfony.com>
7+
*
8+
* For the full copyright and license information, please view the LICENSE
9+
* file that was distributed with this source code.
10+
*/
11+
12+
namespace Symfony\Component\Messenger\Transport\Doctrine;
13+
14+
use Doctrine\DBAL\Connection as DriverConnection;
15+
use Doctrine\DBAL\Schema\Schema;
16+
use Doctrine\DBAL\Schema\Synchronizer\SingleDatabaseSynchronizer;
17+
use Doctrine\DBAL\Types\Type;
18+
19+
/**
20+
* @author Vincent Touzet <vincent.touzet@gmail.com>
21+
*/
22+
class Connection
23+
{
24+
private $tableName;
25+
private $driverConnection;
26+
private $debug;
27+
28+
const STATUS_PENDING = 'pending';
29+
const STATUS_PROCESSING = 'processing';
30+
const STATUS_ACK = 'ack';
31+
32+
public function __construct(string $tableName, DriverConnection $driverConnection, bool $debug = false)
33+
{
34+
$this->tableName = $tableName;
35+
$this->driverConnection = $driverConnection;
36+
$this->debug = $debug;
37+
}
38+
39+
public function publish(string $body, array $headers): void
40+
{
41+
if (!$this->debug) {
42+
$this->setup();
43+
}
44+
$stmt = $this->driverConnection->prepare('INSERT INTO '.$this->tableName.' (`body`, `headers`) VALUES (:body, :headers)');
45+
$stmt->execute(array(
46+
':body' => $body,
47+
':headers' => $headers,
48+
));
49+
}
50+
51+
public function get(): array
52+
{
53+
if (!$this->debug) {
54+
$this->setup();
55+
}
56+
$stmt = $this->driverConnection->prepare("SELECT * FROM {$this->tableName} WHERE status = 'pending' ORDER BY id ASC LIMIT 0,1");
57+
$stmt->execute();
58+
59+
$doctrineEnvelop = $stmt->fetch();
60+
61+
$this->changeStatus($doctrineEnvelop['id'], self::STATUS_PROCESSING);
62+
63+
return $doctrineEnvelop;
64+
}
65+
66+
public function ack($id): bool
67+
{
68+
return $this->changeStatus($id, self::STATUS_ACK);
69+
}
70+
71+
public function nack($id): bool
72+
{
73+
return $this->changeStatus($id, self::STATUS_PENDING);
74+
}
75+
76+
private function changeStatus($id, $status): bool
77+
{
78+
return $this->driverConnection->update($this->tableName, array('status' => $status), array('id' => $id)) > 0;
79+
}
80+
81+
private function setup(): void
82+
{
83+
$synchronizer = new SingleDatabaseSynchronizer($this->driverConnection);
84+
$synchronizer->updateSchema($this->getSchema());
85+
}
86+
87+
private function getSchema(): Schema
88+
{
89+
$schema = new Schema();
90+
$table = $schema->createTable($this->tableName);
91+
$table->addColumn('id', Type::BIGINT)
92+
->setAutoincrement(true)
93+
->setNotnull(true);
94+
$table->addColumn('body', Type::TEXT)
95+
->setNotnull(true);
96+
$table->addColumn('headers', Type::TARRAY)
97+
->setNotnull(true);
98+
$table->addColumn('status', Type::STRING)
99+
->setLength(32)
100+
->setDefault(self::STATUS_PENDING);
101+
$table->setPrimaryKey(array('id'));
102+
103+
return $schema;
104+
}
105+
}

‎src/Symfony/Component/Messenger/Transport/Doctrine/DoctrineReceiver.php

Copy file name to clipboard
+74Lines changed: 74 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,74 @@
1+
<?php
2+
3+
/*
4+
* This file is part of the Symfony package.
5+
*
6+
* (c) Fabien Potencier <fabien@symfony.com>
7+
*
8+
* For the full copyright and license information, please view the LICENSE
9+
* file that was distributed with this source code.
10+
*/
11+
12+
namespace Symfony\Component\Messenger\Transport\Doctrine;
13+
14+
use Symfony\Component\Messenger\Transport\Receiver\ReceiverInterface;
15+
use Symfony\Component\Messenger\Transport\Serialization\Serializer;
16+
17+
/**
18+
* @author Vincent Touzet <vincent.touzet@gmail.com>
19+
*/
20+
class DoctrineReceiver implements ReceiverInterface
21+
{
22+
private $connection;
23+
private $serializer;
24+
private $shouldStop = false;
25+
26+
public function __construct(Connection $connection, Serializer $serializer = null)
27+
{
28+
$this->connection = $connection;
29+
$this->serializer = $serializer ?? Serializer::create();
30+
}
31+
32+
public function receive(callable $handler): void
33+
{
34+
while (!$this->shouldStop) {
35+
$doctrineEnvelop = $this->connection->get();
36+
37+
if (null === $doctrineEnvelop) {
38+
$handler(null);
39+
40+
usleep(200000);
41+
$this->pcntlDispatch();
42+
43+
continue;
44+
}
45+
46+
try {
47+
$handler($this->serializer->decode(array(
48+
'body' => $doctrineEnvelop['body'],
49+
'headers' => $doctrineEnvelop['headers'],
50+
)));
51+
52+
$this->connection->ack($doctrineEnvelop['id']);
53+
} catch (\Throwable $e) {
54+
$this->connection->nack($doctrineEnvelop['id']);
55+
56+
throw $e;
57+
} finally {
58+
$this->pcntlDispatch();
59+
}
60+
}
61+
}
62+
63+
public function stop(): void
64+
{
65+
$this->shouldStop = true;
66+
}
67+
68+
private function pcntlDispatch()
69+
{
70+
if (\function_exists('pcntl_signal_dispatch')) {
71+
pcntl_signal_dispatch();
72+
}
73+
}
74+
}

‎src/Symfony/Component/Messenger/Transport/Doctrine/DoctrineSender.php

Copy file name to clipboard
+40Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,40 @@
1+
<?php
2+
3+
/*
4+
* This file is part of the Symfony package.
5+
*
6+
* (c) Fabien Potencier <fabien@symfony.com>
7+
*
8+
* For the full copyright and license information, please view the LICENSE
9+
* file that was distributed with this source code.
10+
*/
11+
12+
namespace Symfony\Component\Messenger\Transport\Doctrine;
13+
14+
use Symfony\Component\Messenger\Envelope;
15+
use Symfony\Component\Messenger\Transport\Sender\SenderInterface;
16+
use Symfony\Component\Messenger\Transport\Serialization\Serializer;
17+
18+
/**
19+
* @author Vincent Touzet <vincent.touzet@gmail.com>
20+
*/
21+
class DoctrineSender implements SenderInterface
22+
{
23+
private $connection;
24+
private $serializer;
25+
26+
public function __construct(Connection $connection, Serializer $serializer = null)
27+
{
28+
$this->connection = $connection;
29+
$this->serializer = $serializer ?? Serializer::create();
30+
}
31+
32+
public function send(Envelope $envelope): Envelope
33+
{
34+
$encodedMessage = $this->serializer->encode($envelope);
35+
36+
$this->connection->publish($encodedMessage['body'], $encodedMessage['headers']);
37+
38+
return $envelope;
39+
}
40+
}

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

Copy file name to clipboard
+58Lines changed: 58 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,58 @@
1+
<?php
2+
3+
/*
4+
* This file is part of the Symfony package.
5+
*
6+
* (c) Fabien Potencier <fabien@symfony.com>
7+
*
8+
* For the full copyright and license information, please view the LICENSE
9+
* file that was distributed with this source code.
10+
*/
11+
12+
namespace Symfony\Component\Messenger\Transport\Doctrine;
13+
14+
use Symfony\Component\Messenger\Envelope;
15+
use Symfony\Component\Messenger\Transport\Serialization\Serializer;
16+
use Symfony\Component\Messenger\Transport\TransportInterface;
17+
18+
/**
19+
* @author Vincent Touzet <vincent.touzet@gmail.com>
20+
*/
21+
class DoctrineTransport implements TransportInterface
22+
{
23+
private $connection;
24+
private $serializer;
25+
private $receiver;
26+
private $sender;
27+
28+
public function __construct(Connection $connection, Serializer $serializer = null)
29+
{
30+
$this->connection = $connection;
31+
$this->serializer = $serializer ?? Serializer::create();
32+
}
33+
34+
public function receive(callable $handler): void
35+
{
36+
($this->receiver ?? $this->getReceiver())->receive($handler);
37+
}
38+
39+
public function stop(): void
40+
{
41+
($this->receiver ?? $this->getReceiver())->stop();
42+
}
43+
44+
public function send(Envelope $envelope): Envelope
45+
{
46+
return ($this->sender ?? $this->getSender())->send($envelope);
47+
}
48+
49+
private function getReceiver(): DoctrineReceiver
50+
{
51+
return $this->receiver = new DoctrineReceiver($this->connection, $this->serializer);
52+
}
53+
54+
private function getSender(): DoctrineSender
55+
{
56+
return $this->sender = new DoctrineSender($this->connection, $this->serializer);
57+
}
58+
}

0 commit comments

Comments
0 (0)
Morty Proxy This is a proxified and sanitized view of the page, visit original site.