server.py

@@ -2,6 +2,7 @@ import asyncio
 import datetime
 import grpc
 import bson
+import os
 
 from queues import tasks_pb2
 from queues import tasks_pb2_grpc
@@ -9,6 +10,18 @@ from queues import tasks_pb2_grpc
 from utils import time
 from storage.mongo import tasks
 
+from utils import configurator
+
+
+DEFAULT_RETRY_AFTER = 0.2
+
+
+client = configurator.Client(
+    'queues',
+    os.environ['STAGE'],
+    need_poll=True,
+)
+
 
 def get_feature(feature_db, point):
     """Returns Feature at given location or None."""
@@ -36,7 +49,9 @@ class TasksServicer(tasks_pb2_grpc.TasksServicer):
         task = await tasks.take_task(request.queue)
         if not task:
             return tasks_pb2.TakeResponse(task=None)
-        return tasks_pb2.TakeResponse(task=tasks_pb2.Task(id=str(task._id), attempt=task.attempts, payload=task.payload))
+        retry_after_settings = client.get_config('retry_after_settings')
+        retry_after = retry_after_settings.get(request.queue) or retry_after_settings.get('__default__') or DEFAULT_RETRY_AFTER
+        return tasks_pb2.TakeResponse(task=tasks_pb2.Task(id=str(task._id), attempt=task.attempts, payload=task.payload), retry_after=retry_after)
 
     async def Finish(self, request: tasks_pb2.FinishRequest, context) -> tasks_pb2.EmptyResponse:
         if await tasks.finish_task(bson.ObjectId(request.id)):