hat.event.server.engine_runner

  1from collections.abc import Callable
  2import asyncio
  3import logging
  4
  5from hat import aio
  6from hat import json
  7
  8from hat.event import common
  9from hat.event.server.engine import create_engine
 10from hat.event.server.eventer_client import SyncedState
 11from hat.event.server.eventer_client_runner import EventerClientRunner
 12from hat.event.server.eventer_server import EventerServer
 13
 14
 15mlog: logging.Logger = logging.getLogger(__name__)
 16"""Module logger"""
 17
 18
 19class EngineRunner(aio.Resource):
 20
 21    def __init__(self,
 22                 conf: json.Data,
 23                 backend: common.Backend,
 24                 eventer_server: EventerServer,
 25                 eventer_client_runner: EventerClientRunner | None,
 26                 reset_monitor_ready_cb: Callable[[], None]):
 27        self._conf = conf
 28        self._backend = backend
 29        self._eventer_server = eventer_server
 30        self._eventer_client_runner = eventer_client_runner
 31        self._reset_monitor_ready_cb = reset_monitor_ready_cb
 32        self._async_group = aio.Group()
 33        self._engine = None
 34        self._restart = asyncio.Event()
 35
 36        self.async_group.spawn(self._run)
 37
 38    @property
 39    def async_group(self) -> aio.Group:
 40        return self._async_group
 41
 42    async def set_synced(self,
 43                         server_id: common.ServerId,
 44                         state: SyncedState,
 45                         count: int | None):
 46        if not self._engine or not self._engine.is_open:
 47            return
 48
 49        data = {'state': state.name}
 50        if state == SyncedState.SYNCED:
 51            data['count'] = count
 52
 53        source = common.Source(type=common.SourceType.SERVER, id=0)
 54        event = common.RegisterEvent(
 55            type=('event', str(self._conf['server_id']), 'synced',
 56                  str(server_id)),
 57            source_timestamp=None,
 58            payload=common.EventPayloadJson(data))
 59
 60        await self._engine.register(source, [event])
 61
 62    async def _run(self):
 63        try:
 64            mlog.debug("staring engine runner loop")
 65            while True:
 66                await self._wait_while_remote_active()
 67
 68                self._restart.clear()
 69
 70                await self._eventer_server.set_status(common.Status.STARTING,
 71                                                      None)
 72
 73                mlog.debug("creating engine")
 74                self._engine = await create_engine(
 75                    backend=self._backend,
 76                    eventer_server=self._eventer_server,
 77                    module_confs=self._conf['modules'],
 78                    server_id=self._conf['server_id'],
 79                    restart_cb=self._restart.set,
 80                    reset_monitor_ready_cb=self._reset_monitor_ready_cb)
 81                await self._eventer_server.set_status(
 82                    common.Status.OPERATIONAL, self._engine)
 83
 84                async with self._async_group.create_subgroup() as subgroup:
 85                    await asyncio.wait(
 86                        [subgroup.spawn(self._engine.wait_closing),
 87                         subgroup.spawn(self._restart.wait)],
 88                        return_when=asyncio.FIRST_COMPLETED)
 89
 90                if not self._engine.is_open:
 91                    break
 92
 93                await self._close()
 94
 95        except Exception as e:
 96            mlog.error("engine runner loop error: %s", e, exc_info=e)
 97
 98        finally:
 99            mlog.debug("closing engine runner loop")
100            self.close()
101            await aio.uncancellable(self._close())
102
103    async def _close(self):
104        if self._engine:
105            self._engine.close()
106
107        await self._eventer_server.set_status(common.Status.STOPPING, None)
108
109        if self._engine:
110            await self._engine.async_close()
111
112        await self._backend.flush()
113
114        # TODO not needed with _wait_while_remote_active
115        # await self._eventer_server.notify_events([], True, True)
116
117        await self._eventer_server.set_status(common.Status.STANDBY, None)
118
119    async def _wait_while_remote_active(self):
120        if not self._eventer_client_runner:
121            return
122
123        while self._eventer_client_runner.remote_active:
124            event = asyncio.Event()
125            with self._eventer_client_runner.register_remote_active_cb(
126                    lambda _: event.set()):
127                await event.wait()