adding delivery status on publishing

Author: DanielePalaia

examples/getting_started/main.py

@@ -109,7 +109,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")
+        if status.RELEASED:
+            print("message not routed")
+        if status.REJECTED:
+            print("message not rejected")
 
     publisher.close()
 

rabbitmq_amqp_python_client/__init__.py

@@ -52,4 +52,5 @@
     "AddressHelper",
     "AMQPMessagingHandler",
     "ArgumentOutOfRangeException",
+    "Delivery",
 ]

rabbitmq_amqp_python_client/publisher.py

@@ -2,6 +2,7 @@
 from typing import Optional
 
 from .options import SenderOption
+from .qpid.proton._delivery import Delivery
 from .qpid.proton._message import Message
 from .qpid.proton.utils import (
     BlockingConnection,
@@ -23,9 +24,9 @@ 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")

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
 
 
@@ -75,10 +82,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
 
@@ -88,6 +98,7 @@ def test_publish_exchange(connection: Connection) -> None:
     management.delete_queue(queue_name)
     management.close()
 
+    assert accepted is True
     assert raised is False