publish message with custom queue options : flags | attributes

symfony · fabpot · Feb 21, 2019 · Feb 13, 2019 · Feb 13, 2019 · Feb 17, 2019
commit 6f9fdaf7e4a218a88095a102e603bfc3cfd7d52d
diff --git a/src/Symfony/Component/Messenger/Tests/Transport/AmqpExt/ConnectionTest.php b/src/Symfony/Component/Messenger/Tests/Transport/AmqpExt/ConnectionTest.php
@@ -233,6 +233,24 @@ public function testItCanDisableTheSetup()
        $connection = Connection::fromDsn('amqp://localhost/%2f/messages?queue[routing_key]=my_key&auto-setup=false', [], true, $factory);
        $connection->publish('body');
    }
+
+    public function testPublishWithQueueOptions()
+    {
+        $factory = new TestAmqpFactory(
+            $amqpConnection = $this->getMockBuilder(\AMQPConnection::class)->disableOriginalConstructor()->getMock(),
+            $amqpChannel = $this->getMockBuilder(\AMQPChannel::class)->disableOriginalConstructor()->getMock(),
+            $amqpQueue = $this->getMockBuilder(\AMQPQueue::class)->disableOriginalConstructor()->getMock(),
+            $amqpExchange = $this->getMockBuilder(\AMQPExchange::class)->disableOriginalConstructor()->getMock()
+        );
+        $headers = [
+            'type' => '*',
+        ];
+        $amqpExchange->expects($this->once())->method('publish')
+            ->with('body', null, 1, ['delivery_mode' => 2, 'headers' => $headers]);
+
+        $connection = Connection::fromDsn('amqp://localhost/%2f/messages?queue[attributes][delivery_mode]=2&queue[flags]=1', [], true, $factory);
+        $connection->publish('body', $headers);
+    }
 }

 class TestAmqpFactory extends AmqpFactory

diff --git a/src/Symfony/Component/Messenger/Transport/AmqpExt/Connection.php b/src/Symfony/Component/Messenger/Transport/AmqpExt/Connection.php
@@ -133,7 +133,10 @@ public function publish(string $body, array $headers = []): void
            $this->setup();
        }

-        $this->exchange()->publish($body, $this->queueConfiguration['routing_key'] ?? null, AMQP_NOPARAM, ['headers' => $headers]);
+        $flags = $this->queueConfiguration['flags'] ?? AMQP_NOPARAM;
+        $attributes = array_merge_recursive($this->queueConfiguration['attributes'] ?? [], ['headers' => $headers]);
+
+        $this->exchange()->publish($body, $this->queueConfiguration['routing_key'] ?? null, $flags, $attributes);
    }

    /**
-Original file line number
+Diff line change
@@ -133,7 +133,10 @@ public function publish(string $body, array $headers = []): void
                 $this->setup();
             }
-            $this->exchange()->publish($body, $this->queueConfiguration['routing_key'] ?? null, AMQP_NOPARAM, ['headers' => $headers]);
+            $flags = $this->queueConfiguration['flags'] ?? AMQP_NOPARAM;
+            $attributes = array_merge_recursive($this->queueConfiguration['attributes'] ?? [], ['headers' => $headers]);
                   fabpot marked this conversation as resolved.
              

          
            Show resolved
            Hide resolved
        

                
  
            
  

  
    
  
    
      

              Uh oh!

              
There was an error while loading. Please reload this page.
+            $this->exchange()->publish($body, $this->queueConfiguration['routing_key'] ?? null, $flags, $attributes);
         }
         /**
-          Expand Down