fix(examples): address file RPC review feedback

Author: MAMONTBOT

examples/file_client.py

@@ -33,29 +33,42 @@
 
 
 @dataclass(frozen=True)
-class FileReadResponse:
-    error: int
-    data: bytes
-
+class FileReadRequest:
+    _file_path: str
+    _read_offset: int
 
-def _encode_request(file_path: str, read_offset: int) -> bytes:
-    encoded_path = file_path.encode("utf8")
-    if len(encoded_path) > PATH_MAX_LEN:
-        raise ValueError(f"File path length {len(encoded_path)} is too long")
-    return struct.pack(REQUEST_HEADER_FORMAT, read_offset, len(encoded_path)) + encoded_path
+    def serialize(self) -> bytes:
+        encoded_path = self._file_path.encode("utf8")
+        if len(encoded_path) > PATH_MAX_LEN:
+            raise ValueError(f"File path length {len(encoded_path)} is too long")
+        return struct.pack(REQUEST_HEADER_FORMAT, self._read_offset, len(encoded_path)) + encoded_path
 
 
-def _decode_response(payload: bytes) -> FileReadResponse | None:
-    if len(payload) < RESPONSE_HEADER_SIZE:
-        return None
-    error, data_len = struct.unpack_from(RESPONSE_HEADER_FORMAT, payload)
-    if data_len > DATA_MAX:
-        return None
-    data_start = RESPONSE_HEADER_SIZE
-    data_end = data_start + data_len
-    if len(payload) != data_end:
-        return None
-    return FileReadResponse(error=error, data=payload[data_start:data_end])
+@dataclass(frozen=True)
+class FileReadResponse:
+    _error: int
+    _data: bytes
+
+    @property
+    def error(self) -> int:
+        return self._error
+
+    @property
+    def data(self) -> bytes:
+        return self._data
+
+    @staticmethod
+    def deserialize(payload: bytes) -> FileReadResponse | None:
+        if len(payload) < RESPONSE_HEADER_SIZE:
+            return None
+        error, data_len = struct.unpack_from(RESPONSE_HEADER_FORMAT, payload)
+        if data_len > DATA_MAX:
+            return None
+        data_start = RESPONSE_HEADER_SIZE
+        data_end = data_start + data_len
+        if len(payload) != data_end:
+            return None
+        return FileReadResponse(error, payload[data_start:data_end])
 
 
 def _format_remote_error(error: int) -> str:
@@ -75,7 +88,7 @@ async def _receive_response(stream: ResponseStream, expected_server_id: int | No
                 expected_server_id,
             )
             continue
-        decoded = _decode_response(response.message)
+        decoded = FileReadResponse.deserialize(response.message)
         if decoded is None:
             _logger.debug("dropping malformed response from %016x seq=%d", response.remote_id, response.seqno)
             continue
@@ -104,7 +117,7 @@ async def run(file_path: str) -> int:
             _logger.info("requesting offset %d", read_offset)
             stream: ResponseStream | None = None
             try:
-                request = _encode_request(file_path, read_offset)
+                request = FileReadRequest(file_path, read_offset).serialize()
                 stream = await pub.request(Instant.now() + REQUEST_DELIVERY_TIMEOUT, RESPONSE_TIMEOUT, request)
                 server_id, response = await _receive_valid_response(stream, discovered_server_id, RESPONSE_TIMEOUT)
             except ValueError as ex:

examples/file_server.py

@@ -12,7 +12,6 @@
 import logging
 import struct
 from dataclasses import dataclass
-from functools import partial
 from pathlib import Path
 
 from pycyphal2 import Arrival, DeliveryError, NackError, Node, SendError
@@ -32,36 +31,51 @@
 
 @dataclass(frozen=True)
 class FileReadRequest:
-    read_offset: int
-    file_path: str
+    _read_offset: int
+    _file_path: str
+
+    @property
+    def read_offset(self) -> int:
+        return self._read_offset
+
+    @property
+    def file_path(self) -> str:
+        return self._file_path
+
+    @staticmethod
+    def deserialize(payload: bytes) -> FileReadRequest | None:
+        if len(payload) < REQUEST_HEADER_SIZE:
+            return None
+        read_offset, path_len = struct.unpack_from(REQUEST_HEADER_FORMAT, payload)
+        if path_len == 0 or path_len > PATH_MAX_LEN:
+            return None
+        path_end = REQUEST_HEADER_SIZE + path_len
+        if len(payload) != path_end:
+            return None
+        try:
+            file_path = payload[REQUEST_HEADER_SIZE:path_end].decode("utf8")
+        except UnicodeDecodeError:
+            return None
+        return FileReadRequest(read_offset, file_path)
 
 
 @dataclass(frozen=True)
 class FileReadResponse:
