fix(rtc): send and receive data packets without a new ctypes array type (#912)

Author: zek

livekit-rtc/livekit/rtc/participant.py

@@ -14,7 +14,6 @@
 
 from __future__ import annotations
 
-import ctypes
 import asyncio
 import functools
 import inspect
@@ -42,7 +41,7 @@
 from ._proto.track_pb2 import (
     ParticipantTrackPermission,
 )
-from ._utils import BroadcastQueue
+from ._utils import BroadcastQueue, get_address
 from .track import LocalTrack
 from .track_publication import (
     LocalTrackPublication,
@@ -336,11 +335,10 @@ async def publish_data(
             payload = payload.encode("utf-8")
 
         data_len = len(payload)
-        cdata = (ctypes.c_byte * data_len)(*payload)
 
         req = proto_ffi.FfiRequest()
         req.publish_data.local_participant_handle = self._ffi_handle.handle
-        req.publish_data.data_ptr = ctypes.addressof(cdata)
+        req.publish_data.data_ptr = get_address(payload)
         req.publish_data.data_len = data_len
         req.publish_data.reliable = reliable
         req.publish_data.topic = topic

livekit-rtc/livekit/rtc/room.py

@@ -1115,12 +1115,7 @@ def _on_room_event(self, event: proto_room.RoomEvent) -> None:
             if which_val == "user":
                 owned_buffer_info = packet.user.data
                 buffer_info = owned_buffer_info.data
-                native_data = ctypes.cast(
-                    buffer_info.data_ptr,
-                    ctypes.POINTER(ctypes.c_byte * buffer_info.data_len),
-                ).contents
-
-                data = bytes(native_data)
+                data = ctypes.string_at(buffer_info.data_ptr, buffer_info.data_len)
                 FfiHandle(owned_buffer_info.handle.id).dispose()
                 rparticipant = cast(
                     RemoteParticipant,

livekit-rtc/tests/test_dc.py

@@ -12,6 +12,8 @@
 from __future__ import annotations
 
 import asyncio
+import ctypes
+import gc
 from typing import Optional
 
 import pytest
@@ -277,3 +279,45 @@ async def test_data_targeted_with_topic() -> None:
     await _assert_no_data(other, other_collector)
 
     await asyncio.gather(sender.disconnect(), target.disconnect(), other.disconnect())
+
+
+@skip_if_no_credentials()  # type: ignore[untyped-decorator]
+@pytest.mark.asyncio
+async def test_publish_data_makes_no_ctypes_array_type() -> None:
+    """Sending and receiving payloads of new lengths makes no ctypes array type."""
+    room_name = unique_room_name("py-dc-array-type")
+
+    sender = rtc.Room()
+    receiver = rtc.Room()
+
+    await connect_room("sender", room_name, room=sender)
+    await connect_room("receiver", room_name, room=receiver)
+    await ensure_rooms_all_connected([sender, receiver])
+    await ensure_participants_visible(sender, ["receiver"])
+
+    collector = _DataCollector(receiver, sender_identity="sender")
+    lengths = range(1001, 1011)
+    payloads = [bytes([i]) * n for i, n in enumerate(lengths)]
+
+    gc.collect()
+    gc.disable()
+    try:
+        for payload in payloads:
+            await sender.local_participant.publish_data(payload)
+        await wait_until(
+            lambda: len(collector.packets) >= len(payloads),
+            message="receiver did not get all payloads",
+        )
+        array_types = [
+            o
+            for o in gc.get_objects()
+            if isinstance(o, type)
+            and issubclass(o, ctypes.Array)
+            and getattr(o, "_length_", None) in lengths
+        ]
+    finally:
+        gc.enable()
+    assert array_types == []
+    assert collector.payloads() == payloads
+
+    await asyncio.gather(sender.disconnect(), receiver.disconnect())