#23 Add function for delete exchange/queue

CHANGELOG.md

@@ -6,6 +6,7 @@ next release
 
 * Add `EventableConsumerInterface` for control specific errors (consumer timeout, stop after N executes, etc...).
 * Add possible to shuffle spool connections.
+* Add function for delete exchange and queue.
 
 v2.0.1
 ------

src/Adapter/Amqp/Exchange/AmqpExchange.php

@@ -94,4 +94,12 @@ public function publish(Message $message, string $routingKey = ''): void
 
         $this->exchange->publish($message->payload->data, $routingKey, AMQP_NOPARAM, $options);
     }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function delete(): void
+    {
+        $this->exchange->delete();
+    }
 }

src/Adapter/Amqp/Queue/AmqpQueue.php

@@ -104,6 +104,14 @@ public function purge(): void
         $this->queue->purge();
     }
 
+    /**
+     * {@inheritdoc}
+     */
+    public function delete(): void
+    {
+        $this->queue->delete();
+    }
+
     /**
      * {@inheritdoc}
      */

src/Adapter/AmqpLib/Exchange/AmqpExchange.php

@@ -98,4 +98,12 @@ public function publish(Message $message, string $routingKey = ''): void
         $amqplibMessage = new AMQPMessage($message->payload->data, $options);
         $this->channel->getChannel()->basic_publish($amqplibMessage, $this->getName(), $routingKey);
     }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function delete(): void
+    {
+        $this->channel->getChannel()->exchange_delete($this->getName());
+    }
 }

src/Adapter/AmqpLib/Queue/AmqpQueue.php

@@ -114,6 +114,14 @@ public function purge(): void
         $this->channel->getChannel()->queue_purge($this->getName());
     }
 
+    /**
+     * {@inheritdoc}
+     */
+    public function delete(): void
+    {
+        $this->channel->getChannel()->queue_delete($this->getName());
+    }
+
     /**
      * {@inheritdoc}
      *

src/Exchange/ExchangeInterface.php

@@ -42,4 +42,9 @@ public function getName(): string;
      * @param string  $routingKey
      */
     public function publish(Message $message, string $routingKey = ''): void;
+
+    /**
+     * Delete exchange
+     */
+    public function delete(): void;
 }

src/Queue/QueueInterface.php

@@ -65,6 +65,11 @@ public function get(): ?ReceivedMessage;
      */
     public function purge(): void;
 
+    /**
+     * Delete a queue
+     */
+    public function delete(): void;
+
     /**
      * Get a count messages
      *

tests/Functional/Adapter/ExchangeFactoryTestCase.php

@@ -155,6 +155,27 @@ public function shouldSuccessCreateWithBindings(): void
         ], $bindings);
     }
 
+    #[Test]
+    public function shouldSuccessDelete(): void
+    {
+        $this->management->createExchange('direct', 'some');
+
+        $definition = new ExchangeDefinition('some', AMQP_EX_TYPE_DIRECT);
+
+        $factory = $this->createExchangeFactory($definition);
+        $exchange = $factory->create();
+
+        $this->management->exchangeByName('some');
+        $this->addToAssertionCount(1); // because we success get exchange by name
+
+        $exchange->delete();
+
+        $this->expectException(\RuntimeException::class);
+        $this->expectExceptionMessage('The exchange with name "some" was not found.');
+
+        $this->management->exchangeByName('some');
+    }
+
     #[Test]
     public function shouldThrowExceptionOnCreatePassiveIfExchangeNotFound(): void
     {

tests/Functional/Adapter/QueueFactoryTestCase.php

@@ -272,6 +272,65 @@ public function shouldSuccessConsumeMessage(): void
         self::assertTrue($consumed, 'The queue not receive message.');
     }
 
+    #[Test]
+    public function shouldSuccessPurgeQueue(): void
+    {
+        $this->management->createExchange(AMQP_EX_TYPE_DIRECT, 'test.direct');
+        $this->management->createQueue('some');
+        $this->management->queueBind('some', 'test.direct', 'test');
+
+        $this->management->publishMessage('test.direct', 'test', 'some foo bar');
+        $this->management->publishMessage('test.direct', 'test', 'some foo bar');
+
+        $definition = new QueueDefinition(
+            'some',
+            new BindingDefinitions(
+                new BindingDefinition('test.direct', 'test')
+            )
+        );
+
+        $factory = $this->createQueueFactory($definition);
+        $queue = $factory->create();
+
+        $messages = $this->management->queueGetMessages('some', 2);
+
+        self::assertCount(2, $messages);
+
+        $queue->purge();
+
+        $messages = $this->management->queueGetMessages('some', 1);
+
+        self::assertCount(0, $messages);
+    }
+
+    #[Test]
+    public function shouldSuccessDeleteQueue(): void
+    {
+        $this->management->createExchange(AMQP_EX_TYPE_DIRECT, 'test.direct');
+        $this->management->createQueue('some');
+        $this->management->queueBind('some', 'test.direct', 'test');
+
+        $definition = new QueueDefinition(
+            'some',
+            new BindingDefinitions(
+                new BindingDefinition('test.direct', 'test')
+            )
+        );
+
+        $factory = $this->createQueueFactory($definition);
+        $queue = $factory->create();
+
+        $this->management->queueByName('some');
+        $this->addToAssertionCount(1); // Because success get queue by name
+
+        $queue->delete();
+
+        $this->expectException(\RuntimeException::class);
+        $this->expectExceptionMessage('The queue with name "some" was not found.');
+
+        $this->management->queueByName('some');
+    }
+
     #[Test]
     public function shouldSuccessGetCountMessages(): void
     {

tests/Unit/Adapter/Amqp/Exchange/AmqpExchangeTest.php

@@ -129,4 +129,13 @@ public function shouldSuccessPublishMessageWithHeaders(): void
 
         $this->exchange->publish($message, 'foo-bar');
     }
+
+    #[Test]
+    public function shouldSuccessDelete(): void
+    {
+        $this->originalExchange->expects(self::once())
+            ->method('delete');
+
+        $this->exchange->delete();
+    }
 }

tests/Unit/Adapter/Amqp/Queue/AmqpQueueTest.php

@@ -67,6 +67,15 @@ public function shouldSuccessPurge(): void
         $this->queue->purge();
     }
 
+    #[Test]
+    public function shouldSuccessDelete(): void
+    {
+        $this->originQueue->expects(self::once())
+            ->method('delete');
+
+        $this->queue->delete();
+    }
+
     #[Test]
     public function shouldSuccessGetChannel(): void
     {