improvements

Author: DanielePalaia

rabbitmq_amqp_python_client/amqp_message.py

@@ -1,4 +1,4 @@
-from typing import Optional, Union, cast
+from typing import Union, cast
 from uuid import UUID
 
 from proton._data import Described
@@ -16,13 +16,14 @@ def __init__(  # type: ignore
         **kwargs,
     ):
         super().__init__(body=body, **kwargs)
-        self._addr: Optional[str] = None
+        self._addr: str = ""
         self._native_message = None
 
     def to_address(self, addr: str) -> None:
+        self.address = addr
         self._addr = addr
 
-    def address(self) -> Optional[str]:
+    def get_address(self) -> str:
         return self._addr
 
     def native_message(self) -> Message:

rabbitmq_amqp_python_client/connection.py

@@ -82,11 +82,12 @@ def close(self) -> None:
         self._conn.close()
         self._connections.remove(self)
 
-    def publisher(self, destination: str) -> Publisher:
-        if validate_address(destination) is False:
-            raise ArgumentOutOfRangeException(
-                "destination address must start with /queues or /exchanges"
-            )
+    def publisher(self, destination: str = "") -> Publisher:
+        if destination != "":
+            if validate_address(destination) is False:
+                raise ArgumentOutOfRangeException(
+                    "destination address must start with /queues or /exchanges"
+                )
         publisher = Publisher(self._conn, destination)
         return publisher
 

rabbitmq_amqp_python_client/publisher.py

@@ -1,7 +1,9 @@
 import logging
 from typing import Optional
 
+from .address_helper import validate_address
 from .amqp_message import AmqpMessage
+from .exceptions import ArgumentOutOfRangeException
 from .options import SenderOptionUnseattle
 from .qpid.proton._delivery import Delivery
 from .qpid.proton.utils import (
@@ -25,8 +27,18 @@ def _open(self) -> None:
             self._sender = self._create_sender(self._addr)
 
     def publish(self, message: AmqpMessage) -> Delivery:
-        if self._sender is not None:
-            return self._sender.send(message.native_message())
+        if self._addr != "":
+            if self._sender is not None:
+                return self._sender.send(message.native_message())
+        else:
+            if message.get_address() != "":
+                if validate_address(message.get_address()) is False:
+                    raise ArgumentOutOfRangeException(
+                        "destination address must start with /queues or /exchanges"
+                    )
+                if self._sender is not None:
+                    delivery = self._sender.send(message.native_message())
+                    return delivery
 
     def close(self) -> None:
         logger.debug("Closing Sender")

rabbitmq_amqp_python_client/qpid/proton/_message.py

@@ -140,7 +140,6 @@ def _check(self, err: int) -> int:
     def _check_property_keys(self) -> None:
         """
         AMQP allows only string keys for properties. This function checks that this requirement is met
-        and raises a MessageException if not. However, in certain cases, conversions to string are
         automatically performed:
 
         1. When a key is a user-defined (non-AMQP) subclass of str.

tests/test_publisher.py

@@ -9,11 +9,13 @@
     ConnectionClosed,
     Environment,
     ExchangeSpecification,
+    OutcomeState,
     QuorumQueueSpecification,
     StreamSpecification,
 )
 
 from .http_requests import delete_all_connections
+from .utils import publish_per_message
 
 
 def test_publish_queue(connection: Connection) -> None:
@@ -46,6 +48,48 @@ def test_publish_queue(connection: Connection) -> None:
     assert raised is False
 
 
+def test_publish_per_message(connection: Connection) -> None:
+
+    queue_name = "test-queue-1"
+    queue_name_2 = "test-queue-2"
+    management = connection.management()
+
+    management.declare_queue(QuorumQueueSpecification(name=queue_name))
+    management.declare_queue(QuorumQueueSpecification(name=queue_name_2))
+
+    raised = False
+
+    publisher = None
+    accepted = False
+    accepted_2 = True
+
+    try:
+        publisher = connection.publisher()
+        status = publish_per_message(publisher, queue_name)
+        if status.remote_state == OutcomeState.ACCEPTED:
+            accepted = True
+        status = publish_per_message(publisher, queue_name_2)
+        if status.remote_state == OutcomeState.ACCEPTED:
+            accepted_2 = True
+    except Exception:
+        raised = True
+
+    if publisher is not None:
+        publisher.close()
+
+    purged_messages_queue_1 = management.purge_queue(queue_name)
+    purged_messages_queue_2 = management.purge_queue(queue_name_2)
+    management.delete_queue(queue_name)
+    management.delete_queue(queue_name_2)
+    management.close()
+
+    assert accepted is True
+    assert accepted_2 is True
+    assert purged_messages_queue_1 == 1
+    assert purged_messages_queue_2 == 1
+    assert raised is False
+
+
 def test_publish_ssl(connection_ssl: Connection) -> None:
 
     queue_name = "test-queue"

tests/utils.py

@@ -4,9 +4,11 @@
     AmqpMessage,
     BindingSpecification,
     Connection,
+    Delivery,
     ExchangeSpecification,
     ExchangeType,
     Management,
+    Publisher,
     QuorumQueueSpecification,
 )
 
@@ -29,6 +31,13 @@ def publish_messages(
     publisher.close()
 
 
+def publish_per_message(publisher: Publisher, queue_name) -> Delivery:
+    message = AmqpMessage(body="test")
+    message.to_address("/queues/" + queue_name)
+    status = publisher.publish(message)
+    return status
+
+
 def setup_dead_lettering(management: Management) -> str:
 
     exchange_dead_lettering = "exchange-dead-letter"