hat.drivers.cosp

28class TcpConnectionInfo(typing.NamedTuple):
29    name: str | None
30    local_addr: net.TcpAddress
31    local_tsel: int | None
32    local_ssel: int | None
33    remote_addr: net.TcpAddress
34    remote_tsel: int | None
35    remote_ssel: int | None

38class UnixConnectionInfo(typing.NamedTuple):
39    name: str | None
40    addr: net.UnixAddress
41    local_tsel: int | None
42    local_ssel: int | None
43    remote_tsel: int | None
44    remote_ssel: int | None

57async def connect(addr: net.StreamAddress,
58                  user_data: util.Bytes | None = None,
59                  *,
60                  local_ssel: int | None = None,
61                  remote_ssel: int | None = None,
62                  cosp_receive_queue_size: int = 1024,
63                  cosp_send_queue_size: int = 1024,
64                  **kwargs
65                  ) -> 'Connection':
66    """Connect to COSP server
67
68    Additional arguments are passed directly to `hat.drivers.cotp.connect`.
69
70    """
71    conn = await cotp.connect(addr, **kwargs)
72
73    try:
74        cn_spdu = common.Spdu(type=common.SpduType.CN,
75                              extended_spdus=False,
76                              version_number=_params_version,
77                              requirements=_params_requirements,
78                              calling_ssel=local_ssel,
79                              called_ssel=remote_ssel,
80                              user_data=user_data)
81        cn_spdu_bytes = encoder.encode(cn_spdu)
82        await conn.send(cn_spdu_bytes)
83
84        ac_spdu_bytes = await conn.receive()
85        ac_spdu = encoder.decode(memoryview(ac_spdu_bytes))
86        _validate_connect_response(cn_spdu, ac_spdu)
87
88        calling_ssel, called_ssel = _get_ssels(cn_spdu, ac_spdu)
89        return Connection(conn, cn_spdu, ac_spdu, calling_ssel, called_ssel,
90                          cosp_receive_queue_size, cosp_send_queue_size)
91
92    except BaseException:
93        await aio.uncancellable(_close_cotp(conn, _ab_spdu, mlog))
94        raise

 97async def listen(validate_cb: ValidateCb,
 98                 connection_cb: ConnectionCb,
 99                 addr: net.StreamAddress = net.TcpAddress('0.0.0.0', 102),
100                 *,
101                 bind_connections: bool = False,
102                 cosp_receive_queue_size: int = 1024,
103                 cosp_send_queue_size: int = 1024,
104                 **kwargs
105                 ) -> 'Server':
106    """Create COSP listening server
107
108    Additional arguments are passed directly to `hat.drivers.cotp.listen`.
109
110    Args:
111        validate_cb: callback function or coroutine called on new
112            incomming connection request prior to creating new connection
113        connection_cb: new connection callback
114        addr: local listening address
115
116    """
117    server = Server()
118    server._validate_cb = validate_cb
119    server._connection_cb = connection_cb
120    server._bind_connections = bind_connections
121    server._receive_queue_size = cosp_receive_queue_size
122    server._send_queue_size = cosp_send_queue_size
123    server._log = mlog
124
125    server._srv = await cotp.listen(server._on_connection, addr,
126                                    bind_connections=False,
127                                    **kwargs)
128
129    server._log = _create_server_logger(server._srv.info)
130
131    return server

134class Server(aio.Resource):
135    """COSP listening server
136
137    For creating new server see `listen`.
138
139    """
140
141    @property
142    def async_group(self) -> aio.Group:
143        """Async group"""
144        return self._srv.async_group
145
146    @property
147    def info(self) -> net.ServerInfo:
148        """Server info"""
149        return self._srv.info
150
151    async def _on_connection(self, cotp_conn):
152        try:
153            try:
154                cn_spdu_bytes = await cotp_conn.receive()
155                cn_spdu = encoder.decode(memoryview(cn_spdu_bytes))
156                _validate_connect_request(cn_spdu)
157
158                res_user_data = await aio.call(self._validate_cb,
159                                               cn_spdu.user_data)
160
161                ac_spdu = common.Spdu(type=common.SpduType.AC,
162                                      extended_spdus=False,
163                                      version_number=_params_version,
164                                      requirements=_params_requirements,
165                                      calling_ssel=cn_spdu.calling_ssel,
166                                      called_ssel=cn_spdu.called_ssel,
167                                      user_data=res_user_data)
168                ac_spdu_bytes = encoder.encode(ac_spdu)
169                await cotp_conn.send(ac_spdu_bytes)
170
171                calling_ssel, called_ssel = _get_ssels(cn_spdu, ac_spdu)
172                conn = Connection(cotp_conn, cn_spdu, ac_spdu,
173                                  called_ssel, calling_ssel,
174                                  self._receive_queue_size,
175                                  self._send_queue_size)
176
177            except BaseException:
178                await aio.uncancellable(
179                    _close_cotp(cotp_conn, _ab_spdu, self._log))
180                raise
181
182            try:
183                await aio.call(self._connection_cb, conn)
184
185            except BaseException:
186                await aio.uncancellable(conn.async_close())
187                raise
188
189        except Exception as e:
190            self._log.error("error creating new incomming connection: %s",
191                            e, exc_info=e)
192            return
193
194        if not self._bind_connections:
195            return
196
197        try:
198            await conn.wait_closed()
199
200        except BaseException:
201            await aio.uncancellable(conn.async_close())
202            raise

