feature symfony#59273 [Messenger] Add BeanstalkdPriorityStamp to Beanstalkd bridge (HypeMC)

This PR was merged into the 7.3 branch.

Discussion
----------

[Messenger] Add `BeanstalkdPriorityStamp` to Beanstalkd bridge

| Q             | A
| ------------- | ---
| Branch?       | 7.3
| Bug fix?      | no
| New feature?  | yes
| Deprecations? | no
| Issues        | -
| License       | MIT

Extracted from symfony#49652 as the original PR was trying to do two separate things in one.

Beanstalkd has the ability to configure the priority at which messages are reserved. This can now be set using the `BeanstalkdPriorityStamp`:

```php
$this->bus->dispatch(new SomeMessage('some data'), [
    new BeanstalkdPriorityStamp(0),
]);
```

Commits
-------

9926d59 [Messenger] Add BeanstalkdPriorityStamp to Beanstalkd bridge

Author: fabpot

src/Symfony/Component/Messenger/Bridge/Beanstalkd/CHANGELOG.md

@@ -1,6 +1,11 @@
 CHANGELOG
 =========
 
+7.3
+---
+
+ * Add `BeanstalkdPriorityStamp` option to allow setting the message priority
+
 7.2
 ---
 

src/Symfony/Component/Messenger/Bridge/Beanstalkd/Tests/Transport/BeanstalkdSenderTest.php

@@ -13,6 +13,7 @@
 
 use PHPUnit\Framework\TestCase;
 use Symfony\Component\Messenger\Bridge\Beanstalkd\Tests\Fixtures\DummyMessage;
+use Symfony\Component\Messenger\Bridge\Beanstalkd\Transport\BeanstalkdPriorityStamp;
 use Symfony\Component\Messenger\Bridge\Beanstalkd\Transport\BeanstalkdSender;
 use Symfony\Component\Messenger\Bridge\Beanstalkd\Transport\Connection;
 use Symfony\Component\Messenger\Envelope;
@@ -27,7 +28,7 @@ public function testSend()
         $encoded = ['body' => '...', 'headers' => ['type' => DummyMessage::class]];
 
         $connection = $this->createMock(Connection::class);
-        $connection->expects($this->once())->method('send')->with($encoded['body'], $encoded['headers'], 0);
+        $connection->expects($this->once())->method('send')->with($encoded['body'], $encoded['headers'], 0, null);
 
         $serializer = $this->createMock(SerializerInterface::class);
         $serializer->method('encode')->with($envelope)->willReturn($encoded);
@@ -42,7 +43,22 @@ public function testSendWithDelay()
         $encoded = ['body' => '...', 'headers' => ['type' => DummyMessage::class]];
 
         $connection = $this->createMock(Connection::class);
-        $connection->expects($this->once())->method('send')->with($encoded['body'], $encoded['headers'], 500);
+        $connection->expects($this->once())->method('send')->with($encoded['body'], $encoded['headers'], 500, null);
+
+        $serializer = $this->createMock(SerializerInterface::class);
+        $serializer->method('encode')->with($envelope)->willReturn($encoded);
+
+        $sender = new BeanstalkdSender($connection, $serializer);
+        $sender->send($envelope);
+    }
+
+    public function testSendWithPriority()
+    {
+        $envelope = (new Envelope(new DummyMessage('Oy')))->with(new BeanstalkdPriorityStamp(2));
+        $encoded = ['body' => '...', 'headers' => ['type' => DummyMessage::class]];
+
+        $connection = $this->createMock(Connection::class);
+        $connection->expects($this->once())->method('send')->with($encoded['body'], $encoded['headers'], 0, 2);
 
         $serializer = $this->createMock(SerializerInterface::class);
         $serializer->method('encode')->with($envelope)->willReturn($encoded);

src/Symfony/Component/Messenger/Bridge/Beanstalkd/Tests/Transport/ConnectionTest.php

@@ -297,6 +297,41 @@ public function testSend()
         $this->assertSame($id, (int) $returnedId);
     }
 
+    public function testSendWithPriority()
+    {
+        $tube = 'xyz';
+
+        $body = 'foo';
+        $headers = ['test' => 'bar'];
+        $delay = 1000;
+        $priority = 2;
+        $expectedDelay = $delay / 1000;
+
+        $id = 110;
+
+        $client = $this->createMock(PheanstalkInterface::class);
+        $client->expects($this->once())->method('useTube')->with($tube)->willReturn($client);
+        $client->expects($this->once())->method('put')->with(
+            $this->callback(function (string $data) use ($body, $headers): bool {
+                $expectedMessage = json_encode([
+                    'body' => $body,
+                    'headers' => $headers,
+                ]);
+
+                return $expectedMessage === $data;
+            }),
+            $priority,
+            $expectedDelay,
+            90
+        )->willReturn(new Job($id, 'foobar'));
+
+        $connection = new Connection(['tube_name' => $tube], $client);
+
+        $returnedId = $connection->send($body, $headers, $delay, $priority);
+
+        $this->assertSame($id, (int) $returnedId);
+    }
+
     public function testSendWhenABeanstalkdExceptionOccurs()
     {
         $tube = 'xyz';

src/Symfony/Component/Messenger/Bridge/Beanstalkd/Transport/BeanstalkdPriorityStamp.php

@@ -0,0 +1,22 @@
+<?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\Transport;
+
+use Symfony\Component\Messenger\Stamp\StampInterface;
+
+final readonly class BeanstalkdPriorityStamp implements StampInterface
+{
+    public function __construct(
+        public int $priority,
+    ) {
+    }
+}

src/Symfony/Component/Messenger/Bridge/Beanstalkd/Transport/BeanstalkdSender.php

@@ -35,11 +35,12 @@ public function send(Envelope $envelope): Envelope
     {
         $encodedMessage = $this->serializer->encode($envelope);
 
-        /** @var DelayStamp|null $delayStamp */
-        $delayStamp = $envelope->last(DelayStamp::class);
-        $delayInMs = null !== $delayStamp ? $delayStamp->getDelay() : 0;
-
-        $this->connection->send($encodedMessage['body'], $encodedMessage['headers'] ?? [], $delayInMs);
+        $this->connection->send(
+            $encodedMessage['body'],
+            $encodedMessage['headers'] ?? [],
+            $envelope->last(DelayStamp::class)?->getDelay() ?? 0,
+            $envelope->last(BeanstalkdPriorityStamp::class)?->priority,
+        );
 
         return $envelope;
     }

src/Symfony/Component/Messenger/Bridge/Beanstalkd/Transport/Connection.php

@@ -105,11 +105,12 @@ public function getTube(): string
     }
 
     /**
-     * @param int $delay The delay in milliseconds
+     * @param int  $delay    The delay in milliseconds
+     * @param ?int $priority The priority at which the message will be reserved
      *
      * @return string The inserted id
      */
-    public function send(string $body, array $headers, int $delay = 0): string
+    public function send(string $body, array $headers, int $delay = 0, ?int $priority = null): string
     {
         $message = json_encode([
             'body' => $body,
@@ -123,7 +124,7 @@ public function send(string $body, array $headers, int $delay = 0): string
         try {
             $job = $this->client->useTube($this->tube)->put(
                 $message,
-                PheanstalkInterface::DEFAULT_PRIORITY,
+                $priority ?? PheanstalkInterface::DEFAULT_PRIORITY,
                 (int) ($delay / 1000),
                 $this->ttr
             );