refact

Author: Gsantomaggio

examples/oauth/oaut.py

@@ -18,6 +18,7 @@
     OAuth2Options,
     OutcomeState,
     QuorumQueueSpecification,
+    Converter
 )
 
 MESSAGES_TO_PUBLISH = 100
@@ -30,7 +31,7 @@ def __init__(self):
         self._count = 0
 
     def on_amqp_message(self, event: Event):
-        print("received message: " + str(event.message.body))
+        print("received message: " + Converter.bytes_to_string(event.message.body))
 
         # accepting
         self.delivery_context.accept(event)
@@ -85,7 +86,6 @@ def create_connection(environment: Environment) -> Connection:
 
 
 def main() -> None:
-
     exchange_name = "test-exchange"
     queue_name = "example-queue"
     routing_key = "routing-key"
@@ -144,14 +144,13 @@ def main() -> None:
 
     # publish 10 messages
     for i in range(MESSAGES_TO_PUBLISH):
-        print("publishing")
-        status = publisher.publish(Message(body="test"))
+        status = publisher.publish(Message(body=Converter.string_to_bytes("test_{}".format(i))))
         if status.remote_state == OutcomeState.ACCEPTED:
-            print("message accepted")
+            print("message: test_{} accepted".format(i))
         elif status.remote_state == OutcomeState.RELEASED:
-            print("message not routed")
+            print("message: test_{} not routed".format(i))
         elif status.remote_state == OutcomeState.REJECTED:
-            print("message not rejected")
+            print("message: test_{} rejected".format(i))
 
     publisher.close()
 

examples/reconnection/reconnection_example.py

@@ -9,13 +9,12 @@
     ExchangeSpecification,
     ExchangeToQueueBindingSpecification,
     Message,
-    QuorumQueueSpecification,
+    QuorumQueueSpecification, Converter,
 )
 
 # here we keep track of the objects we need to reconnect
 MESSAGES_TO_PUBLISH = 50000
 
-
 environment = Environment(
     uri="amqp://guest:guest@localhost:5672/",
 )
@@ -29,7 +28,7 @@ def __init__(self):
 
     def on_message(self, event: Event):
         if self._count % 1000 == 0:
-            print("received 100 message: " + str(event.message.body))
+            print("received 100 message: " + Converter.bytes_to_string(event.message.body))
 
         # accepting
         self.delivery_context.accept(event)
@@ -79,7 +78,6 @@ def create_connection() -> Connection:
 
 
 def main() -> None:
-
     exchange_name = "test-exchange"
     queue_name = "example-queue"
     routing_key = "routing-key"
@@ -128,7 +126,7 @@ def main() -> None:
                 print("published 1000 messages...")
             try:
                 if publisher is not None:
-                    publisher.publish(Message(body="test"))
+                    publisher.publish(Message(body=Converter.string_to_bytes("test")))
             except ConnectionClosed:
                 print("publisher closing exception, resubmitting")
                 # publisher = connection.publisher(addr)

rabbitmq_amqp_python_client/management.py

@@ -98,11 +98,11 @@ def close(self) -> None:
             self._receiver.close()
 
     def request(
-        self,
-        body: Any,
-        path: str,
-        method: str,
-        expected_response_codes: list[int],
+            self,
+            body: Any,
+            path: str,
+            method: str,
+            expected_response_codes: list[int],
     ) -> Message:
         """
         Send a management request with a new UUID.
@@ -124,16 +124,17 @@ def request(
         )
 
     def _request(
-        self,
-        id: str,
-        body: Any,
-        path: str,
-        method: str,
-        expected_response_codes: list[int],
+            self,
+            id: str,
+            body: Any,
+            path: str,
+            method: str,
+            expected_response_codes: list[int],
     ) -> Message:
         amq_message = Message(
             id=id,
             body=body,
+            inferred=False,
             reply_to="$me",
             address=path,
             subject=method,
@@ -151,10 +152,10 @@ def _request(
         return msg
 
     def declare_exchange(
-        self,
-        exchange_specification: Union[
-            ExchangeSpecification, ExchangeCustomSpecification
-        ],
+            self,
+            exchange_specification: Union[
+                ExchangeSpecification, ExchangeCustomSpecification
+            ],
     ) -> Union[ExchangeSpecification, ExchangeCustomSpecification]:
         """
         Declare a new exchange in RabbitMQ.
