implementing bind

Author: DanielePalaia

examples/getting_started/Makefile

@@ -1,4 +1,5 @@
 from rabbitmq_amqp_python_client import (
+    BindingSpecification,
     Connection,
     ExchangeSpecification,
     QueueSpecification,
@@ -7,8 +8,9 @@
 
 
 def main() -> None:
-    exchange_name = "getting-started-exchange"
+    exchange_name = "test-exchange"
     queue_name = "example-queue"
+    routing_key = "routing-key"
     connection = Connection("amqp://guest:guest@localhost:5672/")
 
     connection.dial()
@@ -21,10 +23,13 @@ def main() -> None:
         QueueSpecification(name=queue_name, queue_type=QueueType.quorum, arguments={})
     )
 
-    """
-    #management.bind(BindingSpecification(source_exchange=exchange_name, destination_queue=queue_name, \
-    binding_key=routing_key))
-    """
+    binding_exchange_queue_path = management.bind(
+        BindingSpecification(
+            source_exchange=exchange_name,
+            destination_queue=queue_name,
+            binding_key=routing_key,
+        )
+    )
 
     """
     addr = exchange_address(exchange_name, routing_key)
@@ -44,9 +49,7 @@ def main() -> None:
     publisher.close()
     """
 
-    """
-    management.unbind(binding_path)
-    """
+    management.unbind(binding_exchange_queue_path)
 
     """
     management.purge_queue(queue_info.name)

examples/getting_started/main.py

@@ -3,6 +3,7 @@
 from .common import QueueType
 from .connection import Connection
 from .entities import (
+    BindingSpecification,
     ExchangeSpecification,
     QueueSpecification,
 )
@@ -20,5 +21,6 @@
     "Connection",
     "ExchangeSpecification",
     "QueueSpecification",
+    "BindingSpecification",
     "QueueType",
 ]

examples/getting_started/requirements.txt

@@ -1,3 +1,6 @@
+from .entities import BindingSpecification
+
+
 def exchange_address(name: str) -> str:
     path = "/exchanges/" + name
 
@@ -8,3 +11,25 @@ def queue_address(name: str) -> str:
     path = "/queues/" + name
 
     return path
+
+
+def path_address() -> str:
+    path = "/bindings"
+
+    return path
+
+
+def binding_path_with_exchange_queue(bind_specification: BindingSpecification) -> str:
+    binding_path_wth_exchange_queue_key = (
+        "/bindings"
+        + "/"
+        + "src="
+        + bind_specification.source_exchange
+        + ";"
+        + "dstq="
+        + bind_specification.destination_queue
+        + ";key="
+        + bind_specification.binding_key
+        + ";args="
+    )
+    return binding_path_wth_exchange_queue_key

rabbitmq_amqp_python_client/__init__.py

@@ -25,3 +25,11 @@ class QueueSpecification:
     dead_letter_exchange: str = ""
     is_auto_delete: bool = False
     is_durable: bool = True
+
+
+@dataclass
+class BindingSpecification:
+    source_exchange: str
+    destination_queue: str
+    # destination_exchange: str
+    binding_key: str

rabbitmq_amqp_python_client/address_helper.py

@@ -9,9 +9,15 @@
     BlockingSender,
 )
 
-from .address_helper import exchange_address, queue_address
+from .address_helper import (
+    binding_path_with_exchange_queue,
+    exchange_address,
+    path_address,
+    queue_address,
+)
 from .common import CommonValues
 from .entities import (
+    BindingSpecification,
     ExchangeSpecification,
     QueueSpecification,
 )
@@ -80,9 +86,6 @@ def _request(
 
         self._validate_reponse_code(int(msg.subject), expected_response_codes)
 
-    # TODO
-    # def delete_queue(self, name:str):
-
     def declare_exchange(
         self, exchange_specification: ExchangeSpecification
     ) -> ExchangeSpecification:
@@ -123,8 +126,6 @@ def declare_queue(
 
         path = queue_address(queue_specification.name)
 
-        print(path)
-
         self.request(
             body,
             path,
@@ -155,8 +156,6 @@ def delete_exchange(self, exchange_name: str) -> None:
     def delete_queue(self, queue_name: str) -> None:
         path = queue_address(queue_name)
 
-        print(path)
-
         self.request(
             None,
             path,
@@ -169,6 +168,7 @@ def delete_queue(self, queue_name: str) -> None:
     def _validate_reponse_code(
         self, response_code: int, expected_response_codes: list[int]
     ) -> None:
+        print("response_code received: " + str(response_code))
         if response_code == CommonValues.response_code_409.value:
             # TODO replace with a new defined Exception
             raise Exception("ErrPreconditionFailed")
@@ -180,10 +180,37 @@ def _validate_reponse_code(
         raise Exception("wrong response code received")
 
     # TODO
-    # def bind(self, bind_specification:BindSpecification):
+    def bind(self, bind_specification: BindingSpecification) -> str:
+        body = {}
+        body["binding_key"] = bind_specification.binding_key
+        body["source"] = bind_specification.source_exchange
+        body["destination_queue"] = bind_specification.destination_queue
+        body["arguments"] = {}  # type: ignore
+
+        path = path_address()
+
+        self.request(
+            body,
+            path,
+            CommonValues.command_post.value,
+            [
+                CommonValues.response_code_204.value,
+            ],
+        )
+
+        binding_path_with_queue = binding_path_with_exchange_queue(bind_specification)
+        return binding_path_with_queue
 
     # TODO
-    # def unbind(self, binding_path:str):
+    def unbind(self, binding_exchange_queue_path: str) -> None:
+        self.request(
+            None,
+            binding_exchange_queue_path,
+            CommonValues.command_delete.value,
+            [
+                CommonValues.response_code_200.value,
+            ],
+        )
 
     # TODO
     # def queue_info(self, queue_name:str):

rabbitmq_amqp_python_client/entities.py

@@ -1,4 +1,5 @@
 from rabbitmq_amqp_python_client import (
+    BindingSpecification,
     Connection,
     ExchangeSpecification,
     QueueSpecification,
@@ -32,13 +33,59 @@ def test_declare_delete_queue() -> None:
     queue_name = "test-queue"
     management = connection.management()
 
-    exchange_info = management.declare_queue(
+    queue_info = management.declare_queue(
         QueueSpecification(name=queue_name, queue_type=QueueType.quorum, arguments={})
     )
 
-    assert exchange_info.name == queue_name
+    assert queue_info.name == queue_name
 
     # Still not working
     # management.delete_queue(queue_name)
 
     connection.close()
+
+
+def test_bind_exchange_to_queue() -> None:
+    connection = Connection("amqp://guest:guest@localhost:5672/")
+    connection.dial()
+
+    exchange_name = "test-bind-exchange-to-queue-exchange"
+    queue_name = "test-bind-exchange-to-queue-queue"
+    routing_key = "routing-key"
+    management = connection.management()
+
+    management.declare_exchange(ExchangeSpecification(name=exchange_name, arguments={}))
+
+    management.declare_queue(
+        QueueSpecification(name=queue_name, queue_type=QueueType.quorum, arguments={})
+    )
+
+    binding_exchange_queue_path = management.bind(
+        BindingSpecification(
+            source_exchange=exchange_name,
+            destination_queue=queue_name,
+            binding_key=routing_key,
+        )
+    )
+
+    print(binding_exchange_queue_path)
+
+    assert (
+        binding_exchange_queue_path
+        == "/bindings/src="
+        + exchange_name
+        + ";dstq="
+        + queue_name
+        + ";key="
+        + routing_key
+        + ";args="
+    )
+
+    # Still not working
+    # management.delete_exchange(exchange_name)
+
+    # Still not working
+    # management.delete_queue(queue_name)
+
+    # Still not working
+    # management.delete_bind(binding_exchange_queue_path)