Key and partition selector

.gitattributes

@@ -0,0 +1 @@
+/Tests export-ignore

EnvelopeItem/KeyStamp.php

@@ -0,0 +1,24 @@
+<?php
+
+namespace Enqueue\MessengerAdapter\EnvelopeItem;
+
+use Interop\Queue\Message;
+use Symfony\Component\Messenger\Stamp\NonSendableStampInterface;
+use Symfony\Component\Messenger\Stamp\StampInterface;
+
+class KeyStamp implements StampInterface
+{
+    private string $key;
+
+    public function getKey(): ?string
+    {
+        return $this->key;
+    }
+
+    public function setKey(string $key = null): self
+    {
+        $this->key = $key;
+
+        return $this;
+    }
+}

EnvelopeItem/PartitionStamp.php

@@ -0,0 +1,24 @@
+<?php
+
+namespace Enqueue\MessengerAdapter\EnvelopeItem;
+
+use Interop\Queue\Message;
+use Symfony\Component\Messenger\Stamp\NonSendableStampInterface;
+use Symfony\Component\Messenger\Stamp\StampInterface;
+
+class PartitionStamp implements StampInterface
+{
+    private int $partition;
+
+    public function getPartition(): int
+    {
+        return $this->partition;
+    }
+
+    public function setPartition(int $partition): self
+    {
+        $this->partition = $partition;
+
+        return $this;
+    }
+}

QueueInteropTransport.php

@@ -15,6 +15,8 @@
 use Enqueue\AmqpTools\RabbitMqDelayPluginDelayStrategy;
 use Enqueue\AmqpTools\RabbitMqDlxDelayStrategy;
 use Enqueue\MessengerAdapter\EnvelopeItem\InteropMessageStamp;
+use Enqueue\MessengerAdapter\EnvelopeItem\KeyStamp;
+use Enqueue\MessengerAdapter\EnvelopeItem\PartitionStamp;
 use Enqueue\MessengerAdapter\EnvelopeItem\TransportConfiguration;
 use Enqueue\MessengerAdapter\Exception\MissingMessageMetadataSetterException;
 use Enqueue\MessengerAdapter\Exception\SendingMessageFailedException;
@@ -168,6 +170,16 @@ public function send(Envelope $envelope): Envelope
             $producer->setTimeToLive($this->options['timeToLive']);
         }
 
+        $partitionStamp = $envelope->last(PartitionStamp::class);
+        if (method_exists($interopMessage, 'setPartition') && !is_null($partitionStamp)) {
+            $interopMessage->setPartition($partitionStamp->getPartition());
+        }
+
+        $keyStamp = $envelope->last(KeyStamp::class);
+        if (method_exists($interopMessage, 'setKey') && !is_null($keyStamp)) {
+            $interopMessage->setKey($keyStamp->getKey());
+        }
+
         try {
             $producer->send($topic, $interopMessage);
         } catch (InteropQueueException $e) {

Tests/EnvelopeItem/KeyStampTest.php

@@ -0,0 +1,21 @@
+<?php
+
+namespace Enqueue\MessengerAdapter\Tests\EnvelopeItem;
+
+use Enqueue\MessengerAdapter\EnvelopeItem\KeyStamp;
+use PHPUnit\Framework\TestCase;
+
+class KeyStampTest extends TestCase
+{
+    public function testPartitionSetter()
+    {
+        $keyStamp = (new KeyStamp())->setKey('key');
+        $this->assertInstanceOf(KeyStamp::class, $keyStamp);
+    }
+
+    public function testPartitionGetter()
+    {
+        $keyStamp = (new KeyStamp())->setKey('key');
+        $this->assertEquals('key', $keyStamp->getKey());
+    }
+}

Tests/EnvelopeItem/PartitionStampTest.php

@@ -0,0 +1,21 @@
+<?php
+
+namespace Enqueue\MessengerAdapter\Tests\EnvelopeItem;
+
+use Enqueue\MessengerAdapter\EnvelopeItem\PartitionStamp;
+use PHPUnit\Framework\TestCase;
+
+class PartitionStampTest extends TestCase
+{
+    public function testPartitionSetter()
+    {
+        $partitionStamp = (new PartitionStamp())->setPartition(1);
+        $this->assertInstanceOf(PartitionStamp::class, $partitionStamp);
+    }
+
+    public function testPartitionGetter()
+    {
+        $partitionStamp = (new PartitionStamp())->setPartition(1);
+        $this->assertEquals(1, $partitionStamp->getPartition());
+    }
+}