@@ -195,10 +196,10 @@ def declare_exchange(
         return exchange_specification
 
     def declare_queue(
-        self,
-        queue_specification: Union[
-            ClassicQueueSpecification, QuorumQueueSpecification, StreamSpecification
-        ],
+            self,
+            queue_specification: Union[
+                ClassicQueueSpecification, QuorumQueueSpecification, StreamSpecification
+            ],
     ) -> Union[
         ClassicQueueSpecification, QuorumQueueSpecification, StreamSpecification
     ]:
@@ -219,7 +220,7 @@ def declare_queue(
         logger.debug("declare_queue operation called")
 
         if isinstance(queue_specification, ClassicQueueSpecification) or isinstance(
-            queue_specification, QuorumQueueSpecification
+                queue_specification, QuorumQueueSpecification
         ):
             body = self._declare_queue(queue_specification)
 
@@ -242,8 +243,8 @@ def declare_queue(
         return queue_specification
 
     def _declare_queue(
-        self,
-        queue_specification: Union[ClassicQueueSpecification, QuorumQueueSpecification],
+            self,
+            queue_specification: Union[ClassicQueueSpecification, QuorumQueueSpecification],
     ) -> dict[str, Any]:
 
         body = {}
@@ -311,7 +312,7 @@ def _declare_queue(
         return body
 
     def _declare_stream(
-        self, stream_specification: StreamSpecification
+            self, stream_specification: StreamSpecification
     ) -> dict[str, Any]:
 
         body = {}
@@ -324,7 +325,7 @@ def _declare_stream(
 
         if stream_specification.max_age is not None:
             args["x-max-age"] = (
-                str(int(stream_specification.max_age.total_seconds())) + "s"
+                    str(int(stream_specification.max_age.total_seconds())) + "s"
             )
 
         if stream_specification.stream_max_segment_size_bytes is not None:
@@ -392,7 +393,7 @@ def delete_queue(self, name: str) -> None:
         )
 
     def _validate_reponse_code(
-        self, response_code: int, expected_response_codes: list[int]
+            self, response_code: int, expected_response_codes: list[int]
     ) -> None:
         if response_code == CommonValues.response_code_409.value:
             raise ValidationCodeException("ErrPreconditionFailed")
@@ -406,10 +407,10 @@ def _validate_reponse_code(
         )
 
     def bind(
-        self,
-        bind_specification: Union[
-            ExchangeToQueueBindingSpecification, ExchangeToExchangeBindingSpecification
-        ],
+            self,
+            bind_specification: Union[
+                ExchangeToQueueBindingSpecification, ExchangeToExchangeBindingSpecification
+            ],
     ) -> str:
         """
         Create a binding between exchanges or between an exchange and a queue.
@@ -462,12 +463,12 @@ def bind(
         return binding_path
 
     def unbind(
-        self,
-        bind_specification: Union[
-            str,
-            ExchangeToQueueBindingSpecification,
-            ExchangeToExchangeBindingSpecification,
-        ],
+            self,
+            bind_specification: Union[
+                str,
+                ExchangeToQueueBindingSpecification,
+                ExchangeToExchangeBindingSpecification,
+            ],
     ) -> None:
         """
         Remove a binding between exchanges or between an exchange and a queue.

rabbitmq_amqp_python_client/publisher.py

@@ -80,6 +80,16 @@ def publish(self, message: Message) -> Delivery:
                 "address specified in both message and publisher"
             )
 
+        if not isinstance(message.body, (bytes, type(None))):
+            raise ArgumentOutOfRangeException(
+                "Message body must be of type bytes or None"
+            )
+
+        if not message.inferred:
+            raise ArgumentOutOfRangeException(
+                "Message inferred must be True"
+            )
+
         if self._addr != "":
             if self._sender is not None:
                 return self._sender.send(message)

rabbitmq_amqp_python_client/qpid/proton/_message.py

@@ -113,20 +113,16 @@ class Message(object):
     DEFAULT_PRIORITY = PN_DEFAULT_PRIORITY
     """ Default AMQP message priority"""
 
-    def __init__(self, body: Union[bytes, dict, None] = None, **kwargs) -> None:
+    def __init__(self, body: Union[str, bytes, dict, None] = None, inferred = True, **kwargs) -> None:
         # validate the types
 
-        if not isinstance(body, (bytes, dict, type(None))):
-            raise ArgumentOutOfRangeException(
-                "Message body must be of type bytes, dict or None"
-            )
-
         self._msg = pn_message()
         self.instructions = None
         self.annotations = None
         self.properties = None
         self.body = body
-        self.inferred = True
+        self.inferred = inferred
+
 
         for k, v in kwargs.items():
             getattr(self, k)  # Raise exception if it's not a valid attribute.