making Message a client class

Author: DanielePalaia

examples/getting_started/getting_started.py

@@ -3,13 +3,13 @@
 
 from rabbitmq_amqp_python_client import (  # SSlConfigurationContext,; SslConfigurationContext,; ClientCert,
     AddressHelper,
+    AmqpMessage,
     AMQPMessagingHandler,
     BindingSpecification,
     Connection,
     Environment,
     Event,
     ExchangeSpecification,
-    Message,
     OutcomeState,
     QuorumQueueSpecification,
 )
@@ -125,7 +125,7 @@ def main() -> None:
     # publish 10 messages
     for i in range(MESSAGES_TO_PUBLISH):
         print("publishing")
-        status = publisher.publish(Message(body="test"))
+        status = publisher.publish(AmqpMessage(body="test"))
         if status.remote_state == OutcomeState.ACCEPTED:
             print("message accepted")
         elif status.remote_state == OutcomeState.RELEASED:

examples/reconnection/reconnection_example.py

@@ -7,6 +7,7 @@
 
 from rabbitmq_amqp_python_client import (
     AddressHelper,
+    AmqpMessage,
     AMQPMessagingHandler,
     BindingSpecification,
     Connection,
@@ -16,7 +17,6 @@
     Event,
     ExchangeSpecification,
     Management,
-    Message,
     Publisher,
     QuorumQueueSpecification,
 )
@@ -123,7 +123,7 @@ def create_connection() -> Connection:
     # connection = Connection(uris=uris, on_disconnection_handler=on_disconnected)
 
     connection = environment.connection(
-        url="amqp://guest:guest@localhost:5672/",
+        uri="amqp://guest:guest@localhost:5672/",
         on_disconnection_handler=on_disconnection,
     )
     connection.dial()
@@ -191,7 +191,7 @@ def main() -> None:
                 print("published 1000 messages...")
             try:
                 if connection_configuration.publisher is not None:
-                    connection_configuration.publisher.publish(Message(body="test"))
+                    connection_configuration.publisher.publish(AmqpMessage(body="test"))
             except ConnectionClosed:
                 print("publisher closing exception, resubmitting")
                 continue

examples/streams/example_with_streams.py

