diff --git a/livekit-rtc/livekit/rtc/participant.py b/livekit-rtc/livekit/rtc/participant.py index 37c54110..6c54d533 100644 --- a/livekit-rtc/livekit/rtc/participant.py +++ b/livekit-rtc/livekit/rtc/participant.py @@ -974,6 +974,56 @@ async def send_file( return _byte_stream_info_from_proto(cb.send_file.info) + async def send_bytes( + self, + data: bytes, + *, + topic: str = "", + name: str | None = None, + mime_type: str | None = None, + destination_identities: Optional[List[str]] = None, + attributes: Optional[Dict[str, str]] = None, + stream_id: str | None = None, + compress: bool = True, + ) -> ByteStreamInfo: + """ + Sends an in-memory buffer as a complete byte stream in one call. + Unlike stream_bytes(), the whole payload is known up front, so short payloads can be + sent as a single packet and larger ones compressed, when every recipient supports it. + `name` defaults to "unknown" and `mime_type` to "application/octet-stream". + """ + req = proto_ffi.FfiRequest() + req.send_bytes.local_participant_handle = self._ffi_handle.handle + req.send_bytes.bytes = data + options = req.send_bytes.options + options.topic = topic + if name is not None: + options.name = name + if mime_type is not None: + options.mime_type = mime_type + if attributes: + options.attributes.update(attributes) + if destination_identities: + options.destination_identities.extend(destination_identities) + if stream_id is not None: + options.id = stream_id + options.compress = compress + options.sender_identity = self.identity + + queue = FfiClient.instance.queue.subscribe() + try: + resp = FfiClient.instance.request(req) + cb: proto_ffi.FfiEvent = await queue.wait_for( + lambda e: e.send_bytes.async_id == resp.send_bytes.async_id + ) + finally: + FfiClient.instance.queue.unsubscribe(queue) + + if cb.send_bytes.HasField("error"): + raise StreamError(cb.send_bytes.error.description) + + return _byte_stream_info_from_proto(cb.send_bytes.info) + async def publish_data_track( self, *, diff --git a/tests/rtc/test_e2e_data_streams.py b/tests/rtc/test_e2e_data_streams.py index 2f39a42e..b8b492fd 100644 --- a/tests/rtc/test_e2e_data_streams.py +++ b/tests/rtc/test_e2e_data_streams.py @@ -192,6 +192,40 @@ async def read() -> None: await asyncio.gather(receiver.disconnect(), sender.disconnect()) +@pytest.mark.asyncio +@skip_if_no_credentials() # type: ignore[untyped-decorator] +async def test_send_bytes_small() -> None: + """Small in-memory buffer via send_bytes: info assertions + round trip.""" + receiver, sender = await connect_rooms(unique_room_name("ds-send-bytes-small")) + try: + bytes_to_send = b"hello world" + received: asyncio.Future[bytes] = asyncio.get_event_loop().create_future() + + def handler(reader: rtc.ByteStreamReader, participant_identity: str) -> None: + assert participant_identity == "sender" + + async def read() -> None: + chunks = [chunk async for chunk in reader] + received.set_result(b"".join(chunks)) + + asyncio.create_task(read()) + + receiver.register_byte_stream_handler("some-topic", handler) + + info = await sender.local_participant.send_bytes( + bytes_to_send, topic="some-topic", name="hello.bin" + ) + assert info.stream_id + assert info.topic == "some-topic" + assert info.name == "hello.bin" + assert info.size == len(bytes_to_send) + assert info.mime_type == "application/octet-stream" + + assert await asyncio.wait_for(received, timeout=5.0) == bytes_to_send + finally: + await asyncio.gather(receiver.disconnect(), sender.disconnect()) + + @pytest.mark.asyncio @skip_if_no_credentials() # type: ignore[untyped-decorator] async def test_send_text_large_compressible() -> None: