implementing purge operation

Author: DanielePalaia

examples/getting_started/main.py

@@ -15,16 +15,19 @@ def main() -> None:
     routing_key = "routing-key"
     connection = Connection("amqp://guest:guest@localhost:5672/")
 
+    print("connection to amqp server")
     connection.dial()
 
     management = connection.management()
 
+    print("declaring exchange and queue")
     management.declare_exchange(ExchangeSpecification(name=exchange_name, arguments={}))
 
     management.declare_queue(
         QueueSpecification(name=queue_name, queue_type=QueueType.quorum, arguments={})
     )
 
+    print("binding queue to exchange")
     bind_name = management.bind(
         BindingSpecification(
             source_exchange=exchange_name,
@@ -35,22 +38,27 @@ def main() -> None:
 
     addr = exchange_address(exchange_name, routing_key)
 
+    print("create a publisher and publish a test message")
     publisher = connection.publisher(addr)
 
     publisher.publish(Message(body="test"))
 
     publisher.close()
 
+    print("unbind")
     management.unbind(bind_name)
 
-    # management.purge_queue(queue_info.name)
+    print("purging queue")
+    management.purge_queue(queue_name)
 
+    print("delete queue")
     management.delete_queue(queue_name)
 
+    print("delete exchange")
     management.delete_exchange(exchange_name)
 
+    print("closing connections")
     management.close()
-
     connection.close()
 
 

rabbitmq_amqp_python_client/address_helper.py

@@ -16,6 +16,12 @@ def queue_address(name: str) -> str:
     return path
 
 
+def purge_queue_address(name: str) -> str:
+    path = "/queues/" + name + "/messages"
+
+    return path
+
+
 def path_address() -> str:
     path = "/bindings"
 

rabbitmq_amqp_python_client/management.py

@@ -6,6 +6,7 @@
     binding_path_with_exchange_queue,
     exchange_address,
     path_address,
+    purge_queue_address,
     queue_address,
 )
 from .common import CommonValues
@@ -157,7 +158,6 @@ def delete_exchange(self, exchange_name: str) -> None:
             path,
             CommonValues.command_delete.value,
             [
-                CommonValues.response_code_200.value,
                 CommonValues.response_code_204.value,
             ],
         )
@@ -172,7 +172,6 @@ def delete_queue(self, queue_name: str) -> None:
             CommonValues.command_delete.value,
             [
                 CommonValues.response_code_200.value,
-                CommonValues.response_code_204.value,
             ],
         )
 
@@ -223,7 +222,6 @@ def unbind(self, binding_exchange_queue_path: str) -> None:
             binding_exchange_queue_path,
             CommonValues.command_delete.value,
             [
-                CommonValues.response_code_200.value,
                 CommonValues.response_code_204.value,
             ],
         )
@@ -232,4 +230,17 @@ def unbind(self, binding_exchange_queue_path: str) -> None:
     # def queue_info(self, queue_name:str):
 
     # TODO
-    # def purge_queue(self, queue_name:str):
+    def purge_queue(self, queue_name: str):
+        logger.debug("purge_queue operation called")
+        path = purge_queue_address(queue_name)
+
+        print("path: " + path)
+
+        self.request(
+            None,
+            path,
+            CommonValues.command_delete.value,
+            [
+                CommonValues.response_code_200.value,
+            ],
+        )

rabbitmq_amqp_python_client/qpid/proton/_endpoints.py

@@ -1317,7 +1317,6 @@ def send(
         Where the object is a :class:`Message`, this will send the message over
         this link, creating a new delivery for the purpose.
         """
-        print("XXXXXXXXXXXX")
         if hasattr(obj, "send"):
             return obj.send(self, tag=tag)
         else:

tests/test_management.py

@@ -26,7 +26,7 @@ def test_declare_delete_exchange() -> None:
     connection.close()
 
 
-def test_declare_delete_queue() -> None:
+def test_declare_purge_delete_queue() -> None:
     connection = Connection("amqp://guest:guest@localhost:5672/")
     connection.dial()
 
@@ -39,7 +39,8 @@ def test_declare_delete_queue() -> None:
 
     assert queue_info.name == queue_name
 
-    # Still not working
+    management.purge_queue(queue_name)
+
     management.delete_queue(queue_name)
 
     connection.close()
@@ -81,11 +82,8 @@ def test_bind_exchange_to_queue() -> None:
         + ";args="
     )
 
-    # Still not working
     management.delete_exchange(exchange_name)
 
-    # Still not working
     management.delete_queue(queue_name)
 
-    # Still not working
     management.unbind(binding_exchange_queue_path)

tests/test_publisher.py

@@ -28,5 +28,5 @@ def test_bind_exchange_to_queue() -> None:
     assert raised is False
 
     publisher.close()
-    # Still not working
+
     management.delete_queue(queue_name)