symfony · fabpot · Aug 13, 2020 · Apr 25, 2020
diff --git a/composer.json b/composer.json
@@ -122,6 +122,7 @@
         "nyholm/psr7": "^1.0",
         "ocramius/proxy-manager": "^2.1",
         "paragonie/sodium_compat": "^1.8",
+        "pda/pheanstalk": "^4.0",
         "php-http/httplug": "^1.0|^2.0",
         "predis/predis": "~1.1",
         "psr/http-client": "^1.0",

diff --git a/src/Symfony/Bundle/FrameworkBundle/DependencyInjection/FrameworkExtension.php b/src/Symfony/Bundle/FrameworkBundle/DependencyInjection/FrameworkExtension.php
@@ -84,6 +84,7 @@
 use Symfony\Component\Mailer\Mailer;
 use Symfony\Component\Messenger\Bridge\AmazonSqs\Transport\AmazonSqsTransportFactory;
 use Symfony\Component\Messenger\Bridge\Amqp\Transport\AmqpTransportFactory;
+use Symfony\Component\Messenger\Bridge\Beanstalkd\Transport\BeanstalkdTransportFactory;
 use Symfony\Component\Messenger\Bridge\Redis\Transport\RedisTransportFactory;
 use Symfony\Component\Messenger\Handler\MessageHandlerInterface;
 use Symfony\Component\Messenger\MessageBus;
@@ -1672,6 +1673,10 @@ private function registerMessengerConfiguration(array $config, ContainerBuilder
             $container->getDefinition('messenger.transport.sqs.factory')->addTag('messenger.transport_factory');
         }
 
+        if (class_exists(BeanstalkdTransportFactory::class)) {
+            $container->getDefinition('messenger.transport.beanstalkd.factory')->addTag('messenger.transport_factory');
+        }
+
         if (null === $config['default_bus'] && 1 === \count($config['buses'])) {
             $config['default_bus'] = key($config['buses']);
         }
@@ -1730,6 +1735,7 @@ private function registerMessengerConfiguration(array $config, ContainerBuilder
             $container->removeDefinition('messenger.transport.amqp.factory');
             $container->removeDefinition('messenger.transport.redis.factory');
             $container->removeDefinition('messenger.transport.sqs.factory');
+            $container->removeDefinition('messenger.transport.beanstalkd.factory');
             $container->removeAlias(SerializerInterface::class);
         } else {
             $container->getDefinition('messenger.transport.symfony_serializer')

diff --git a/src/Symfony/Bundle/FrameworkBundle/Resources/config/messenger.php b/src/Symfony/Bundle/FrameworkBundle/Resources/config/messenger.php
@@ -13,6 +13,7 @@
 
 use Symfony\Component\Messenger\Bridge\AmazonSqs\Transport\AmazonSqsTransportFactory;
 use Symfony\Component\Messenger\Bridge\Amqp\Transport\AmqpTransportFactory;
+use Symfony\Component\Messenger\Bridge\Beanstalkd\Transport\BeanstalkdTransportFactory;
 use Symfony\Component\Messenger\Bridge\Redis\Transport\RedisTransportFactory;
 use Symfony\Component\Messenger\EventListener\DispatchPcntlSignalListener;
 use Symfony\Component\Messenger\EventListener\SendFailedMessageForRetryListener;
@@ -123,6 +124,8 @@
 
         ->set('messenger.transport.sqs.factory', AmazonSqsTransportFactory::class)
 
+        ->set('messenger.transport.beanstalkd.factory', BeanstalkdTransportFactory::class)
+
         // retry
         ->set('messenger.retry_strategy_locator')
             ->args([

diff --git a/...ny/Bundle/FrameworkBundle/Tests/DependencyInjection/Fixtures/php/messenger_transports.php b/...ny/Bundle/FrameworkBundle/Tests/DependencyInjection/Fixtures/php/messenger_transports.php
@@ -22,6 +22,7 @@
             ],
             'failed' => 'in-memory:///',
             'redis' => 'redis://127.0.0.1:6379/messages',
+            'beanstalkd' => 'beanstalkd://127.0.0.1:11300',
         ],
     ],
 ]);
diff --git a/...ny/Bundle/FrameworkBundle/Tests/DependencyInjection/Fixtures/xml/messenger_transports.xml b/...ny/Bundle/FrameworkBundle/Tests/DependencyInjection/Fixtures/xml/messenger_transports.xml
@@ -20,6 +20,7 @@
             </framework:transport>
             <framework:transport name="failed" dsn="in-memory:///" />
             <framework:transport name="redis" dsn="redis://127.0.0.1:6379/messages" />
