commit 87d7e6ced49ebd16a460259c92088304ac4cd46a parent 971f56d4a6de6747af36fd8869a31937804e7531 Author: Stéphan Kochen <git@stephank.nl> Date: Thu, 03 Sep 2026 20:31:23 +0200 Acknowledge response stream segments incrementally
diff --git a/custom_components/inlite/lib/inlite_ble/hub.py b/custom_components/inlite/lib/inlite_ble/hub.py index 9fa4e2af23a657c694f65c5be7fdd3f528147b4a..d16f3e940b5f504f598b9ee53ea98534a2ead967 100644 --- a/custom_components/inlite/lib/inlite_ble/hub.py +++ b/custom_components/inlite/lib/inlite_ble/hub.py @@ -118,7 +118,7 @@ self._loop: asyncio.AbstractEventLoop | None = None # Control packets are queued so a response FLUSH arriving immediately # after a request-completion ACK cannot be lost by clearing an Event. self._ack_queue: asyncio.Queue[bytes] = asyncio.Queue() - self._stream_event = asyncio.Event() + self._stream_queue: asyncio.Queue[bytes] = asyncio.Queue() self._last_stream_data = b"" self._stream_buffer = bytearray() self._stream_invalid = False @@ -274,8 +274,14 @@ and decrypted["dest_id"] == self._crypto.controller_address ): loop.call_soon_threadsafe(self._ack_queue.put_nowait, payload) elif pkt_type == PKT_BLOCK_STREAM: - if self._accept_stream_segment(payload): - loop.call_soon_threadsafe(self._stream_event.set) + if ( + decrypted["src_id"] == self._device_id + and decrypted["dest_id"] == self._crypto.controller_address + ): + # ACK each stream segment from the command coroutine. This + # preserves packet order and avoids writing from the BLE + # notification callback. + loop.call_soon_threadsafe(self._stream_queue.put_nowait, payload) elif pkt_type == PKT_BLOCK_DATA_BLK: if len(payload) >= 3 and payload[1:3] == b"\x0c\x00": self._discovered_device_ids.add(decrypted["src_id"]) @@ -416,11 +422,12 @@ has no response stream; callers of normal commands must subsequently invoke ``_receive_response_stream``. """ dest = self._device_id - self._stream_event.clear() self._stream_buffer.clear() self._stream_invalid = False while not self._ack_queue.empty(): self._ack_queue.get_nowait() + while not self._stream_queue.empty(): + self._stream_queue.get_nowait() # ``build_block_data_payload`` retains its offset prefix for callers # and tests. The stream transport owns that prefix, however, so strip @@ -468,7 +475,7 @@ return success async def _receive_response_stream(self) -> bytes | None: - """Drain the hub's FLUSH → STREAM → FLUSH response exchange.""" + """Drain the hub's FLUSH → STREAM* → FLUSH response exchange.""" dest = self._device_id # The next queued control packet is the hub's response FLUSH(0). @@ -477,27 +484,52 @@ await asyncio.sleep(WRITE_DELAY) await self._write_mesh(dest, PKT_BLOCK_ACK, build_ack_payload(0)) try: - await asyncio.wait_for(self._stream_event.wait(), STREAM_TIMEOUT) + while True: + response_type, payload = await self._wait_response_part() + if response_type == PKT_BLOCK_STREAM: + if not self._accept_stream_segment(payload): + _LOGGER.warning("Discarding malformed STREAM response") + return None + # Android acknowledges every segment by its cumulative stream + # offset; wait for the next segment or final FLUSH afterwards. + await self._write_mesh( + dest, PKT_BLOCK_ACK, build_ack_payload(len(self._stream_buffer)) + ) + continue + + flush_count = ( + payload[0] | (payload[1] << 8) + if len(payload) >= 2 + else len(self._stream_buffer) + ) + await asyncio.sleep(WRITE_DELAY) + # The 0xef terminator is a third byte, not an arithmetic adjustment + # of the 16-bit byte count. + await self._write_mesh( + dest, PKT_BLOCK_ACK, build_ack_payload(flush_count, end=True) + ) + return bytes(self._stream_buffer) except asyncio.TimeoutError: - _LOGGER.warning("STREAM timeout waiting for hub response") return None - # More contiguous ATT notifications may already be in flight. - await asyncio.sleep(WRITE_DELAY) - if self._stream_invalid: - _LOGGER.warning("Discarding malformed STREAM response") - return None - stream_data = bytes(self._last_stream_data) - await self._write_mesh(dest, PKT_BLOCK_ACK, build_ack_payload(len(stream_data))) - flush = await self._wait_ack(timeout=STREAM_TIMEOUT) - flush_count = ( - flush[0] | (flush[1] << 8) if len(flush) >= 2 else len(stream_data) + async def _wait_response_part(self) -> tuple[int, bytes]: + """Return whichever arrives next: a stream segment or final FLUSH.""" + ack_task = asyncio.create_task(self._ack_queue.get()) + stream_task = asyncio.create_task(self._stream_queue.get()) + done, pending = await asyncio.wait( + (ack_task, stream_task), timeout=STREAM_TIMEOUT, + return_when=asyncio.FIRST_COMPLETED, ) - await asyncio.sleep(WRITE_DELAY) - # The 0xef terminator is a third byte, not an arithmetic adjustment - # of the 16-bit byte count. - await self._write_mesh(dest, PKT_BLOCK_ACK, build_ack_payload(flush_count, end=True)) - return stream_data + for task in pending: + task.cancel() + if pending: + await asyncio.gather(*pending, return_exceptions=True) + if not done: + _LOGGER.warning("Timed out waiting for hub response stream") + raise asyncio.TimeoutError + if stream_task in done: + return PKT_BLOCK_STREAM, stream_task.result() + return PKT_BLOCK_FLUSH, ack_task.result() async def _send_acknowledged_command( self, opcode: int, command_data: bytes diff --git a/tests/test_hub.py b/tests/test_hub.py index a625d0b24d3d547605e59e3b9388d25271de7dbe..39a8f52a6a0395a20511f63e9e584562fa560db9 100644 --- a/tests/test_hub.py +++ b/tests/test_hub.py @@ -3,7 +3,12 @@ import asyncio from inlite_ble.hub import BLE_PACKET_PART_SIZE, STREAM_DATA_SIZE, InliteHub, ZoneState -from inlite_ble.protocol import CHAR_CONTINUATION_UUID, CHAR_WRITE_UUID +from inlite_ble.protocol import ( + CHAR_CONTINUATION_UUID, + CHAR_WRITE_UUID, + PKT_BLOCK_FLUSH, + PKT_BLOCK_STREAM, +) class TestZoneState: @@ -82,16 +87,57 @@ writes.append((packet_type, data)) hub._write_mesh = write_mesh # type: ignore[method-assign] hub._ack_queue.put_nowait(b"\x00\x00") - hub._ack_queue.put_nowait(b"\x03\x00") - hub._stream_buffer.extend(b"abc") - hub._last_stream_data = b"abc" - hub._stream_event.set() + + parts = iter( + [ + (PKT_BLOCK_STREAM, b"\x00\x00abc"), + (PKT_BLOCK_FLUSH, b"\x03\x00"), + ] + ) + + async def wait_response_part() -> tuple[int, bytes]: + return next(parts) + + hub._wait_response_part = wait_response_part # type: ignore[method-assign] assert await hub._receive_response_stream() == b"abc" return writes writes = asyncio.run(run()) assert writes == [(0x72, b"\x00\x00"), (0x72, b"\x03\x00"), (0x72, b"\x03\x00\xef")] + + def test_response_stream_acknowledges_each_segment(self) -> None: + async def run() -> list[tuple[int, bytes]]: + hub = InliteHub(device_id=1, passphrase="test") + writes: list[tuple[int, bytes]] = [] + + async def write_mesh(_dest: int, packet_type: int, data: bytes) -> None: + writes.append((packet_type, data)) + + hub._write_mesh = write_mesh # type: ignore[method-assign] + hub._ack_queue.put_nowait(b"\x00\x00") + parts = iter( + [ + (PKT_BLOCK_STREAM, b"\x00\x00abc"), + (PKT_BLOCK_STREAM, b"\x03\x00de"), + (PKT_BLOCK_FLUSH, b"\x05\x00"), + ] + ) + + async def wait_response_part() -> tuple[int, bytes]: + return next(parts) + + hub._wait_response_part = wait_response_part # type: ignore[method-assign] + assert await hub._receive_response_stream() == b"abcde" + return writes + + writes = asyncio.run(run()) + assert writes == [ + (0x72, b"\x00\x00"), + (0x72, b"\x03\x00"), + (0x72, b"\x05\x00"), + (0x72, b"\x05\x00\xef"), + ] def test_complete_notification_reassembles_continuation(self) -> None: received: list[dict] = []