hat.gateway.runner

  1from collections.abc import Iterable
  2import asyncio
  3import collections
  4import logging
  5
  6from hat import aio
  7from hat import json
  8from hat.drivers import tcp
  9import hat.event.component
 10import hat.event.eventer
 11
 12from hat.gateway import common
 13import hat.gateway.engine
 14
 15
 16mlog: logging.Logger = logging.getLogger(__name__)
 17"""Module logger"""
 18
 19
 20class MainRunner(aio.Resource):
 21
 22    def __init__(self, conf: json.Data):
 23        self._conf = conf
 24        self._loop = asyncio.get_running_loop()
 25        self._async_group = aio.Group()
 26        self._eventer_component = None
 27        self._eventer_client = None
 28        self._eventer_runner = None
 29
 30        self.async_group.spawn(self._run)
 31
 32    @property
 33    def async_group(self) -> aio.Group:
 34        return self._async_group
 35
 36    async def _run(self):
 37        try:
 38            mlog.debug("starting main runner loop")
 39            await self._start()
 40
 41            await self._loop.create_future()
 42
 43        except Exception as e:
 44            mlog.error("main runner loop error: %s", e, exc_info=e)
 45
 46        finally:
 47            mlog.debug("closing main runner loop")
 48            self.close()
 49            await aio.uncancellable(self._stop())
 50
 51    async def _start(self):
 52        event_server_conf = self._conf['event_server']
 53
 54        subscriptions = collections.deque()
 55        for device_conf in self._conf['devices']:
 56            info = common.import_device_info(device_conf['module'])
 57            subscriptions.append(('gateway', info.type, device_conf['name'],
 58                                  'system', '*'))
 59
 60        if 'monitor_component' in event_server_conf:
 61            monitor_component_conf = event_server_conf['monitor_component']
 62
 63            mlog.debug("creating eventer component")
 64            self._eventer_component = await hat.event.component.connect(
 65                addr=tcp.Address(monitor_component_conf['host'],
 66                                 monitor_component_conf['port']),
 67                name=self._conf['name'],
 68                group=monitor_component_conf['gateway_group'],
 69                server_group=monitor_component_conf['event_server_group'],
 70                client_name=f"gateway/{self._conf['name']}",
 71                runner_cb=self._create_eventer_runner,
 72                status_cb=self._on_component_status,
 73                events_cb=self._on_component_events,
 74                eventer_kwargs={'subscriptions': subscriptions})
 75            _bind_resource(self.async_group, self._eventer_component)
 76
 77            await self._eventer_component.set_ready(True)
 78
 79        elif 'eventer_server' in event_server_conf:
 80            eventer_server_conf = event_server_conf['eventer_server']
 81
 82            mlog.debug("creating eventer client")
 83            self._eventer_client = await hat.event.eventer.connect(
 84                addr=tcp.Address(eventer_server_conf['host'],
 85                                 eventer_server_conf['port']),
 86                client_name=f"gateway/{self._conf['name']}",
 87                subscriptions=subscriptions,
 88                status_cb=self._on_client_status,
 89                events_cb=self._on_client_events)
 90            _bind_resource(self.async_group, self._eventer_client)
 91
 92            mlog.debug("creating eventer runner")
 93            self._eventer_runner = EventerRunner(
 94                conf=self._conf,
 95                eventer_client=self._eventer_client)
 96            _bind_resource(self.async_group, self._eventer_runner)
 97
 98        else:
 99            raise Exception('invalid configuration')
100
101    async def _stop(self):
102        if self._eventer_runner and not self._eventer_component:
103            await self._eventer_runner.async_close()
104
105        if self._eventer_client:
106            await self._eventer_client.async_close()
107
108        if self._eventer_component:
109            await self._eventer_component.async_close()
110
111    async def _create_eventer_runner(self, monitor_component, server_data,
112                                     eventer_client):
113        mlog.debug("creating eventer runner")
114        self._eventer_runner = EventerRunner(conf=self._conf,
115                                             eventer_client=eventer_client)
116
117        return self._eventer_runner
118
119    def _process_status(self, status):
120        if not self._eventer_runner:
121            return
122
123        self._eventer_runner.process_status(status)
124
125    async def _process_events(self, events):
126        if not self._eventer_runner:
127            return
128
129        await self._eventer_runner.process_events(events)
130
131    def _on_component_status(self, eventer_component, eventer_client, status):
132        self._process_status(status)
133
134    def _on_client_status(self, eventer_client, status):
135        self._process_status(status)
136
137    async def _on_component_events(self, eventer_component, eventer_client,
138                                   events):
139        await self._process_events(events)
140
141    async def _on_client_events(self, eventer_client, events):
142        await self._process_events(events)
143
144
145class EventerRunner(aio.Resource):
146
147    def __init__(self,
148                 conf: json.Data,
149                 eventer_client: hat.event.eventer.Client):
150        self._conf = conf
151        self._eventer_client = eventer_client
152        self._engine = None
153        self._status_event = asyncio.Event()
154        self._async_group = aio.Group()
155
156        self._status_event.set()
157        self.async_group.spawn(self._run)
158
159    @property
160    def async_group(self) -> aio.Group:
161        return self._async_group
162
163    def process_status(self, status: hat.event.common.Status):
164        self._status_event.set()
165
166    async def process_events(self, events: Iterable[hat.event.common.Event]):
167        if not self.is_open or not self._engine or not self._engine.is_open:
168            return
169
170        await self._engine.process_events(events)
171
172    async def _run(self):
173        try:
174            mlog.debug("starting eventer runner loop")
175            while True:
176                await self._status_event.wait()
177
178                self._status_event.clear()
179                if not self._is_active():
180                    continue
181
182                mlog.debug("creating engine")
183                self._engine = hat.gateway.engine.Engine(
184                    conf=self._conf,
185                    eventer_client=self._eventer_client)
186
187                async with self.async_group.create_subgroup() as subgroup:
188                    engine_closing_task = subgroup.spawn(
189                        self._engine.wait_closing)
190
191                    while True:
192                        if engine_closing_task.done():
193                            return
194
195                        self._status_event.clear()
196                        if not self._is_active():
197                            break
198
199                        status_task = subgroup.spawn(self._status_event.wait)
200
201                        await asyncio.wait(
202                            [engine_closing_task, status_task],
203                            return_when=asyncio.FIRST_COMPLETED)
204
205                await self._engine.async_close()
206
207        except Exception as e:
208            mlog.error("eventer runner loop error: %s", e, exc_info=e)
209
210        finally:
211            mlog.debug("closing eventer runner loop")
212            self.close()
213
214            if self._engine:
215                await aio.uncancellable(self._engine.async_close())
216
217    def _is_active(self):
218        if not self._eventer_client.is_open:
219            return False
220
221        if self._conf['event_server'].get('require_operational'):
222            return self._eventer_client.status == hat.event.common.Status.OPERATIONAL  # NOQA
223
224        return True
225
226
227def _bind_resource(async_group, resource):
228    async_group.spawn(aio.call_on_done, resource.wait_closing(),
229                      async_group.close)