hat.gateway.devices.smpp.client

  1import asyncio
  2import contextlib
  3import logging
  4import typing
  5
  6from hat import aio
  7from hat import util
  8from hat.drivers import smpp
  9from hat.drivers import ssl
 10from hat.drivers import tcp
 11import hat.event.common
 12import hat.event.eventer
 13
 14from hat.gateway import common
 15
 16
 17mlog: logging.Logger = logging.getLogger(__name__)
 18
 19
 20class SmppClientDevice(common.Device):
 21
 22    def __init__(self,
 23                 conf: common.DeviceConf,
 24                 eventer_client: hat.event.eventer.Client,
 25                 event_type_prefix: common.EventTypePrefix):
 26        self._conf = conf
 27        self._eventer_client = eventer_client
 28        self._event_type_prefix = event_type_prefix
 29        self._async_group = aio.Group()
 30        self._msg_queue = aio.Queue()
 31        self._log = _create_logger_adapter(conf['name'])
 32
 33        self.async_group.spawn(self._connection_loop)
 34
 35    @property
 36    def async_group(self) -> aio.Group:
 37        return self._async_group
 38
 39    async def process_event(self, event: hat.event.common.Event):
 40        try:
 41            self._log.debug('received event: %s', event)
 42            msg = _msg_from_event(
 43                event_type_prefix=self._event_type_prefix,
 44                message_encoding=self._conf['message_encoding'],
 45                event=event)
 46
 47            if self._msg_queue is None:
 48                raise Exception('connection closed')
 49
 50            await self._msg_queue.put(msg)
 51
 52        except Exception as e:
 53            self._log.warning('error processing event: %s', e, exc_info=e)
 54
 55    async def _connection_loop(self):
 56        conn = None
 57
 58        async def cleanup():
 59            with contextlib.suppress(ConnectionError):
 60                await self._register_status('DISCONNECTED')
 61
 62            if conn:
 63                await conn.async_close()
 64
 65        try:
 66            while True:
 67                await self._register_status('CONNECTING')
 68                try:
 69                    ssl_ctx = (ssl.create_ssl_ctx(ssl.SslProtocol.TLS_CLIENT)
 70                               if self._conf['ssl'] else None)
 71                    conn = await aio.wait_for(
 72                        smpp.connect(
 73                            addr=tcp.Address(
 74                                host=self._conf['remote_address']['host'],
 75                                port=self._conf['remote_address']['port']),
 76                            system_id=self._conf['system_id'],
 77                            password=self._conf['password'],
 78                            enquire_link_delay=self._conf['enquire_link_delay'],  # NOQA
 79                            enquire_link_timeout=self._conf['enquire_link_timeout'],  # NOQA
 80                            ssl=ssl_ctx),
 81                        self._conf['connect_timeout'])
 82
 83                except Exception as e:
 84                    self._log.warning('connection failed: %s', e, exc_info=e)
 85                    await asyncio.sleep(self._conf['reconnect_delay'])
 86                    continue
 87
 88                except aio.CancelledWithResultError as e:
 89                    conn = e.result
 90                    raise
 91
 92                await self._register_status('CONNECTED')
 93
 94                try:
 95                    self._msg_queue = aio.Queue(1024)
 96
 97                    if conn.is_open:
 98                        conn.async_group.spawn(self._send_loop, conn,
 99                                               self._msg_queue)
100                        await conn.wait_closing()
101
102                finally:
103                    self._msg_queue = None
104
105                await self._register_status('DISCONNECTED')
106                await conn.async_close()
107                self._conn = None
108
109        except ConnectionError:
110            pass
111
112        except Exception as e:
113            self._log.error('connection loop error: %s', e, exc_info=e)
114
115        finally:
116            self._log.debug('closing connection loop')
117            self.close()
118            await aio.uncancellable(cleanup())
119
120    async def _send_loop(self, conn, msg_queue):
121        try:
122            short_message = self._conf['short_message']
123            priority = smpp.Priority[self._conf['priority']]
124            data_coding = smpp.DataCoding[self._conf['data_coding']]
125
126            while True:
127                msg = await msg_queue.get()
128
129                self._log.debug('sending message to %s', msg.address)
130                msg_id = await conn.send_message(dst_addr=msg.address,
131                                                 msg=msg.message,
132                                                 short_message=short_message,
133                                                 priority=priority,
134                                                 data_coding=data_coding)
135
136                self._log.debug('message sent with id %s', msg_id.hex())
137
138        except ConnectionError:
139            self._log.debug('connection closed')
140
141        except asyncio.TimeoutError:
142            self._log.debug('send timeout')
143
144        except Exception as e:
145            self._log.error('send loop error: %s', e, exc_info=e)
146
147        finally:
148            self._log.debug('closing send loop')
149            conn.close()
150
151    async def _register_status(self, status):
152        event = hat.event.common.RegisterEvent(
153            type=(*self._event_type_prefix, 'gateway', 'status'),
154            source_timestamp=None,
155            payload=hat.event.common.EventPayloadJson(status))
156
157        await self._eventer_client.register([event])
158        self._log.debug('registered status %s', status)
159
160
161info: common.DeviceInfo = common.DeviceInfo(
162    type="smpp_client",
163    create=SmppClientDevice,
164    json_schema_id="hat-gateway://smpp.yaml#/$defs/client",
165    json_schema_repo=common.json_schema_repo)
166
167
168class _Msg(typing.NamedTuple):
169    address: str
170    message: util.Bytes
171
172
173def _msg_from_event(event_type_prefix, message_encoding, event):
174    suffix = event.type[len(event_type_prefix):]
175    if suffix != ('system', 'message'):
176        raise Exception('unsupported event type')
177
178    if not isinstance(event.payload.data['address'], str):
179        raise Exception('invalid address type')
180
181    if not isinstance(event.payload.data['message'], str):
182        raise Exception('invalid message type')
183
184    return _Msg(address=event.payload.data['address'],
185                message=event.payload.data['message'].encode(message_encoding))
186
187
188def _create_logger_adapter(name):
189    extra = {'meta': {'type': 'SmppClientDevice',
190                      'name': name}}
191
192    return logging.LoggerAdapter(mlog, extra)