removing amqpmessage wrapper layer

Author: DanielePalaia

rabbitmq_amqp_python_client/__init__.py

@@ -2,7 +2,6 @@
 
 from .address_helper import AddressHelper
 from .amqp_consumer_handler import AMQPMessagingHandler
-from .amqp_message import AmqpMessage
 from .common import ExchangeType, QueueType
 from .connection import Connection
 from .consumer import Consumer
@@ -19,6 +18,7 @@
 from .qpid.proton._data import symbol  # noqa: E402
 from .qpid.proton._delivery import Delivery, Disposition
 from .qpid.proton._events import Event
+from .qpid.proton._message import Message
 from .qpid.proton._utils import ConnectionClosed
 from .qpid.proton.handlers import MessagingHandler
 from .queues import (
@@ -52,7 +52,7 @@
     "BindingSpecification",
     "QueueType",
     "Publisher",
-    "AmqpMessage",
+    "Message",
     "Consumer",
     "MessagingHandler",
     "Event",
@@ -69,4 +69,5 @@
     "OffsetSpecification",
     "OutcomeState",
     "Environment",
+    "message_to_address_helper",
 ]

rabbitmq_amqp_python_client/address_helper.py

@@ -1,4 +1,5 @@
 from .entities import BindingSpecification
+from .qpid.proton._message import Message
 
 
 def _is_unreserved(char: str) -> bool:
@@ -73,6 +74,11 @@ def binding_path_with_exchange_queue(
         )
         return binding_path_wth_exchange_queue_key
 
+    @staticmethod
+    def message_to_address_helper(message: Message, address: str) -> Message:
+        message.address = address
+        return message
+
 
 def validate_address(address: str) -> bool:
     if address.startswith("/queues") or address.startswith("/exchanges"):

rabbitmq_amqp_python_client/amqp_message.py

@@ -1,13 +1,13 @@
 import logging
 from typing import Literal, Optional, Union, cast
 
-from .amqp_message import AmqpMessage
 from .entities import StreamOptions
 from .options import (
     ReceiverOptionUnsettled,
     ReceiverOptionUnsettledWithFilters,
 )
 from .qpid.proton._handlers import MessagingHandler
