Merge pull request #2 from OpenMatchmaking/feature-retry-send-request

 Implemented sending the message with retries and timeouts

Author: Relrin

sage_utils/__init__.py

@@ -1,4 +1,4 @@
 __title__ = 'sage-utils'
-__version__ = '0.3.1'
+__version__ = '0.4.0'
 __license__ = 'BSD'
 VERSION = __version__

sage_utils/amqp/clients.py

@@ -30,7 +30,7 @@ def __init__(self, app, routing_key, request_exchange='',
     def response_queue_name(self):
         return self._response_queue_name
 
-    async def connect(self):
+    async def connect(self, consume_timeout=None):
         self.transport, self.protocol = await self.app.amqp.connect()
         self.channel = await self.protocol.channel()
 
@@ -53,18 +53,22 @@ async def connect(self):
                 prefetch_size=0,
                 connection_global=False
             )
-            await self.channel.basic_consume(
-                self.on_response,
-                queue_name=self._response_queue_name,
+            await asyncio.wait_for(
+                self.channel.basic_consume(
+                    self.on_response,
+                    queue_name=self._response_queue_name,
+                ),
+                timeout=consume_timeout,
+                loop=self.app.loop
             )
 
     async def on_response(self, _channel, body, _envelope, _properties):
         self._response = json.loads(body)
         self.waiter.set()
 
-    async def send(self, payload={}, properties={}, raw_data=False):
+    async def send(self, payload={}, properties={}, raw_data=False, consume_timeout=None):
         if not self.protocol:
-            await self.connect()
+            await self.connect(consume_timeout=consume_timeout)
 
         request_properties = deepcopy(self.DEFAULT_PROPERTIES)
         request_properties.update({'reply_to': self.response_queue_name})

sage_utils/amqp/workers.py

@@ -1,12 +1,22 @@
+import asyncio
+import logging
+
+from aioamqp.exceptions import AioamqpException
 from sage_utils.amqp.base import AmqpWorker
 from sage_utils.amqp.clients import RpcAmqpClient
 
 
+LOGGER = logging.getLogger(__name__)
+
+
 class BaseRegisterWorker(AmqpWorker):
     """
     Base class for implementing worker which is registering a new
     microservice in Open Matchmaking platform.
     """
+    MAX_RETRIES = 5
+    RETRY_TIMEOUT = 10
+
     REQUEST_QUEUE_NAME = "auth.microservices.register"
     REQUEST_EXCHANGE_NAME = "open-matchmaking.direct"
     RESPONSE_EXCHANGE_NAME = "open-matchmaking.responses.direct"
@@ -24,8 +34,31 @@ async def run(self, *args, **kwargs):
             response_queue='',
             response_exchange=self.RESPONSE_EXCHANGE_NAME
         )
-        response = await client.send(self.get_microservice_data(self.app))
 
-        assert 'error' not in response.keys(), response['error']
-        assert 'content' in response.keys()
-        assert response['content'] == 'OK'
+        is_registered = False
+        microservice_data = self.get_microservice_data(self.app)
+        for _ in range(self.MAX_RETRIES):
+            try:
+                response = await client.send(
+                    payload=microservice_data,
+                    consume_timeout=self.RETRY_TIMEOUT
+                )
+
+                if 'error' in response.keys():
+                    LOGGER.error("Received validation errors: {}".format(response['error']))
+                else:
+                    assert 'content' in response.keys()
+                    assert response['content'] == 'OK'
+                    is_registered = True
+
+                break
+            except (AioamqpException, TimeoutError):
+                LOGGER.error(
+                    "Can't receive a response because the Auth/Auth microservice isn't "
+                    "responding or the required queues and the exchanges aren't created. "
+                    "Retry after {} second(s).".format(self.RETRY_TIMEOUT)
+                )
+                await asyncio.sleep(self.RETRY_TIMEOUT, loop=self.app.loop)
+
+        if not is_registered:
+            raise ConnectionError('Occurred an error during registering microservice.')