hat.event.server.eventer_server

  1"""Eventer server"""
  2
  3from collections.abc import Collection, Iterable
  4import logging
  5
  6from hat import aio
  7from hat.drivers import tcp
  8
  9from hat.event import common
 10from hat.event import eventer
 11
 12
 13mlog: logging.Logger = logging.getLogger(__name__)
 14"""Module logger"""
 15
 16
 17async def create_eventer_server(addr: tcp.Address,
 18                                backend: common.Backend,
 19                                server_id: int,
 20                                *,
 21                                server_token: str | None = None,
 22                                **kwargs
 23                                ) -> 'EventerServer':
 24    """Create eventer server"""
 25    server = EventerServer()
 26    server._backend = backend
 27    server._server_id = server_id
 28    server._server_token = server_token
 29    server._engine = None
 30
 31    server._srv = await eventer.listen(addr,
 32                                       connected_cb=server._on_connected,
 33                                       disconnected_cb=server._on_disconnected,
 34                                       register_cb=server._on_register,
 35                                       query_cb=server._on_query,
 36                                       **kwargs)
 37
 38    return server
 39
 40
 41class EventerServer(aio.Resource):
 42    """Eventer server
 43
 44    For creating new server see `create_eventer_server` coroutine.
 45
 46    """
 47
 48    @property
 49    def async_group(self) -> aio.Group:
 50        """Async group"""
 51        return self._srv.async_group
 52
 53    def get_client_names(self) -> Iterable[tuple[common.Source, str]]:
 54        """Get client names"""
 55        for info in self._srv.get_conn_infos():
 56            yield _get_source(info.id), info.client_name
 57
 58    async def set_status(self,
 59                         status: common.Status,
 60                         engine: common.Engine | None):
 61        """Set status"""
 62        if status == common.Status.OPERATIONAL:
 63            if not engine:
 64                raise ValueError('invalid status/engine')
 65
 66        else:
 67            if engine:
 68                raise ValueError('invalid status/engine')
 69
 70        self._engine = engine
 71        await self._srv.set_status(status)
 72
 73    async def notify_events(self,
 74                            events: Collection[common.Event],
 75                            persisted: bool,
 76                            with_ack: bool = False):
 77        """Notify events"""
 78        await self._srv.notify_events(events, persisted, with_ack)
 79
 80    async def _on_connected(self, info):
 81        if (info.client_token is not None and
 82                info.client_token != self._server_token):
 83            raise Exception('invalid client token')
 84
 85        if not self._engine or not self._engine.is_open:
 86            return
 87
 88        source = _get_source(info.id)
 89        register_event = self._create_eventer_event(info, 'CONNECTED')
 90        await self._engine.register(source, [register_event])
 91
 92    async def _on_disconnected(self, info):
 93        if not self._engine or not self._engine.is_open:
 94            return
 95
 96        source = _get_source(info.id)
 97        register_event = self._create_eventer_event(info, 'DISCONNECTED')
 98        await self._engine.register(source, [register_event])
 99
100    async def _on_register(self, info, register_events):
101        if not self._engine:
102            return
103
104        source = _get_source(info.id)
105        return await self._engine.register(source, register_events)
106
107    async def _on_query(self, info, params):
108        return await self._backend.query(params)
109
110    def _create_eventer_event(self, info, status):
111        return common.RegisterEvent(
112            type=('event', str(self._server_id), 'eventer', info.client_name),
113            source_timestamp=None,
114            payload=common.EventPayloadJson(status))
115
116
117def _get_source(source_id):
118    return common.Source(type=common.SourceType.EVENTER,
119                         id=source_id)