205class Connection(aio.Resource):
206    """COSP connection
207
208    For creating new connection see `connect` or `listen`.
209
210    """
211
212    def __init__(self,
213                 conn: cotp.Connection,
214                 cn_spdu: common.Spdu,
215                 ac_spdu: common.Spdu,
216                 local_ssel: int | None,
217                 remote_ssel: int | None,
218                 receive_queue_size: int,
219                 send_queue_size: int):
220        self._conn = conn
221        self._conn_req_user_data = cn_spdu.user_data
222        self._conn_res_user_data = ac_spdu.user_data
223        self._loop = asyncio.get_running_loop()
224        self._info = _connection_info_from_cotp(info=conn.info,
225                                                local_ssel=local_ssel,
226                                                remote_ssel=remote_ssel)
227        self._close_spdu = None
228        self._receive_queue = aio.Queue(receive_queue_size)
229        self._send_queue = aio.Queue(send_queue_size)
230        self._async_group = aio.Group()
231        self._log = _create_connection_logger(self._info)
232
233        self.async_group.spawn(aio.call_on_cancel, self._on_close)
234        self.async_group.spawn(self._receive_loop)
235        self.async_group.spawn(self._send_loop)
236        self.async_group.spawn(aio.call_on_done, conn.wait_closing(),
237                               self.close)
238
239    @property
240    def async_group(self) -> aio.Group:
241        """Async group"""
242        return self._async_group
243
244    @property
245    def info(self) -> ConnectionInfo:
246        """Connection info"""
247        return self._info
248
249    @property
250    def conn_req_user_data(self) -> util.Bytes:
251        """Connect request's user data"""
252        return self._conn_req_user_data
253
254    @property
255    def conn_res_user_data(self) -> util.Bytes:
256        """Connect response's user data"""
257        return self._conn_res_user_data
258
259    def close(self, user_data: util.Bytes | None = None):
260        """Close connection"""
261        self._close(common.Spdu(common.SpduType.FN,
262                                transport_disconnect=True,
263                                user_data=user_data))
264
265    async def async_close(self, user_data: util.Bytes | None = None):
266        """Async close"""
267        self.close(user_data)
268        await self.wait_closed()
269
270    async def receive(self) -> util.Bytes:
271        """Receive data"""
272        try:
273            return await self._receive_queue.get()
274
275        except aio.QueueClosedError:
276            raise ConnectionError()
277
278    async def send(self, data: util.Bytes):
279        """Send data"""
280        try:
281            await self._send_queue.put((data, None))
282
283        except aio.QueueClosedError:
284            raise ConnectionError()
285
286    async def drain(self):
287        """Drain output buffer"""
288        try:
289            future = self._loop.create_future()
290            await self._send_queue.put((None, future))
291            await future
292
293        except aio.QueueClosedError:
294            raise ConnectionError()
295
296    async def _on_close(self):
297        await _close_cotp(self._conn, self._close_spdu, self._log)
298
299    def _close(self, spdu):
300        if not self.is_open:
301            return
302
303        self._close_spdu = spdu
304        self._async_group.close()
305
306    async def _receive_loop(self):
307        try:
308            data = bytearray()
309            while True:
310                spdu_bytes = await self._conn.receive()
311                spdu = encoder.decode(memoryview(spdu_bytes))
312
313                if spdu.type == common.SpduType.DT:
314                    data.extend(spdu.data)
315
316                    if spdu.end is None or spdu.end:
317                        await self._receive_queue.put(data)
318                        data = bytearray()
319
320                elif spdu.type == common.SpduType.FN:
321                    self._close(_dn_spdu)
322                    break
323
324                elif spdu.type == common.SpduType.AB:
325                    self._close(None)
326                    break
327
328                else:
329                    self._close(_ab_spdu)
330                    break
331
332        except ConnectionError:
333            pass
334
335        except Exception as e:
336            self._log.error("receive loop error: %s", e, exc_info=e)
337
338        finally:
339            self.close()
340            self._receive_queue.close()
341
342    async def _send_loop(self):
343        future = None
344        try:
345            while True:
346                data, future = await self._send_queue.get()
347
348                if data is None:
349                    await self._conn.drain()
350
351                else:
352                    spdu = common.Spdu(type=common.SpduType.DT,
353                                       data=data)
354                    spdu_bytes = encoder.encode(spdu)
355
356                    msg = bytes(itertools.chain(common.give_tokens_spdu_bytes,
357                                                spdu_bytes))
358
359                    await self._conn.send(msg)
360
361                if future and not future.done():
362                    future.set_result(None)
363
364        except ConnectionError:
365            pass
366
367        except Exception as e:
368            self._log.error("send loop error: %s", e, exc_info=e)
369
370        finally:
371            self.close()
372            self._send_queue.close()
373
374            while True:
375                if future and not future.done():
376                    future.set_result(None)
377                if self._send_queue.empty():
378                    break
379                _, future = self._send_queue.get_nowait()