@@ -2,11 +2,11 @@
 
 from rabbitmq_amqp_python_client import (  # SSlConfigurationContext,; SslConfigurationContext,; ClientCert,
     AddressHelper,
+    AmqpMessage,
     AMQPMessagingHandler,
     Connection,
     Environment,
     Event,
-    Message,
     OffsetSpecification,
     StreamOptions,
     StreamSpecification,
@@ -120,15 +120,15 @@ def main() -> None:
     # publish with a filter of apple
     for i in range(MESSAGES_TO_PUBLISH):
         publisher.publish(
-            Message(
+            AmqpMessage(
                 body="apple: " + str(i), annotations={"x-stream-filter-value": "apple"}
             )
         )
 
     # publish with a filter of banana
     for i in range(MESSAGES_TO_PUBLISH):
         publisher.publish(
-            Message(
+            AmqpMessage(
                 body="banana: " + str(i),
                 annotations={"x-stream-filter-value": "banana"},
             )

examples/tls/tls_example.py

@@ -3,14 +3,14 @@
 
 from rabbitmq_amqp_python_client import (  # SSlConfigurationContext,; SslConfigurationContext,; ClientCert,
     AddressHelper,
+    AmqpMessage,
     AMQPMessagingHandler,
     BindingSpecification,
     ClientCert,
     Connection,
     Environment,
     Event,
     ExchangeSpecification,
-    Message,
     QuorumQueueSpecification,
     SslConfigurationContext,
 )
@@ -124,7 +124,7 @@ def main() -> None:
 
     # publish 10 messages
     for i in range(messages_to_publish):
-        status = publisher.publish(Message(body="test"))
+        status = publisher.publish(AmqpMessage(body="test"))
         if status.ACCEPTED:
             print("message accepted")
         elif status.RELEASED:

rabbitmq_amqp_python_client/__init__.py

@@ -2,6 +2,7 @@
 
 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
@@ -18,7 +19,6 @@
 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",
-    "Message",
+    "AmqpMessage",
     "Consumer",
     "MessagingHandler",
     "Event",

rabbitmq_amqp_python_client/amqp_message.py

@@ -0,0 +1,29 @@
+from typing import Optional, Union, cast
+from uuid import UUID
+
+from proton._data import Described
+
+from .qpid.proton._message import Message
+
+
+class AmqpMessage(Message):  # type: ignore
+
+    def __init__(  # type: ignore
+        self,
+        body: Union[  # type: ignore
+            bytes, str, dict, list, int, float, "UUID", "Described", None
+        ] = None,
+        **kwargs,
+    ):
+        super().__init__(body=body, **kwargs)
+        self._addr: Optional[str] = None
+        self._native_message = None
+
+    def to_address(self, addr: str) -> None:
+        self._addr = addr
+
+    def address(self) -> Optional[str]:
+        return self._addr
+
+    def native_message(self) -> Message:
+        return cast(Message, self)

rabbitmq_amqp_python_client/consumer.py

@@ -1,13 +1,13 @@
 import logging
-from typing import Literal, Optional, Union
+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,9 +38,12 @@ def _open(self) -> None:
             logger.debug("Creating Sender")
             self._receiver = self._create_receiver(self._addr)
 
-    def consume(self, timeout: Union[None, Literal[False], float] = False) -> Message:
+    def consume(  # type: ignore
+        self, timeout: Union[None, Literal[False], float] = False
+    ) -> AmqpMessage:
         if self._receiver is not None:
-            return self._receiver.receive(timeout=timeout)
+            message = self._receiver.receive(timeout=timeout)
+            return cast(AmqpMessage, message)
 
     def close(self) -> None:
         logger.debug("Closing the receiver")

rabbitmq_amqp_python_client/publisher.py

@@ -1,9 +1,9 @@
 import logging
 from typing import Optional
 
+from .amqp_message import AmqpMessage
 from .options import SenderOptionUnseattle
 from .qpid.proton._delivery import Delivery
-from .qpid.proton._message import Message
 from .qpid.proton.utils import (
     BlockingConnection,
     BlockingSender,
@@ -13,7 +13,7 @@
 
 
 class Publisher:
-    def __init__(self, conn: BlockingConnection, addr: str):
+    def __init__(self, conn: BlockingConnection, addr: str = ""):
         self._sender: Optional[BlockingSender] = None
         self._conn = conn
         self._addr = addr
@@ -24,9 +24,9 @@ def _open(self) -> None:
             logger.debug("Creating Sender")
             self._sender = self._create_sender(self._addr)
 
-    def publish(self, message: Message) -> Delivery:
+    def publish(self, message: AmqpMessage) -> Delivery:
         if self._sender is not None:
-            return self._sender.send(message)
+            return self._sender.send(message.native_message())
 
     def close(self) -> None:
         logger.debug("Closing Sender")

tests/test_publisher.py

@@ -2,13 +2,13 @@
 
 from rabbitmq_amqp_python_client import (
     AddressHelper,
+    AmqpMessage,
     ArgumentOutOfRangeException,
     BindingSpecification,
     Connection,
     ConnectionClosed,
     Environment,
     ExchangeSpecification,
-    Message,
     QuorumQueueSpecification,
     StreamSpecification,
 )
@@ -30,7 +30,7 @@ def test_publish_queue(connection: Connection) -> None:
 
     try:
         publisher = connection.publisher("/queues/" + queue_name)
-        status = publisher.publish(Message(body="test"))
+        status = publisher.publish(AmqpMessage(body="test"))
         if status.ACCEPTED:
             accepted = True
     except Exception:
@@ -57,7 +57,7 @@ def test_publish_ssl(connection_ssl: Connection) -> None:
 
     try:
         publisher = connection_ssl.publisher("/queues/" + queue_name)
-        publisher.publish(Message(body="test"))
+        publisher.publish(AmqpMessage(body="test"))
     except Exception:
         raised = True
 
@@ -78,7 +78,7 @@ def test_publish_to_invalid_destination(connection: Connection) -> None:
     publisher = None
     try:
         publisher = connection.publisher("/invalid-destination/" + queue_name)
-        publisher.publish(Message(body="test"))
+        publisher.publish(AmqpMessage(body="test"))
     except ArgumentOutOfRangeException:
         raised = True
     except Exception:
@@ -116,7 +116,7 @@ def test_publish_exchange(connection: Connection) -> None:
 
     try:
         publisher = connection.publisher(addr)
-        status = publisher.publish(Message(body="test"))
+        status = publisher.publish(AmqpMessage(body="test"))
         if status.ACCEPTED:
             accepted = True
     except Exception:
@@ -145,7 +145,7 @@ def test_publish_purge(connection: Connection) -> None:
     try:
         publisher = connection.publisher("/queues/" + queue_name)
         for i in range(messages_to_publish):
-            publisher.publish(Message(body="test"))
+            publisher.publish(AmqpMessage(body="test"))
     except Exception:
         raised = True
 
@@ -210,7 +210,7 @@ def on_disconnected():
                 # simulate a disconnection
                 delete_all_connections()
             try:
-                publisher.publish(Message(body="test"))
+                publisher.publish(AmqpMessage(body="test"))
 
             except ConnectionClosed:
                 disconnected = True
@@ -264,4 +264,4 @@ def test_queue_info_for_stream_with_validations(connection: Connection) -> None:
 
     for i in range(messages_to_send):
 
-        publisher.publish(Message(body="test"))
+        publisher.publish(AmqpMessage(body="test"))

tests/utils.py

@@ -1,12 +1,12 @@
 from typing import Optional
 
 from rabbitmq_amqp_python_client import (
+    AmqpMessage,
     BindingSpecification,
     Connection,
     ExchangeSpecification,
     ExchangeType,
     Management,
-    Message,
     QuorumQueueSpecification,
 )
 
@@ -25,7 +25,7 @@ def publish_messages(
     publisher = connection.publisher("/queues/" + queue_name)
     # publish messages_to_send messages
     for i in range(messages_to_send):
-        publisher.publish(Message(body="test" + str(i), annotations=annotations))
+        publisher.publish(AmqpMessage(body="test" + str(i), annotations=annotations))
     publisher.close()