+            <framework:transport name="beanstalkd" dsn="beanstalkd://127.0.0.1:11300" />
         </framework:messenger>
     </framework:config>
 </container>
diff --git a/...ny/Bundle/FrameworkBundle/Tests/DependencyInjection/Fixtures/yml/messenger_transports.yml b/...ny/Bundle/FrameworkBundle/Tests/DependencyInjection/Fixtures/yml/messenger_transports.yml
@@ -19,3 +19,4 @@ framework:
                     max_delay: 100
             failed: 'in-memory:///'
             redis: 'redis://127.0.0.1:6379/messages'
+            beanstalkd: 'beanstalkd://127.0.0.1:11300'
diff --git a/src/Symfony/Bundle/FrameworkBundle/Tests/DependencyInjection/FrameworkExtensionTest.php b/src/Symfony/Bundle/FrameworkBundle/Tests/DependencyInjection/FrameworkExtensionTest.php
@@ -660,6 +660,16 @@ public function testMessengerTransports()
 
         $this->assertTrue($container->hasDefinition('messenger.transport.redis.factory'));
 
+        $this->assertTrue($container->hasDefinition('messenger.transport.beanstalkd'));
+        $transportFactory = $container->getDefinition('messenger.transport.beanstalkd')->getFactory();
+        $transportArguments = $container->getDefinition('messenger.transport.beanstalkd')->getArguments();
+
+        $this->assertEquals([new Reference('messenger.transport_factory'), 'createTransport'], $transportFactory);
+        $this->assertCount(3, $transportArguments);
+        $this->assertSame('beanstalkd://127.0.0.1:11300', $transportArguments[0]);
+
+        $this->assertTrue($container->hasDefinition('messenger.transport.beanstalkd.factory'));
+
         $this->assertSame(10, $container->getDefinition('messenger.retry.multiplier_retry_strategy.customised')->getArgument(0));
         $this->assertSame(7, $container->getDefinition('messenger.retry.multiplier_retry_strategy.customised')->getArgument(1));
         $this->assertSame(3, $container->getDefinition('messenger.retry.multiplier_retry_strategy.customised')->getArgument(2));

