Add data stream support (#347)

Co-authored-by: github-actions <41898282+github-actions[bot]@users.noreply.github.com>
Co-authored-by: David Zhao <[email protected]>

Author: 3 people

.gitattributes

@@ -1,5 +1,6 @@
 **/*.dll filter=lfs diff=lfs merge=lfs -text
 **/*.so filter=lfs diff=lfs merge=lfs -text
 **/*.dylib filter=lfs diff=lfs merge=lfs -text
+**/*.jpg filter=lfs diff=lfs merge=lfs -text
 livekit-protocol/livekit/protocol/** linguist-generated=true
 livekit-rtc/livekit/rtc/_proto/** linguist-generated=true

examples/data-streams/data_streams.py

@@ -0,0 +1,104 @@
+import os
+import logging
+import asyncio
+from signal import SIGINT, SIGTERM
+from livekit import rtc
+
+# Set the following environment variables with your own values
+TOKEN = os.environ.get("LIVEKIT_TOKEN")
+URL = os.environ.get("LIVEKIT_URL")
+
+
+async def main(room: rtc.Room):
+    logging.basicConfig(level=logging.INFO)
+    logger = logging.getLogger(__name__)
+
+    async def greetParticipant(identity: str):
+        text_writer = await room.local_participant.stream_text(
+            destination_identities=[identity], topic="chat"
+        )
+        for char in "Hi! Just a friendly message":
+            await text_writer.write(char)
+        await text_writer.aclose()
+
+        await room.local_participant.send_file(
+            "./green_tree_python.jpg",
+            destination_identities=[identity],
+            topic="welcome",
+        )
+
+    async def on_chat_message_received(
+        reader: rtc.TextStreamReader, participant_identity: str
+    ):
+        full_text = await reader.read_all()
+        logger.info(
+            "Received chat message from %s: '%s'", participant_identity, full_text
+        )
+
+    async def on_welcome_image_received(
+        reader: rtc.ByteStreamReader, participant_identity: str
+    ):
+        logger.info(
+            "Received image from %s: '%s'", participant_identity, reader.info["name"]
+        )
+        with open(reader.info["name"], mode="wb") as f:
+            async for chunk in reader:
+                f.write(chunk)
+
+            f.close()
+
+    @room.on("participant_connected")
+    def on_participant_connected(participant: rtc.RemoteParticipant):
+        logger.info(
+            "participant connected: %s %s", participant.sid, participant.identity
+        )
+        asyncio.create_task(greetParticipant(participant.identity))
+
+    room.set_text_stream_handler(
+        lambda reader, participant_identity: asyncio.create_task(
+            on_chat_message_received(reader, participant_identity)
+        ),
+        "chat",
+    )
+
+    room.set_byte_stream_handler(
+        lambda reader, participant_identity: asyncio.create_task(
+            on_welcome_image_received(reader, participant_identity)
+        ),
+        "welcome",
+    )
+
+    # By default, autosubscribe is enabled. The participant will be subscribed to
+    # all published tracks in the room
+    await room.connect(URL, TOKEN)
+    logger.info("connected to room %s", room.name)
+
+    for identity, participant in room.remote_participants.items():
+        logger.info("Sending a welcome message to %s", identity)
+        await greetParticipant(participant.identity)
+
+
+if __name__ == "__main__":
+    logging.basicConfig(
+        level=logging.INFO,
+        handlers=[
+            logging.FileHandler("data_stream_example.log"),
+            logging.StreamHandler(),
+        ],
+    )
+
+    loop = asyncio.get_event_loop()
+    room = rtc.Room(loop=loop)
+
+    async def cleanup():
+        await room.disconnect()
+        loop.stop()
+
+    asyncio.ensure_future(main(room))
+    for signal in [SIGINT, SIGTERM]:
+        loop.add_signal_handler(signal, lambda: asyncio.ensure_future(cleanup()))
+
+    try:
+        loop.run_forever()
+    finally:
+        loop.close()

examples/data-streams/green_tree_python.jpg

@@ -74,6 +74,15 @@
 from .utils import combine_audio_frames
 from .rpc import RpcError, RpcInvocationData
 from .synchronizer import AVSynchronizer
+from .data_stream import (
+    TextStreamInfo,
+    TextStreamUpdate,
+    ByteStreamInfo,
+    TextStreamReader,
+    TextStreamWriter,
+    ByteStreamWriter,
+    ByteStreamReader,
+)
 
 __all__ = [
     "ConnectionQuality",
@@ -140,5 +149,12 @@
     "EventEmitter",
     "combine_audio_frames",
     "AVSynchronizer",
+    "TextStreamUpdate",
+    "TextStreamInfo",
+    "ByteStreamInfo",
+    "TextStreamReader",
+    "TextStreamWriter",
+    "ByteStreamReader",
+    "ByteStreamWriter",
     "__version__",
 ]

livekit-rtc/livekit/rtc/__init__.py

@@ -130,3 +130,15 @@ def generate_random_base62(length=12):
     """
     global _base62_characters
     return "".join(random.choice(_base62_characters) for _ in range(length))
+
+
+# adapted from https://stackoverflow.com/a/6043797
+def split_utf8(s: str, n: int):
+    """Split UTF-8 s into chunks of maximum length n."""
+    while len(s) > n:
+        k = n
+        while (ord(s[k]) & 0xC0) == 0x80:
+            k -= 1
+        yield s[:k]
+        s = s[k:]
+    yield s