+from .qpid.proton._message import Message
 from .qpid.proton.utils import (
     BlockingConnection,
     BlockingReceiver,
@@ -38,12 +38,10 @@ def _open(self) -> None:
             logger.debug("Creating Sender")
             self._receiver = self._create_receiver(self._addr)
 
-    def consume(  # type: ignore
-        self, timeout: Union[None, Literal[False], float] = False
-    ) -> AmqpMessage:
+    def consume(self, timeout: Union[None, Literal[False], float] = False) -> Message:
         if self._receiver is not None:
             message = self._receiver.receive(timeout=timeout)
-            return cast(AmqpMessage, message)
+            return cast(Message, message)
 
     def close(self) -> None:
         logger.debug("Closing the receiver")

rabbitmq_amqp_python_client/consumer.py

@@ -2,10 +2,10 @@
 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._message import Message
 from .qpid.proton.utils import (
     BlockingConnection,
     BlockingSender,
@@ -26,18 +26,18 @@ def _open(self) -> None:
             logger.debug("Creating Sender")
             self._sender = self._create_sender(self._addr)
 
-    def publish(self, message: AmqpMessage) -> Delivery:
+    def publish(self, message: Message) -> Delivery:
         if self._addr != "":
             if self._sender is not None:
-                return self._sender.send(message.native_message())
+                return self._sender.send(message)
         else:
-            if message.get_address() != "":
-                if validate_address(message.get_address()) is False:
+            if message.address != "":
+                if validate_address(message.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())
+                    delivery = self._sender.send(message)
                     return delivery
 
     def close(self) -> None:

rabbitmq_amqp_python_client/publisher.py

@@ -2,11 +2,11 @@
 
 from rabbitmq_amqp_python_client import (
     AddressHelper,
-    AmqpMessage,
     ArgumentOutOfRangeException,
     Connection,
     ConnectionClosed,
     Environment,
+    Message,
     OutcomeState,
     QuorumQueueSpecification,
     StreamSpecification,
@@ -30,7 +30,7 @@ def test_publish_queue(connection: Connection) -> None:
 
     try:
         publisher = connection.publisher("/queues/" + queue_name)
-        status = publisher.publish(AmqpMessage(body="test"))
+        status = publisher.publish(Message(body="test"))
         if status.remote_state == OutcomeState.ACCEPTED:
             accepted = True
     except Exception:
@@ -103,7 +103,7 @@ def test_publish_ssl(connection_ssl: Connection) -> None:
 
     try:
         publisher = connection_ssl.publisher("/queues/" + queue_name)
-        publisher.publish(AmqpMessage(body="test"))
+        publisher.publish(Message(body="test"))
     except Exception:
         raised = True
 
@@ -124,7 +124,7 @@ def test_publish_to_invalid_destination(connection: Connection) -> None:
     publisher = None
     try:
         publisher = connection.publisher("/invalid-destination/" + queue_name)
-        publisher.publish(AmqpMessage(body="test"))
+        publisher.publish(Message(body="test"))
     except ArgumentOutOfRangeException:
         raised = True
     except Exception:
@@ -141,8 +141,10 @@ def test_publish_per_message_to_invalid_destination(connection: Connection) -> N
     queue_name = "test-queue-1"
     raised = False
 
-    message = AmqpMessage(body="test")
-    message.to_address("/invalid_destination/" + queue_name)
+    message = Message(body="test")
+    message = AddressHelper.message_to_address_helper(
+        message, "/invalid_destination/" + queue_name
+    )
     publisher = connection.publisher()
 
     try:
@@ -174,7 +176,7 @@ def test_publish_exchange(connection: Connection) -> None:
 
     try:
         publisher = connection.publisher(addr)
-        status = publisher.publish(AmqpMessage(body="test"))
+        status = publisher.publish(Message(body="test"))
         if status.ACCEPTED:
             accepted = True
     except Exception:
@@ -204,7 +206,7 @@ def test_publish_purge(connection: Connection) -> None:
     try:
         publisher = connection.publisher("/queues/" + queue_name)
         for i in range(messages_to_publish):
-            publisher.publish(AmqpMessage(body="test"))
+            publisher.publish(Message(body="test"))
     except Exception:
         raised = True
 
@@ -269,7 +271,7 @@ def on_disconnected():
                 # simulate a disconnection
                 delete_all_connections()
             try:
-                publisher.publish(AmqpMessage(body="test"))
+                publisher.publish(Message(body="test"))
 
             except ConnectionClosed:
                 disconnected = True
@@ -323,9 +325,9 @@ def test_queue_info_for_stream_with_validations(connection: Connection) -> None:
 
     for i in range(messages_to_send):
 
-        publisher.publish(AmqpMessage(body="test"))
+        publisher.publish(Message(body="test"))
+
 
-'''
 def test_publish_per_message_exchange(connection: Connection) -> None:
 
     exchange_name = "test-exchange-per-message"
@@ -338,16 +340,16 @@ def test_publish_per_message_exchange(connection: Connection) -> None:
     raised = False
 
     publisher = None
-    # accepted = False
+    accepted = False
     accepted_2 = False
 
     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.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)
         )
@@ -366,8 +368,7 @@ def test_publish_per_message_exchange(connection: Connection) -> None:
 
     management.close()
 
-    # assert accepted is True
+    assert accepted is True
     assert accepted_2 is True
-    assert purged_messages_queue == 1
+    assert purged_messages_queue == 2
     assert raised is False
-'''

tests/test_publisher.py

@@ -1,13 +1,14 @@
 from typing import Optional
 
 from rabbitmq_amqp_python_client import (
-    AmqpMessage,
+    AddressHelper,
     BindingSpecification,
     Connection,
     Delivery,
     ExchangeSpecification,
     ExchangeType,
     Management,
+    Message,
     Publisher,
     QuorumQueueSpecification,
 )
@@ -27,13 +28,13 @@ def publish_messages(
     publisher = connection.publisher("/queues/" + queue_name)
     # publish messages_to_send messages
     for i in range(messages_to_send):
-        publisher.publish(AmqpMessage(body="test" + str(i), annotations=annotations))
+        publisher.publish(Message(body="test" + str(i), annotations=annotations))
     publisher.close()
 
 
 def publish_per_message(publisher: Publisher, addr: str) -> Delivery:
-    message = AmqpMessage(body="test")
-    message.to_address(addr)
+    message = Message(body="test")
+    message = AddressHelper.message_to_address_helper(message, addr)
     status = publisher.publish(message)
     return status