integrate qpid-proton lib

Author: DanielePalaia

.github/workflows/build-test.yaml

@@ -39,13 +39,13 @@ jobs:
       - name: poetry install
         run: poetry install --no-root
       - name: isort check-only
-        run: poetry run isort --check-only .
+        run: poetry run isort --skip rabbitmq_amqp_python_client/qpid --check-only .
       - name: black check
         run: poetry run black --check .
       - name: flake8
         run: poetry run flake8 --exclude=venv,local_tests,docs/examples --max-line-length=120 --ignore=E203,W503
       - name: mypy
         run: | 
-          poetry run mypy .
+          poetry run mypy --exclude=rabbitmq_amqp_python_client/qpid .
       - name: poetry run pytest
         run: poetry run pytest

examples/getting_started/main.py

@@ -21,36 +21,34 @@ def main() -> None:
 
     management.declare_exchange(ExchangeSpecification(name=exchange_name, arguments={}))
 
-    binding_exchange_queue_path = management.declare_queue(
+    management.declare_queue(
         QueueSpecification(name=queue_name, queue_type=QueueType.quorum, arguments={})
     )
 
+    bind_name = management.bind(
+        BindingSpecification(
+            source_exchange=exchange_name,
+            destination_queue=queue_name,
+            binding_key=routing_key,
+        )
+    )
 
-    #management.bind(
-    #    BindingSpecification(
-    #        source_exchange=exchange_name,
-    #        destination_queue=queue_name,
-    #        binding_key=routing_key,
-    #    )
-    #)
-
-    #addr = exchange_address(exchange_name, routing_key)
+    addr = exchange_address(exchange_name, routing_key)
 
-    #publisher = connection.publisher(addr)
+    publisher = connection.publisher(addr)
 
-    #publisher.publish(Message(body="test"))
+    publisher.publish(Message(body="test"))
 
-    #publisher.close()
+    publisher.close()
 
-    #management.unbind(binding_exchange_queue_path)
+    management.unbind(bind_name)
 
     # management.purge_queue(queue_info.name)
 
     management.delete_queue(queue_name)
 
     management.delete_exchange(exchange_name)
 
-
     management.close()
 
     connection.close()

poetry.lock

@@ -8,15 +8,13 @@ readme = "README.md"
 
 [tool.poetry.dependencies]
 python = "^3.9"
-python-qpid-proton = "^0.39.0"
 
 [tool.poetry.dev-dependencies]
 flake8 = "^7.1.1"
 isort = "^5.9.3"
 mypy = "^0.910"
 pytest = "^7.4.0"
 black = "^24.3.0"
-python-qpid-proton = "^0.39.0"
 
 [build-system]
 requires = ["poetry-core"]

pyproject.toml

@@ -1,8 +1,6 @@
 from importlib import metadata
 
-from proton import Message
-
-from .address_helper import exchange_address
+from .address_helper import exchange_address, queue_address
 from .common import QueueType
 from .connection import Connection
 from .entities import (
@@ -11,6 +9,7 @@
     QueueSpecification,
 )
 from .publisher import Publisher
+from .qpid.proton._message import Message
 
 try:
     __version__ = metadata.version(__package__)
@@ -29,5 +28,6 @@
     "QueueType",
     "Publisher",
     "exchange_address",
+    "queue_address",
     "Message",
 ]

rabbitmq_amqp_python_client/__init__.py

@@ -2,14 +2,6 @@
 import uuid
 from typing import Any, Optional
 
-from proton import Message
-from proton._data import Data
-from proton.utils import (
-    BlockingConnection,
-    BlockingReceiver,
-    BlockingSender,
-)
-
 from .address_helper import (
     binding_path_with_exchange_queue,
     exchange_address,
@@ -24,6 +16,12 @@
 )
 from .exceptions import ValidationCodeException
 from .options import ReceiverOption, SenderOption
+from .qpid.proton._message import Message
+from .qpid.proton.utils import (
+    BlockingConnection,
+    BlockingReceiver,
+    BlockingSender,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -154,14 +152,13 @@ def delete_exchange(self, exchange_name: str) -> None:
         logger.debug("delete_exchange operation called")
         path = exchange_address(exchange_name)
 
-        print(path)
-
         self.request(
-            Data.NULL,
+            None,
             path,
             CommonValues.command_delete.value,
             [
                 CommonValues.response_code_200.value,
+                CommonValues.response_code_204.value,
             ],
         )
 
@@ -175,6 +172,7 @@ def delete_queue(self, queue_name: str) -> None:
             CommonValues.command_delete.value,
             [
                 CommonValues.response_code_200.value,
+                CommonValues.response_code_204.value,
             ],
         )
 
@@ -226,6 +224,7 @@ def unbind(self, binding_exchange_queue_path: str) -> None:
             CommonValues.command_delete.value,
             [
                 CommonValues.response_code_200.value,
+                CommonValues.response_code_204.value,
             ],
         )
 

rabbitmq_amqp_python_client/management.py

@@ -1,6 +1,9 @@
-from proton._data import PropertyDict, symbol  # noqa: E402
-from proton._endpoints import Link  # noqa: E402
-from proton.reactor import LinkOption  # noqa: E402
+from .qpid.proton._data import (  # noqa: E402
+    PropertyDict,
+    symbol,
+)
+from .qpid.proton._endpoints import Link  # noqa: E402
+from .qpid.proton.reactor import LinkOption  # noqa: E402
 
 
 class SenderOption(LinkOption):  # type: ignore

rabbitmq_amqp_python_client/options.py

@@ -1,15 +1,14 @@
 import logging
 from typing import Optional
 
-from proton import Message
-from proton.utils import (
+from .options import SenderOption
+from .qpid.proton._message import Message
+from .qpid.proton.utils import (
     BlockingConnection,
     BlockingReceiver,
     BlockingSender,
 )
 
-from .options import SenderOption
-
 logger = logging.getLogger(__name__)
 
 
@@ -22,7 +21,6 @@ def __init__(self, conn: BlockingConnection, addr: str):
         self._open()
 
     def _open(self) -> None:
-        print("addr is " + str(self._addr))
         if self._sender is None:
             logger.debug("Creating Sender")
             self._sender = self._create_sender(self._addr)