diff --git a/livekit-rtc/livekit/rtc/participant.py b/livekit-rtc/livekit/rtc/participant.py index 6c54d533..07b038b3 100644 --- a/livekit-rtc/livekit/rtc/participant.py +++ b/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 diff --git a/livekit-rtc/livekit/rtc/room.py b/livekit-rtc/livekit/rtc/room.py index e58321eb..e533d205 100644 --- a/livekit-rtc/livekit/rtc/room.py +++ b/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, diff --git a/livekit-rtc/tests/test_dc.py b/livekit-rtc/tests/test_dc.py index aff64e3f..7c4ddb0e 100644 --- a/livekit-rtc/tests/test_dc.py +++ b/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())