the callback is run in a separate thread

Author: sanderegg

services/director-v2/src/simcore_service_director_v2/modules/comp_scheduler/_worker.py

@@ -1,29 +1,34 @@
 import functools
 import logging
-from typing import cast
+from typing import Callable, cast
 
 from fastapi import FastAPI
+from models_library.projects import ProjectID
+from models_library.users import UserID
 from servicelib.logging_utils import log_context
 
 from ..rabbitmq import get_rabbitmq_client
-from ._models import SchedulePipelineRabbitMessage
+from ._models import Iteration, SchedulePipelineRabbitMessage
 from ._scheduler_base import BaseCompScheduler
 from ._scheduler_factory import create_scheduler
 
 _logger = logging.getLogger(__name__)
 
 
 def _empty_wake_up_callack(
-    # app: FastAPI, user_id: UserID, project_id: ProjectID, iteration: Iteration
-) -> None:
+    app: FastAPI, user_id: UserID, project_id: ProjectID, iteration: Iteration
+) -> Callable[[], None]:
+    def _cb() -> None:
+        ...
+
     # async def _async_cb():
     #     db_engine = get_db_engine(app)
     #     rabbit_mq_client = get_rabbitmq_client(app)
     #     comp_run = await CompRunsRepository.instance(db_engine).get(
     #         user_id=user_id, project_id=project_id, iteration=iteration
     #     )
     #     await request_pipeline_scheduling(comp_run, rabbit_mq_client, db_engine)
-    ...
+    return _cb
 
 
 def _get_scheduler_worker(app: FastAPI) -> BaseCompScheduler:
@@ -34,15 +39,16 @@ async def _handle_distributed_pipeline(app: FastAPI, data: bytes) -> bool:
 
     with log_context(_logger, logging.DEBUG, msg="handling scheduling"):
         to_schedule_pipeline = SchedulePipelineRabbitMessage.parse_raw(data)
-        # get_rabbitmq_client(app).publish(
-        #     SchedulePipelineRabbitMessage.get_channel_name(),
-        #     to_schedule_pipeline,
-        # )
         await _get_scheduler_worker(app).schedule_pipeline(
             user_id=to_schedule_pipeline.user_id,
             project_id=to_schedule_pipeline.project_id,
             iteration=to_schedule_pipeline.iteration,
-            wake_up_callback=_empty_wake_up_callack,
+            wake_up_callback=_empty_wake_up_callack(
+                app,
+                to_schedule_pipeline.user_id,
+                to_schedule_pipeline.project_id,
+                to_schedule_pipeline.iteration,
+            ),
         )
         return True
 

services/director-v2/tests/unit/with_dbs/comp_scheduler/test_worker.py

@@ -17,7 +17,6 @@
 from simcore_service_director_v2.models.comp_runs import RunMetadataDict
 from simcore_service_director_v2.modules.comp_scheduler._manager import run_new_pipeline
 from simcore_service_director_v2.modules.comp_scheduler._worker import (
-    _empty_wake_up_callack,
     _get_scheduler_worker,
 )
 
@@ -68,5 +67,5 @@ async def test_worker_properly_calls_scheduler_api(
         user_id=published_project.project.prj_owner,
         project_id=published_project.project.uuid,
         iteration=1,
-        wake_up_callback=_empty_wake_up_callack,
+        wake_up_callback=mock.ANY,
     )