-    error: int
-    data: bytes
-
-
-def _decode_request(payload: bytes) -> FileReadRequest | None:
-    if len(payload) < REQUEST_HEADER_SIZE:
-        return None
-    read_offset, path_len = struct.unpack_from(REQUEST_HEADER_FORMAT, payload)
-    if path_len == 0 or path_len > PATH_MAX_LEN:
-        return None
-    path_end = REQUEST_HEADER_SIZE + path_len
-    if len(payload) != path_end:
-        return None
-    try:
-        file_path = payload[REQUEST_HEADER_SIZE:path_end].decode("utf8")
-    except UnicodeDecodeError:
-        return None
-    return FileReadRequest(read_offset=read_offset, file_path=file_path)
+    _error: int
+    _data: bytes
+
+    @property
+    def error(self) -> int:
+        return self._error
 
+    @property
+    def data(self) -> bytes:
+        return self._data
 
-def _encode_response(response: FileReadResponse) -> bytes:
-    if len(response.data) > DATA_MAX:
-        raise ValueError(f"Response data is too large: {len(response.data)}")
-    return struct.pack(RESPONSE_HEADER_FORMAT, response.error, len(response.data)) + response.data
+    def serialize(self) -> bytes:
+        if len(self._data) > DATA_MAX:
+            raise ValueError(f"Response data is too large: {len(self._data)}")
+        return struct.pack(RESPONSE_HEADER_FORMAT, self._error, len(self._data)) + self._data
 
 
 def _errno_from_exception(ex: BaseException) -> int:
@@ -78,13 +92,13 @@ def _read_chunk(file_path: str, offset: int) -> FileReadResponse:
             file.seek(offset)
             data = file.read(DATA_MAX)
     except (OSError, ValueError, OverflowError) as ex:
-        return FileReadResponse(error=_errno_from_exception(ex), data=b"")
-    return FileReadResponse(error=0, data=data)
+        return FileReadResponse(_errno_from_exception(ex), b"")
+    return FileReadResponse(0, data)
 
 
 async def _serve_request(arrival: Arrival, request: FileReadRequest) -> None:
     response = _read_chunk(request.file_path, request.read_offset)
-    payload = _encode_response(response)
+    payload = response.serialize()
     _logger.info(
         "responding: file=%r offset=%d size=%d error=%d",
         request.file_path,
@@ -104,36 +118,20 @@ async def _serve_request(arrival: Arrival, request: FileReadRequest) -> None:
         _logger.warning("response send failed: remote=%016x error=%s", arrival.breadcrumb.remote_id, ex)
 
 
-def _on_task_done(tasks: set[asyncio.Task[None]], task: asyncio.Task[None]) -> None:
-    tasks.discard(task)
-    if task.cancelled():
-        return
-    exc = task.exception()
-    if exc is not None:
-        _logger.error("file request task failed: %s", exc)
-
-
 async def run() -> None:
     transport = UDPTransport.new()
     node = Node.new(transport, NAME)
     sub = node.subscribe(TOPIC)
-    tasks: set[asyncio.Task[None]] = set()
     _logger.info("file server ready on %r via %s", TOPIC, transport)
     try:
         async for arrival in sub:
-            request = _decode_request(arrival.message)
+            request = FileReadRequest.deserialize(arrival.message)
             if request is None:
                 _logger.debug("dropping malformed request of size %d", len(arrival.message))
                 continue
-            task = asyncio.create_task(_serve_request(arrival, request), name=f"file:{arrival.breadcrumb.tag}")
-            tasks.add(task)
-            task.add_done_callback(partial(_on_task_done, tasks))
+            await _serve_request(arrival, request)
     finally:
         sub.close()
-        for task in list(tasks):
-            task.cancel()
-        if tasks:
-            await asyncio.gather(*tasks, return_exceptions=True)
         node.close()
         transport.close()
 

src/pycyphal2/__init__.py

@@ -155,7 +155,7 @@ async def main():
 from ._transport import Transport as Transport
 from ._transport import TransportArrival as TransportArrival
 
-__version__ = "2.0.0.dev6"
+__version__ = "2.0.0.dev7"
 
 # pdoc needs __all__ to display re-exported members.
 __all__ = [

tests/test_examples.py

@@ -6,14 +6,14 @@
 
 import pytest
 
-from examples.file_client import _decode_response, _format_remote_error, _receive_valid_response
-from examples.file_server import _decode_request, _read_chunk
+from examples.file_client import FileReadResponse, _format_remote_error, _receive_valid_response
+from examples.file_server import FileReadRequest, _read_chunk
 from pycyphal2 import Instant, LivenessError, Response, ResponseStream
 
 
 def test_file_example_decoders_reject_trailing_garbage() -> None:
-    assert _decode_request(struct.pack("<QH", 0, 1) + b"x" + b"y") is None
-    assert _decode_response(struct.pack("<IH", 0, 0) + b"x") is None
+    assert FileReadRequest.deserialize(struct.pack("<QH", 0, 1) + b"x" + b"y") is None
+    assert FileReadResponse.deserialize(struct.pack("<IH", 0, 0) + b"x") is None
 
 
 def test_file_server_read_chunk_maps_invalid_inputs_to_errors() -> None: