Merge branch 'main' into ssl_implementation

Author: DanielePalaia

README.md

@@ -9,6 +9,13 @@ This library is in early stages of development. It is meant to be used with Rabb
 - poetry install: resolves and install dependencies
 - poetry run pytest: run the tests
 
+## Installation
+
+The client is distributed via [`PIP`](https://pypi.org/project/rabbitmq-amqp-python-client/):
+```bash
+ pip install rabbitmq-amqp-python-client
+```
+
 ## Getting Started
 
 An example is provided in ./getting_started_main.py you can run it after starting a RabbitMQ 4.0 broker with:

examples/getting_started/main.py

@@ -81,7 +81,7 @@ def main() -> None:
     exchange_name = "test-exchange"
     queue_name = "example-queue"
     routing_key = "routing-key"
-    messages_to_publish = 100
+    messages_to_publish = 100000
 
     print("connection to amqp server")
     connection = create_connection()
@@ -120,7 +120,13 @@ def main() -> None:
 
     # publish 10 messages
     for i in range(messages_to_publish):
-        publisher.publish(Message(body="test"))
+        status = publisher.publish(Message(body="test"))
+        if status.ACCEPTED:
+            print("message accepted")
+        elif status.RELEASED:
+            print("message not routed")
+        elif status.REJECTED:
+            print("message not rejected")
 
     publisher.close()
 

rabbitmq_amqp_python_client/__init__.py

@@ -58,4 +58,5 @@
     "ArgumentOutOfRangeException",
     "SslConfigurationContext",
     "ClientCert",
+    "Delivery",
 ]

rabbitmq_amqp_python_client/options.py

@@ -21,6 +21,21 @@ def test(self, link: Link) -> bool:
         return bool(link.is_sender)
 
 
+class SenderOptionUnseattle(LinkOption):  # type: ignore
+    def __init__(self, addr: str):
+        self._addr = addr
+
+    def apply(self, link: Link) -> None:
+        link.source.address = self._addr
+        link.snd_settle_mode = Link.SND_UNSETTLED
+        link.rcv_settle_mode = Link.RCV_FIRST
+        link.properties = PropertyDict({symbol("paired"): True})
+        link.source.dynamic = False
+
+    def test(self, link: Link) -> bool:
+        return bool(link.is_sender)
+
+
 class ReceiverOption(LinkOption):  # type: ignore
     def __init__(self, addr: str):
         self._addr = addr

rabbitmq_amqp_python_client/publisher.py

@@ -1,7 +1,8 @@
 import logging
 from typing import Optional
 
-from .options import SenderOption
+from .options import SenderOptionUnseattle
+from .qpid.proton._delivery import Delivery
 from .qpid.proton._message import Message
 from .qpid.proton.utils import (
     BlockingConnection,
@@ -23,14 +24,14 @@ def _open(self) -> None:
             logger.debug("Creating Sender")
             self._sender = self._create_sender(self._addr)
 
-    def publish(self, message: Message) -> None:
+    def publish(self, message: Message) -> Delivery:
         if self._sender is not None:
-            self._sender.send(message)
+            return self._sender.send(message)
 
     def close(self) -> None:
         logger.debug("Closing Sender")
         if self._sender is not None:
             self._sender.close()
 
     def _create_sender(self, addr: str) -> BlockingSender:
-        return self._conn.create_sender(addr, options=SenderOption(addr))
+        return self._conn.create_sender(addr, options=SenderOptionUnseattle(addr))

tests/test_publisher.py

@@ -18,17 +18,24 @@ def test_publish_queue(connection: Connection) -> None:
 
     raised = False
 
+    publisher = None
+    accepted = False
+
     try:
         publisher = connection.publisher("/queues/" + queue_name)
-        publisher.publish(Message(body="test"))
+        status = publisher.publish(Message(body="test"))
+        if status.ACCEPTED:
+            accepted = True
     except Exception:
         raised = True
 
-    publisher.close()
+    if publisher is not None:
+        publisher.close()
 
     management.delete_queue(queue_name)
     management.close()
 
+    assert accepted is True
     assert raised is False
 
 
@@ -98,10 +105,13 @@ def test_publish_exchange(connection: Connection) -> None:
     addr = AddressHelper.exchange_address(exchange_name, routing_key)
 
     raised = False
+    accepted = False
 
     try:
         publisher = connection.publisher(addr)
-        publisher.publish(Message(body="test"))
+        status = publisher.publish(Message(body="test"))
+        if status.ACCEPTED:
+            accepted = True
     except Exception:
         raised = True
 
@@ -111,6 +121,7 @@ def test_publish_exchange(connection: Connection) -> None:
     management.delete_queue(queue_name)
     management.close()
 
+    assert accepted is True
     assert raised is False