Implement zerocopy writes for the encrypted protocol

aiohomekit/controller/ip/connection.py

@@ -16,8 +16,10 @@
 from __future__ import annotations
 
 import asyncio
+from collections.abc import Iterable
 import logging
 import socket
+from struct import Struct
 from typing import TYPE_CHECKING, Any
 
 import aiohappyeyeballs
@@ -45,6 +47,9 @@
 from aiohomekit.protocol.tlv import TLV
 from aiohomekit.utils import async_create_task, asyncio_timeout
 
+PACK_UNSIGNED_SHORT_LITTLE = Struct("<H").pack
+
+
 if TYPE_CHECKING:
     from .pairing import IpPairing
 
@@ -80,11 +85,11 @@ def __init__(self, connection: HomeKitConnection) -> None:
         self.current_response = HttpResponse()
         self.loop = asyncio.get_running_loop()
 
-    def connection_made(self, transport):
+    def connection_made(self, transport: asyncio.Transport) -> None:
         super().connection_made(transport)
         self.transport = transport
 
-    def connection_lost(self, exception):
+    def connection_lost(self, exception: Exception) -> None:
         self.connection._connection_lost(exception)
         self._cancel_pending_requests()
 
@@ -94,10 +99,14 @@ def _handle_timeout(self, fut: asyncio.Future[Any]) -> None:
             fut.set_exception(asyncio.TimeoutError)
 
     async def send_bytes(self, payload: bytes) -> HttpResponse:
+        """Send bytes to the device."""
+        return await self._send_lines((payload,))
+
+    async def _send_lines(self, payload: Iterable[bytes]) -> HttpResponse:
         """Send bytes to the device."""
         if self.transport.is_closing():
             # FIXME: It would be nice to try and wait for the reconnect in future.
-            # In that case we need to make sure we do it at a layer above send_bytes otherwise
+            # In that case we need to make sure we do it at a layer above send_lines otherwise
             # we might encrypt payloads with the last sessions keys then wait for a new connection
             # to send them - and on that connection the keys would be different.
             # Also need to make sure that the new connection has chance to pair-verify before
@@ -113,7 +122,7 @@ async def send_bytes(self, payload: bytes) -> HttpResponse:
         timeout_handle = loop.call_at(loop.time() + 30, self._handle_timeout, result)
         timeout_expired = False
         try:
-            self.transport.write(payload)
+            self.transport.writelines(payload)
             return await result
         except (asyncio.TimeoutError, BaseException) as ex:
             # If we get a timeout or any other exception then we need to
@@ -188,18 +197,14 @@ async def send_bytes(self, payload: bytes) -> HttpResponse:
         while len(payload) > 0:
             current = payload[:1024]
             payload = payload[1024:]
-            len_bytes = len(current).to_bytes(2, byteorder="little")
+            len_bytes = PACK_UNSIGNED_SHORT_LITTLE(len(current))
             buffer.append(len_bytes)
             buffer.append(
-                self.encryptor.encrypt(
-                    len_bytes,
-                    PACK_NONCE(self.c2a_counter),
-                    bytes(current),
-                )
+                self.encryptor.encrypt(len_bytes, PACK_NONCE(self.c2a_counter), current)
             )
             self.c2a_counter += 1
 
-        return await super().send_bytes(b"".join(buffer))
+        return await self._send_lines(buffer)
 
     def data_received(self, data: bytes) -> None:
         """

tests/test_ip_pairing.py

@@ -157,7 +157,7 @@ async def test_put_characteristics_cancelled(pairing: IpPairing):
     characteristics = await pairing.put_characteristics([(1, 9, True)])
     characteristics = await pairing.get_characteristics([(1, 9)])
 
-    with mock.patch.object(pairing.connection.transport, "write"):
+    with mock.patch.object(pairing.connection.transport, "writelines"):
         task = asyncio.create_task(pairing.put_characteristics([(1, 9, False)]))
         await asyncio.sleep(0)
         for future in pairing.connection.protocol.result_cbs: