feat: datachannels (#3)

Author: theomonnom

examples/basic_room/room.py

@@ -13,14 +13,6 @@ async def main():
     audio_stream = None
     video_stream = None
 
-    logging.info("connecting to %s", URL)
-    try:
-        await room.connect(URL, TOKEN)
-        logging.info("connected to room %s", room.name)
-    except livekit.ConnectError as e:
-        logging.error("failed to connect to the room: %s", e)
-        return False
-
     @room.on("participant_connected")
     def on_participant_connected(participant: livekit.RemoteParticipant):
         logging.info(
@@ -65,8 +57,20 @@ def on_audio_frame(frame: livekit.AudioFrame):
     def on_track_unsubscribed(track: livekit.Track, publication: livekit.RemoteTrackPublication, participant: livekit.RemoteParticipant):
         logging.info("track unsubscribed: %s", publication.sid)
 
+    @room.on("data_received")
+    def on_data_received(data: bytes, kind: livekit.DataPacketKind, participant: livekit.Participant):
+        logging.info("received data from %s: %s", participant.identity, data)
+
     try:
+        logging.info("connecting to %s", URL)
+        await room.connect(URL, TOKEN)
+        logging.info("connected to room %s", room.name)
+
+        await room.local_participant.publish_data("hello world")
+
         await room.run()
+    except livekit.ConnectError as e:
+        logging.error("failed to connect to the room: %s", e)
     except asyncio.CancelledError:
         logging.info("closing the room")
         await room.close()

livekit/__init__.py

@@ -6,7 +6,7 @@
 from ._proto.video_frame_pb2 import (
     VideoRotation, VideoFormatType, VideoFrameBufferType)
 from ._proto.track_pb2 import (TrackKind, TrackSource, StreamState)
-from ._proto.room_pb2 import (TrackPublishOptions)
+from ._proto.room_pb2 import (TrackPublishOptions, DataPacketKind)
 
 from .room import (Room, ConnectError)
 from .participant import (Participant, LocalParticipant, RemoteParticipant)

livekit/lib/darwin/arm64/liblivekit_ffi.dylib

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:3506be9e3e8fc713fa1fc425a48dbed71a86cfddea6beab68a7721d59b4fa792
-size 27984241
+oid sha256:c4910845710a43ba065bf4e697b4f99c908167b636080f56f1ad2a5337cbc82f
+size 27965777

livekit/lib/darwin/x86_64/liblivekit_ffi.dylib

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:698dad5c7e83f01f5421805293c2335365dd40b3d5629d548ae82e4dbee71b8e
-size 32685760
+oid sha256:ffd4891c955d4743f5689ee5242a4c4b572a9c7a38e277bd035071d1a30463f4
+size 32682136

livekit/lib/linux/x86_64/liblivekit_ffi.so

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:aa2fdeeb7fbf1ffd001b6324f94e4e60fe5e2eb708f52445e0bf19c3b89bff4d
-size 36215176
+oid sha256:d9d16748f346ea8f607e0d6df6f0dece90d9284ee0d50e8eacdd66f5051ddeb0
+size 36168880

livekit/lib/windows/arm64/livekit_ffi.dll

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:35446a0fda6ec3b9f3f85200d3ed6ffbc588a531e3bd6ca0da431c5b4bb1c89e
+size 16523776

livekit/lib/windows/x86_64/livekit_ffi.dll

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:942a965d09f7a59e357ef6f72b402e4637a257281125a38c82f373252bc6db9f
-size 22181376
+oid sha256:3db42643671dee8f37be783e92308eb027c000736cb580fb17155951e24239b9
+size 22211584

livekit/participant.py

@@ -6,8 +6,9 @@
 from ._ffi_client import (FfiClient, FfiHandle)
 from typing import TYPE_CHECKING
 import weakref
+import ctypes
 import asyncio
-from livekit import TrackPublishOptions
+from livekit import (TrackPublishOptions, DataPacketKind)
 
 if TYPE_CHECKING:
     from livekit import (Room, Participant)
@@ -18,6 +19,11 @@ def __init__(self, message: str):
         self.message = message
 
 
+class PublishDataError(Exception):
+    def __init__(self, message: str):
+        self.message = message
+
+
 class Participant():
     def __init__(self, info: proto_participant.ParticipantInfo):
         self._info = info
@@ -45,6 +51,52 @@ def __init__(self, info: proto_participant.ParticipantInfo, room: weakref.ref['R
         super().__init__(info)
         self._room = room
 
+    async def publish_data(self,
+                           # TODO(theomonnom): Allow ctypes.Array as payload?
+                           payload: bytes or str,
+                           kind: DataPacketKind = DataPacketKind.KIND_RELIABLE,
+                           destination_sids: list[str] or list['RemoteParticipant'] = []):
+
+        room = self._room()
+        if room is None:
+            raise Exception('room is closed')
+
+        if isinstance(payload, str):
+            payload = payload.encode('utf-8')
+
+        data_len = len(payload)
+
+        cdata = (ctypes.c_byte * data_len)(*payload)
+
+        sids = []
+        for p in destination_sids:
+            if isinstance(p, RemoteParticipant):
+                sids.append(p.sid)
+            else:
+                sids.append(p)
+
+        req = proto_ffi.FfiRequest()
+        req.publish_data.room_handle.id = room._ffi_handle.handle
+        req.publish_data.data_ptr = ctypes.addressof(cdata)
+        req.publish_data.data_size = data_len
+        req.publish_data.kind = kind
+        req.publish_data.destination_sids.extend(sids)
+
+        ffi_client = FfiClient()
+        resp = ffi_client.request(req)
+        future = asyncio.Future()
+
+        @ffi_client.on('publish_data')
+        def on_publish_callback(cb: proto_room.PublishDataCallback):
+            if cb.async_id == resp.publish_data.async_id:
+                future.set_result(cb)
+                ffi_client.remove_listener(
+                    'publish_data', on_publish_callback)
+
+        resp: proto_room.PublishDataCallback = await future
+        if resp.error:
+            raise PublishDataError(resp.error)
+
     async def publish_track(self, track: Track, options: TrackPublishOptions) -> TrackPublication:
         if not isinstance(track, LocalAudioTrack) and not isinstance(track, LocalVideoTrack):
             raise Exception('cannot publish a remote track')

livekit/room.py

@@ -9,6 +9,7 @@
 from .track import (RemoteAudioTrack, RemoteVideoTrack)
 from livekit import TrackKind
 import weakref
+import ctypes
 
 
 class ConnectError(Exception):
@@ -31,6 +32,21 @@ def __del__(self):
         ffi_client = FfiClient()
         ffi_client.remove_listener('room_event', self._on_room_event)
 
+    @property
+    def sid(self) -> str:
+        return self._room_info.sid
+
+    @property
+    def name(self) -> str:
+        return self._room_info.name
+
+    @property
+    def metadata(self) -> str:
+        return self._room_info.metadata
+
+    def isconnected(self) -> bool:
+        return self._ffi_handle is not None
+
     async def connect(self, url: str, token: str):
         # TODO(theomonnom): We should be more flexible about the event loop
         ffi_client = FfiClient()
@@ -65,6 +81,9 @@ def on_connect_callback(cb: proto_room.ConnectCallback):
             self._create_remote_participant(participant_info)
 
     async def close(self):
+        if not self.isconnected():
+            return
+
         ffi_client = FfiClient()
 
         req = proto_ffi.FfiRequest()
@@ -134,6 +153,14 @@ def _on_room_event(self, event: proto_room.RoomEvent):
             track = publication._track
             publication._track = None
             self.emit('track_unsubscribed', track, publication, participant)
+        elif which == 'data_received':
+            participant = self.participants[event.data_received.participant_sid]
+            data = ctypes.cast(event.data_received.data_ptr,
+                               ctypes.POINTER(ctypes.c_byte * event.data_received.data_size)).contents
+            data = bytes(data)
+            FfiHandle(event.data_received.handle.id)
+            self.emit('data_received', data,
+                      event.data_received.kind, participant)
 
     def _create_remote_participant(self, info: proto_participant.ParticipantInfo) -> RemoteParticipant:
         if info.sid in self.participants:
@@ -147,15 +174,3 @@ def _create_remote_participant(self, info: proto_participant.ParticipantInfo) ->
             participant.tracks[publication.sid] = publication
 
         return participant
-
-    @property
-    def sid(self) -> str:
-        return self._room_info.sid
-
-    @property
-    def name(self) -> str:
-        return self._room_info.name
-
-    @property
-    def metadata(self) -> str:
-        return self._room_info.metadata