diff --git a/src/Symfony/Component/Messenger/Bridge/Beanstalkd/.gitattributes b/src/Symfony/Component/Messenger/Bridge/Beanstalkd/.gitattributes
@@ -0,0 +1,3 @@
+/Tests export-ignore
+/phpunit.xml.dist export-ignore
+/.gitignore export-ignore
diff --git a/src/Symfony/Component/Messenger/Bridge/Beanstalkd/.gitignore b/src/Symfony/Component/Messenger/Bridge/Beanstalkd/.gitignore
@@ -0,0 +1,3 @@
+vendor/
+composer.lock
+phpunit.xml
diff --git a/src/Symfony/Component/Messenger/Bridge/Beanstalkd/CHANGELOG.md b/src/Symfony/Component/Messenger/Bridge/Beanstalkd/CHANGELOG.md
@@ -0,0 +1,7 @@
+CHANGELOG
+=========
+
+5.2.0
+-----
+
+ * Introduced the Beanstalkd bridge.
diff --git a/src/Symfony/Component/Messenger/Bridge/Beanstalkd/LICENSE b/src/Symfony/Component/Messenger/Bridge/Beanstalkd/LICENSE
@@ -0,0 +1,19 @@
+Copyright (c) 2018-2020 Fabien Potencier
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is furnished
+to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.
diff --git a/src/Symfony/Component/Messenger/Bridge/Beanstalkd/README.md b/src/Symfony/Component/Messenger/Bridge/Beanstalkd/README.md
@@ -0,0 +1,14 @@
+Beanstalkd Messenger
+====================
+
+Provides Beanstalkd integration for Symfony Messenger.
+
+Full DSN with options: `beanstalkd://<ip>:<port>?tube_name=<name>&timeout=<timeoutInSeconds>&ttr=<ttrInSeconds>`
+
+Resources
+---------
+
+  * [Contributing](https://symfony.com/doc/current/contributing/index.html)
+  * [Report issues](https://github.com/symfony/symfony/issues) and
+    [send Pull Requests](https://github.com/symfony/symfony/pulls)
+    in the [main Symfony repository](https://github.com/symfony/symfony)
diff --git a/src/Symfony/Component/Messenger/Bridge/Beanstalkd/Tests/Fixtures/DummyMessage.php b/src/Symfony/Component/Messenger/Bridge/Beanstalkd/Tests/Fixtures/DummyMessage.php
@@ -0,0 +1,18 @@
+<?php
+
+namespace Symfony\Component\Messenger\Bridge\Beanstalkd\Tests\Fixtures;
+
+class DummyMessage
+{
+    private $message;
+
+    public function __construct(string $message)
+    {
+        $this->message = $message;
+    }
+
+    public function getMessage(): string
+    {
+        return $this->message;
+    }
+}
diff --git a/src/Symfony/Component/Messenger/Bridge/Beanstalkd/Tests/Transport/BeanstalkdReceiverTest.php b/src/Symfony/Component/Messenger/Bridge/Beanstalkd/Tests/Transport/BeanstalkdReceiverTest.php
@@ -0,0 +1,100 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Bridge\Beanstalkd\Tests\Transport;
+
+use PHPUnit\Framework\TestCase;
+use Symfony\Component\Messenger\Bridge\Beanstalkd\Tests\Fixtures\DummyMessage;
+use Symfony\Component\Messenger\Bridge\Beanstalkd\Transport\BeanstalkdReceivedStamp;
+use Symfony\Component\Messenger\Bridge\Beanstalkd\Transport\BeanstalkdReceiver;
+use Symfony\Component\Messenger\Bridge\Beanstalkd\Transport\Connection;
+use Symfony\Component\Messenger\Exception\MessageDecodingFailedException;
+use Symfony\Component\Messenger\Transport\Serialization\PhpSerializer;
+use Symfony\Component\Messenger\Transport\Serialization\Serializer;
+use Symfony\Component\Serializer as SerializerComponent;
+use Symfony\Component\Serializer\Encoder\JsonEncoder;
+use Symfony\Component\Serializer\Normalizer\ObjectNormalizer;
+
+final class BeanstalkdReceiverTest extends TestCase
+{
+    public function testItReturnsTheDecodedMessageToTheHandler()
+    {
+        $serializer = $this->createSerializer();
+
+        $tube = 'foo bar';
+
+        $beanstalkdEnvelope = $this->createBeanstalkdEnvelope();
+        $connection = $this->getMockBuilder(Connection::class)->disableOriginalConstructor()->getMock();
+        $connection->expects($this->once())->method('get')->willReturn($beanstalkdEnvelope);
+        $connection->expects($this->once())->method('getTube')->willReturn($tube);
+
+        $receiver = new BeanstalkdReceiver($connection, $serializer);
+        $actualEnvelopes = $receiver->get();
+        $this->assertCount(1, $actualEnvelopes);
+        $this->assertEquals(new DummyMessage('Hi'), $actualEnvelopes[0]->getMessage());
+
+        /** @var BeanstalkdReceivedStamp $receivedStamp */
+        $receivedStamp = $actualEnvelopes[0]->last(BeanstalkdReceivedStamp::class);
+
+        $this->assertInstanceOf(BeanstalkdReceivedStamp::class, $receivedStamp);
+        $this->assertSame('1', $receivedStamp->getId());
+        $this->assertSame($tube, $receivedStamp->getTube());
+    }
+
+    public function testItReturnsEmptyArrayIfThereAreNoMessages()
+    {
+        $serializer = $this->createSerializer();
+
+        $connection = $this->getMockBuilder(Connection::class)->disableOriginalConstructor()->getMock();
+        $connection->expects($this->once())->method('get')->willReturn(null);
+
+        $receiver = new BeanstalkdReceiver($connection, $serializer);
+        $actualEnvelopes = $receiver->get();
+        $this->assertIsArray($actualEnvelopes);
+        $this->assertCount(0, $actualEnvelopes);
+    }
+
+    public function testItRejectTheMessageIfThereIsAMessageDecodingFailedException()
+    {
+        $this->expectException(MessageDecodingFailedException::class);
+
+        $serializer = $this->createMock(PhpSerializer::class);
+        $serializer->expects($this->once())->method('decode')->willThrowException(new MessageDecodingFailedException());
+
+        $beanstalkdEnvelope = $this->createBeanstalkdEnvelope();
+        $connection = $this->getMockBuilder(Connection::class)->disableOriginalConstructor()->getMock();
+        $connection->expects($this->once())->method('get')->willReturn($beanstalkdEnvelope);
+        $connection->expects($this->once())->method('reject');
+
+        $receiver = new BeanstalkdReceiver($connection, $serializer);
+        $receiver->get();
+    }
+
+    private function createBeanstalkdEnvelope(): array
+    {
+        return [
+            'id' => '1',
+            'body' => '{"message": "Hi"}',
+            'headers' => [
+                'type' => DummyMessage::class,
+            ],
+        ];
+    }
+
+    private function createSerializer(): Serializer
+    {
+        $serializer = new Serializer(
+            new SerializerComponent\Serializer([new ObjectNormalizer()], ['json' => new JsonEncoder()])
+        );
+
+        return $serializer;
+    }
+}
diff --git a/src/Symfony/Component/Messenger/Bridge/Beanstalkd/Tests/Transport/BeanstalkdSenderTest.php b/src/Symfony/Component/Messenger/Bridge/Beanstalkd/Tests/Transport/BeanstalkdSenderTest.php
@@ -0,0 +1,55 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Bridge\Beanstalkd\Tests\Transport;
+
+use PHPUnit\Framework\TestCase;
+use Symfony\Component\Messenger\Bridge\Beanstalkd\Tests\Fixtures\DummyMessage;
+use Symfony\Component\Messenger\Bridge\Beanstalkd\Transport\BeanstalkdSender;
+use Symfony\Component\Messenger\Bridge\Beanstalkd\Transport\Connection;
+use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Stamp\DelayStamp;
+use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
+
+final class BeanstalkdSenderTest extends TestCase
+{
+    public function testSend()
+    {
+        $envelope = new Envelope(new DummyMessage('Oy'));
+        $encoded = ['body' => '...', 'headers' => ['type' => DummyMessage::class]];
+
+        $connection = $this->getMockBuilder(Connection::class)
+            ->disableOriginalConstructor()
+            ->getMock();
+        $connection->expects($this->once())->method('send')->with($encoded['body'], $encoded['headers'], 0);
+
+        $serializer = $this->getMockBuilder(SerializerInterface::class)->getMock();
+        $serializer->method('encode')->with($envelope)->willReturnOnConsecutiveCalls($encoded);
+
+        $sender = new BeanstalkdSender($connection, $serializer);
+        $sender->send($envelope);
+    }
+
+    public function testSendWithDelay()
+    {
+        $envelope = (new Envelope(new DummyMessage('Oy')))->with(new DelayStamp(500));
+        $encoded = ['body' => '...', 'headers' => ['type' => DummyMessage::class]];
+
+        $connection = $this->createMock(Connection::class);
+        $connection->expects($this->once())->method('send')->with($encoded['body'], $encoded['headers'], 500);
+
+        $serializer = $this->createMock(SerializerInterface::class);
+        $serializer->method('encode')->with($envelope)->willReturnOnConsecutiveCalls($encoded);
+
+        $sender = new BeanstalkdSender($connection, $serializer);
+        $sender->send($envelope);
+    }
+}
diff --git a/.../Component/Messenger/Bridge/Beanstalkd/Tests/Transport/BeanstalkdTransportFactoryTest.php b/.../Component/Messenger/Bridge/Beanstalkd/Tests/Transport/BeanstalkdTransportFactoryTest.php
@@ -0,0 +1,40 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Bridge\Beanstalkd\Tests\Transport;
+
+use PHPUnit\Framework\TestCase;
+use Symfony\Component\Messenger\Bridge\Beanstalkd\Transport\BeanstalkdTransport;
+use Symfony\Component\Messenger\Bridge\Beanstalkd\Transport\BeanstalkdTransportFactory;
+use Symfony\Component\Messenger\Bridge\Beanstalkd\Transport\Connection;
+use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
+
+final class BeanstalkdTransportFactoryTest extends TestCase
+{
+    public function testSupports()
+    {
+        $factory = new BeanstalkdTransportFactory();
+
+        $this->assertTrue($factory->supports('beanstalkd://127.0.0.1', []));
+        $this->assertFalse($factory->supports('doctrine://127.0.0.1', []));
+    }
+
+    public function testCreateTransport()
+    {
+        $factory = new BeanstalkdTransportFactory();
+        $serializer = $this->createMock(SerializerInterface::class);
+
+        $this->assertEquals(
+            new BeanstalkdTransport(Connection::fromDsn('beanstalkd://127.0.0.1'), $serializer),
+            $factory->createTransport('beanstalkd://127.0.0.1', [], $serializer)
+        );
+    }
+}