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

@@ -4,16 +4,18 @@
     AddressHelper,
     AmqpMessage,
     ArgumentOutOfRangeException,
-    BindingSpecification,
     Connection,
     ConnectionClosed,
     Environment,
-    ExchangeSpecification,
+    OutcomeState,
     QuorumQueueSpecification,
     StreamSpecification,
+    BindingSpecification,
+    ExchangeSpecification,
 )
 
 from .http_requests import delete_all_connections
+from .utils import create_binding, publish_per_message
 
 
 def test_publish_queue(connection: Connection) -> None:
@@ -31,7 +33,7 @@ def test_publish_queue(connection: Connection) -> None:
     try:
         publisher = connection.publisher("/queues/" + queue_name)
         status = publisher.publish(AmqpMessage(body="test"))
-        if status.ACCEPTED:
+        if status.remote_state == OutcomeState.ACCEPTED:
             accepted = True
     except Exception:
         raised = True
@@ -46,6 +48,107 @@ 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, addr=AddressHelper.queue_address(queue_name)
+        )
+        if status.remote_state == OutcomeState.ACCEPTED:
+            accepted = True
+        status = publish_per_message(
+            publisher, addr=AddressHelper.queue_address(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_per_message_to_exchange(connection: Connection) -> None:
+
+    exchange_name = "test-exchange"
+    queue_name = "test-queue"
+    management = connection.management()
+    routing_key = "routing-key"
+
+    management.declare_exchange(ExchangeSpecification(name=exchange_name))
+
+    management.declare_queue(QuorumQueueSpecification(name=queue_name))
+
+    bind_path = management.bind(
+        BindingSpecification(
+            source_exchange=exchange_name,
+            destination_queue=queue_name,
+            binding_key=routing_key,
+        )
+    )
+
+    raised = False
+
+    publisher = None
+    # accepted = False
+    accepted_2 = True
+
+    try:
+        publisher = connection.publisher()
+        # status = publish_per_message(
+        #    publisher, addr=AddressHelper.exchange_address(exchange_name, routing_key)
+        # )
+        # if status.remote_state == OutcomeState.ACCEPTED:
+        #    accepted = True
+        status = publish_per_message(
+            publisher, addr=AddressHelper.queue_address(queue_name)
+        )
+        if status.remote_state == OutcomeState.ACCEPTED:
+            accepted_2 = True
+    except Exception:
+        raised = True
+
+    # if publisher is not None:
+    publisher.close()
+
+    # purged_messages_queue = management.purge_queue(queue_name)
+    management.unbind(bind_path)
+    management.delete_exchange(exchange_name)
+    management.delete_queue(queue_name)
+    management.close()
+
+    # assert accepted is True
+    assert accepted_2 is True
+    # assert purged_messages_queue == 1
+    assert raised is False
+
+
 def test_publish_ssl(connection_ssl: Connection) -> None:
 
     queue_name = "test-queue"
@@ -90,24 +193,36 @@ def test_publish_to_invalid_destination(connection: Connection) -> None:
     assert raised is True
 
 
+def test_publish_per_message_to_invalid_destination(connection: Connection) -> None:
+
+    queue_name = "test-queue-1"
+    raised = False
+
+    message = AmqpMessage(body="test")
+    message.to_address("/invalid_destination/" + queue_name)
+    publisher = connection.publisher()
+
+    try:
+        publisher.publish(message)
+    except ArgumentOutOfRangeException:
+        raised = True
+    except Exception:
+        raised = False
+
+    if publisher is not None:
+        publisher.close()
+
+    assert raised is True
+
+
 def test_publish_exchange(connection: Connection) -> None:
 
     exchange_name = "test-exchange"
     queue_name = "test-queue"
     management = connection.management()
     routing_key = "routing-key"
 
-    management.declare_exchange(ExchangeSpecification(name=exchange_name))
-
-    management.declare_queue(QuorumQueueSpecification(name=queue_name))
-
-    management.bind(
-        BindingSpecification(
-            source_exchange=exchange_name,
-            destination_queue=queue_name,
-            binding_key=routing_key,
-        )
-    )
+    create_binding(management, exchange_name, queue_name, routing_key)
 
     addr = AddressHelper.exchange_address(exchange_name, routing_key)
 

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, addr: str) -> Delivery:
+    message = AmqpMessage(body="test")
+    message.to_address(addr)
+    status = publisher.publish(message)
+    return status
+
+
 def setup_dead_lettering(management: Management) -> str:
 
     exchange_dead_lettering = "exchange-dead-letter"
@@ -55,6 +64,25 @@ def setup_dead_lettering(management: Management) -> str:
     return bind_path
 
 
+def create_binding(
+    management: Management, exchange_name: str, queue_name: str, routing_key: str
+) -> str:
+
+    management.declare_exchange(ExchangeSpecification(name=exchange_name))
+
+    management.declare_queue(QuorumQueueSpecification(name=queue_name))
+
+    bind_name = management.bind(
+        BindingSpecification(
+            source_exchange=exchange_name,
+            destination_queue=queue_name,
+            binding_key=routing_key,
+        )
+    )
+
+    return bind_name
+
+
 def cleanup_dead_lettering(management: Management, bind_path: str) -> None:
 
     exchange_dead_lettering = "exchange-dead-letter"