rpc example useing direct reply queue

Signed-off-by: Gabriele Santomaggio <[email protected]>

Author: Gsantomaggio

examples/rpc/client.py

@@ -13,8 +13,12 @@ class Requester:
     def __init__(self, request_queue_name: str, environment: Environment):
         self.connection = environment.connection()
         self.connection.dial()
-        self.publisher = self.connection.publisher(AddressHelper.queue_address(request_queue_name))
-        self.consumer = self.connection.consumer(consumer_options=DirectReplyToConsumerOptions())
+        self.publisher = self.connection.publisher(
+            AddressHelper.queue_address(request_queue_name)
+        )
+        self.consumer = self.connection.consumer(
+            consumer_options=DirectReplyToConsumerOptions()
+        )
         print("connected both publisher and consumer")
         print("consumer reply address is {}".format(self.consumer.address))
 
@@ -35,9 +39,15 @@ def main() -> None:
         request_body = "hello {}".format(i)
         print("******************************************************")
         print("Sending request: {}".format(request_body))
-        response_message = responder.send_request(request_body=request_body, correlation_id=correlation_id)
+        response_message = responder.send_request(
+            request_body=request_body, correlation_id=correlation_id
+        )
         response_body = Converter.bytes_to_string(response_message.body)
-        print("Received response: {} - correlation_id: {}".format(response_body, response_message.correlation_id))
+        print(
+            "Received response: {} - correlation_id: {}".format(
+                response_body, response_message.correlation_id
+            )
+        )
         print("------------------------------------------------------")
         time.sleep(1)
 

examples/rpc/server.py

@@ -18,6 +18,7 @@
 
 # create a responder
 
+
 class Responder:
     class ResponderMessageHandler(AMQPMessagingHandler):
 
@@ -31,18 +32,20 @@ def set_publisher(self, publisher):
         def on_amqp_message(self, event: Event):
             # process the message and create a response
             print("******************************************************")
-            print("received message: {} ".format(Converter.bytes_to_string(event.message.body)))
-            response_body = Converter.bytes_to_string(
-                event.message.body) + "-from the server"
-            response_message = Message(
-                body=Converter.string_to_bytes(response_body))
+            print(
+                "received message: {} ".format(
+                    Converter.bytes_to_string(event.message.body)
+                )
+            )
+            response_body = (
+                Converter.bytes_to_string(event.message.body) + "-from the server"
+            )
+            response_message = Message(body=Converter.string_to_bytes(response_body))
             # publish response to the reply_to address with the same correlation_id
             response_message.correlation_id = event.message.correlation_id
             response_message.address = event.message.reply_to
             print("sending back: {} ".format(response_body))
-            status = self._publisher.publish(
-                message=response_message
-            )
+            status = self._publisher.publish(message=response_message)
             if status.remote_state == OutcomeState.ACCEPTED:
                 print("message accepted to {}".format(response_message.address))
             elif status.remote_state == OutcomeState.RELEASED:
@@ -65,14 +68,16 @@ def start(self):
         self.connection.dial()
         self.connection.management().delete_queue(self.request_queue_name)
         self.connection.management().declare_queue(
-            queue_specification=QuorumQueueSpecification(self.request_queue_name))
+            queue_specification=QuorumQueueSpecification(self.request_queue_name)
+        )
         self.publisher = self.connection.publisher()
         handler = self.ResponderMessageHandler()
         handler.set_publisher(self.publisher)
 
-        self.consumer = self.connection.consumer(destination=AddressHelper.queue_address(self.request_queue_name),
-                                                 message_handler=handler
-                                                 )
+        self.consumer = self.connection.consumer(
+            destination=AddressHelper.queue_address(self.request_queue_name),
+            message_handler=handler,
+        )
         addr = self.consumer.address
         print("Responder listening on address: {}".format(addr))
         try: