diff --git a/livekit-rtc/livekit/rtc/room.py b/livekit-rtc/livekit/rtc/room.py index 0601f9a4..fa9eb60f 100644 --- a/livekit-rtc/livekit/rtc/room.py +++ b/livekit-rtc/livekit/rtc/room.py @@ -189,6 +189,8 @@ def __init__( self._connection_state = ConnectionState.CONN_DISCONNECTED self._first_sid_future = asyncio.Future[str]() self._local_participant: LocalParticipant | None = None + self._active_speakers: list[Participant] = [] + self._is_connecting: bool = False # active incoming stream readers, keyed by FFI reader handle id self._text_stream_readers: Dict[int, TextStreamReader] = {} @@ -209,20 +211,60 @@ async def sid(self) -> str: Returns: str: The session ID of the room. + + Raises: + RuntimeError: If the room is disconnected and no SID was received. """ if self._info.sid: return self._info.sid + if self._connection_state == ConnectionState.CONN_DISCONNECTED and not self._is_connecting: + raise RuntimeError("cannot get sid of a disconnected room") + # This future is shared state for the Room. Caller-side timeouts, e.g. # asyncio.wait_for(room.sid, ...), should not cancel it for future waiters. return await asyncio.shield(self._first_sid_future) + async def get_sid(self, timeout: float | None = None) -> str: + """Asynchronously retrieves the session ID (SID) of the room with an optional timeout. + + Parameters: + timeout (float | None, optional): Maximum time in seconds to wait for the SID. + + Returns: + str: The session ID of the room. + + Raises: + RuntimeError: If the room is disconnected and no SID was received. + asyncio.TimeoutError: If the timeout expires before the SID is available. + """ + if timeout is None: + return await self.sid + return await asyncio.wait_for(self.sid, timeout) + def _resolve_first_sid(self, sid: str) -> None: # The room SID can arrive through the initial connect result or later # room update events. Resolve the first-SID waiter from any source that # carries it so waiters do not depend on event ordering. - if sid and not self._first_sid_future.done(): - self._first_sid_future.set_result(sid) + if sid: + self._info.sid = sid + if not self._first_sid_future.done(): + self._first_sid_future.set_result(sid) + + def _fail_first_sid(self, exc: BaseException) -> None: + if not self._first_sid_future.done(): + self._first_sid_future.set_exception(exc) + # Retrieve the exception to suppress unretrieved future warnings if unawaited + self._first_sid_future.exception() + + @property + def active_speakers(self) -> list[Participant]: + """Gets the list of currently active speakers in the room. + + Returns: + list[Participant]: The list of active speakers. + """ + return list(self._active_speakers) @property def local_participant(self) -> LocalParticipant: @@ -469,6 +511,9 @@ def on_participant_connected(participant): """ self._server_url = url self._token = token + self._is_connecting = True + if self._first_sid_future.done() and not self._info.sid: + self._first_sid_future = asyncio.Future[str]() req = proto_ffi.FfiRequest() req.connect.url = url req.connect.token = token @@ -560,43 +605,55 @@ def on_participant_connected(participant): ), ) - queue = FfiClient.instance.queue.subscribe() try: - resp = FfiClient.instance.request(req) - cb: proto_ffi.FfiEvent = await queue.wait_for( - lambda e: e.connect.async_id == resp.connect.async_id - ) - finally: - FfiClient.instance.queue.unsubscribe(queue) - - if cb.connect.error: - FfiClient.instance.queue.unsubscribe(self._ffi_queue) - raise ConnectError(cb.connect.error) + queue = FfiClient.instance.queue.subscribe() + try: + resp = FfiClient.instance.request(req) + cb: proto_ffi.FfiEvent = await queue.wait_for( + lambda e: e.connect.async_id == resp.connect.async_id + ) + finally: + FfiClient.instance.queue.unsubscribe(queue) - self._ffi_handle = FfiHandle(cb.connect.result.room.handle.id) + if cb.connect.error: + FfiClient.instance.queue.unsubscribe(self._ffi_queue) + raise ConnectError(cb.connect.error) - self._e2ee_manager = E2EEManager( - self._ffi_handle.handle, options.encryption or options.e2ee - ) + self._ffi_handle = FfiHandle(cb.connect.result.room.handle.id) - self._info = cb.connect.result.room.info - self._resolve_first_sid(self._info.sid) - self._connection_state = ConnectionState.CONN_CONNECTED - - self._local_participant = LocalParticipant( - self._room_queue, cb.connect.result.local_participant - ) + self._e2ee_manager = E2EEManager( + self._ffi_handle.handle, options.encryption or options.e2ee + ) - for pt in cb.connect.result.participants: - rp = self._create_remote_participant(pt.participant) + self._info = cb.connect.result.room.info + self._resolve_first_sid(self._info.sid) + self._connection_state = ConnectionState.CONN_CONNECTED - # add the initial remote participant tracks - for owned_publication_info in pt.publications: - publication = RemoteTrackPublication(owned_publication_info) - rp._track_publications[publication.sid] = publication + self._local_participant = LocalParticipant( + self._room_queue, cb.connect.result.local_participant + ) - # start listening to room events - self._task = self._loop.create_task(self._listen_task()) + for pt in cb.connect.result.participants: + rp = self._create_remote_participant(pt.participant) + + # add the initial remote participant tracks + for owned_publication_info in pt.publications: + publication = RemoteTrackPublication(owned_publication_info) + rp._track_publications[publication.sid] = publication + + # start listening to room events + self._task = self._loop.create_task(self._listen_task()) + except BaseException as e: + if self._ffi_handle is None and hasattr(self, "_ffi_queue"): + FfiClient.instance.queue.unsubscribe(self._ffi_queue) + self._fail_first_sid( + RuntimeError("room connection was cancelled") + if isinstance(e, asyncio.CancelledError) + else e + ) + raise + finally: + self._is_connecting = False # unsubscribe from the FFI queue once the listen task ends. # disconnect() unsubscribes too, but it early-returns when the @@ -713,6 +770,10 @@ async def disconnect( if self._connection_state != ConnectionState.CONN_DISCONNECTED: self.local_participant._info.disconnect_reason = reason self._connection_state = ConnectionState.CONN_DISCONNECTED + self._fail_first_sid(RuntimeError("room disconnected before receiving sid")) + if self._active_speakers: + self._active_speakers = [] + self.emit("active_speakers_changed", []) self.emit("connection_state_changed", self._connection_state) self.emit("disconnected", reason) @@ -744,6 +805,7 @@ async def _listen_task(self) -> None: await self._drain_rpc_invocation_tasks() self._error_stream_readers() self._dispose_open_stream_writers() + self._fail_first_sid(RuntimeError("room disconnected before receiving sid")) def _on_rpc_method_invocation(self, rpc_invocation: RpcMethodInvocationEvent) -> None: if self._local_participant is None: @@ -770,9 +832,15 @@ def _on_room_event(self, event: proto_room.RoomEvent) -> None: self.emit("participant_connected", rparticipant) elif which == "participant_disconnected": identity = event.participant_disconnected.participant_identity - rparticipant = self._remote_participants.pop(identity) - rparticipant._info.disconnect_reason = event.participant_disconnected.disconnect_reason - self.emit("participant_disconnected", rparticipant) + if identity in self._remote_participants: + rparticipant = self._remote_participants.pop(identity) + rparticipant._info.disconnect_reason = ( + event.participant_disconnected.disconnect_reason + ) + self.emit("participant_disconnected", rparticipant) + if any(p.identity == identity for p in self._active_speakers): + self._active_speakers = [p for p in self._active_speakers if p.identity != identity] + self.emit("active_speakers_changed", list(self._active_speakers)) elif which == "participant_active": rp = self._retrieve_remote_participant(event.participant_active.participant_identity) if rp: @@ -982,10 +1050,12 @@ def _on_room_event(self, event: proto_room.RoomEvent) -> None: # TODO: pass participant identity for identity in event.active_speakers_changed.participant_identities: participant = self._retrieve_participant(identity) - assert isinstance(participant, Participant) - speakers.append(participant) + if participant is not None: + speakers.append(participant) - self.emit("active_speakers_changed", speakers) + if [p.identity for p in speakers] != [p.identity for p in self._active_speakers]: + self._active_speakers = speakers + self.emit("active_speakers_changed", list(speakers)) elif which == "room_metadata_changed": old_metadata = self.metadata self._info.metadata = event.room_metadata_changed.metadata @@ -1126,6 +1196,10 @@ def _on_room_event(self, event: proto_room.RoomEvent) -> None: self._connection_state = connection_state self.emit("connection_state_changed", connection_state) elif which == "disconnected": + self._fail_first_sid(RuntimeError("room disconnected before receiving sid")) + if self._active_speakers: + self._active_speakers = [] + self.emit("active_speakers_changed", []) self.emit("disconnected", event.disconnected.reason) elif which == "reconnecting": self.emit("reconnecting") @@ -1243,8 +1317,12 @@ def _retrieve_remote_participant(self, identity: str) -> Optional[RemoteParticip def _retrieve_participant(self, identity: str) -> Optional[Participant]: """Retrieve a local or remote participant by identity""" - if identity and identity == self.local_participant.identity: - return self.local_participant + if ( + identity + and self._local_participant is not None + and identity == self._local_participant.identity + ): + return self._local_participant return self._retrieve_remote_participant(identity) @@ -1260,7 +1338,13 @@ def _create_remote_participant( def __repr__(self) -> str: sid = "unknown" - if self._first_sid_future.done(): + if self._info.sid: + sid = self._info.sid + elif ( + self._first_sid_future.done() + and not self._first_sid_future.cancelled() + and self._first_sid_future.exception() is None + ): sid = self._first_sid_future.result() return f"rtc.Room(sid={sid}, name={self.name}, metadata={self.metadata}, connection_state={ConnectionState.Name(self._connection_state)})" diff --git a/livekit-rtc/tests/test_ffi_handle_lifecycle.py b/livekit-rtc/tests/test_ffi_handle_lifecycle.py index 08750bfa..1d4a2a58 100644 --- a/livekit-rtc/tests/test_ffi_handle_lifecycle.py +++ b/livekit-rtc/tests/test_ffi_handle_lifecycle.py @@ -14,7 +14,6 @@ from __future__ import annotations -import gc import subprocess import sys @@ -43,25 +42,46 @@ def test_active_runtime_invalid_handle_asserts() -> None: def test_live_handles_marked_consumed_and_no_assert_after_dispose() -> None: - handle_id = _allocate_real_handle() - handle = FfiHandle(handle_id) - assert not handle.disposed + code = """ +import gc +from livekit.rtc._ffi_client import FfiClient, FfiHandle +from livekit.rtc._proto import audio_frame_pb2, ffi_pb2 + +req = ffi_pb2.FfiRequest() +req.new_audio_source.type = audio_frame_pb2.AUDIO_SOURCE_NATIVE +req.new_audio_source.sample_rate = 48000 +req.new_audio_source.num_channels = 1 +req.new_audio_source.queue_size_ms = 100 +resp = FfiClient.instance.request(req) +handle_id = int(resp.new_audio_source.source.handle.id) +assert handle_id != 0 + +handle = FfiHandle(handle_id) +assert not handle.disposed - # Global dispose marks all live handles consumed and sets _ffi_disposed - FfiClient.instance._dispose() - assert handle.disposed +# Global dispose marks all live handles consumed and sets _ffi_disposed +FfiClient.instance._dispose() +assert handle.disposed - # Explicit dispose or __del__ after global dispose should be a no-op and not assert - handle.dispose() - handle.__del__() +# Explicit dispose or __del__ after global dispose should be a no-op and not assert +handle.dispose() +handle.__del__() - # Newly instantiated handle after global dispose also safely no-ops without assert - late_handle = FfiHandle(12345) - late_handle.dispose() +# Newly instantiated handle after global dispose also safely no-ops without assert +late_handle = FfiHandle(12345) +late_handle.dispose() - del handle - del late_handle - gc.collect() +del handle +del late_handle +gc.collect() +""" + result = subprocess.run( + [sys.executable, "-c", code], + capture_output=True, + text=True, + ) + assert result.returncode == 0, f"Subprocess failed:\n{result.stderr}" + assert "AssertionError" not in result.stderr def test_subprocess_exit_with_undisposed_handle_has_no_assertion_error() -> None: diff --git a/tests/rtc/test_room_sid_and_speakers.py b/tests/rtc/test_room_sid_and_speakers.py new file mode 100644 index 00000000..7d17203e --- /dev/null +++ b/tests/rtc/test_room_sid_and_speakers.py @@ -0,0 +1,254 @@ +# Copyright 2026 LiveKit, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import asyncio + +import pytest + +from livekit import rtc +from livekit.rtc import ConnectionState, Participant, RemoteParticipant +from livekit.rtc._proto import ffi_pb2 as proto_ffi +from livekit.rtc._proto import participant_pb2 as proto_participant +from livekit.rtc._proto import room_pb2 as proto_room + + +@pytest.mark.asyncio +async def test_room_sid_disconnected_raises_runtime_error() -> None: + room = rtc.Room() + with pytest.raises(RuntimeError, match="cannot get sid of a disconnected room"): + await room.sid + + +@pytest.mark.asyncio +async def test_room_get_sid_timeout() -> None: + room = rtc.Room() + room._connection_state = ConnectionState.CONN_CONNECTED + with pytest.raises(asyncio.TimeoutError): + await room.get_sid(timeout=0.05) + + +@pytest.mark.asyncio +async def test_room_sid_resolution_and_get_sid() -> None: + room = rtc.Room() + room._connection_state = ConnectionState.CONN_CONNECTED + + async def resolve_soon() -> None: + await asyncio.sleep(0.01) + room._resolve_first_sid("RM_test123") + + task = asyncio.create_task(resolve_soon()) + sid = await room.get_sid(timeout=1.0) + await task + assert sid == "RM_test123" + assert await room.sid == "RM_test123" + assert "RM_test123" in repr(room) + + +@pytest.mark.asyncio +async def test_room_sid_failed_on_disconnect() -> None: + room = rtc.Room() + room._connection_state = ConnectionState.CONN_CONNECTED + + async def wait_sid() -> None: + await room.sid + + wait_task = asyncio.create_task(wait_sid()) + await asyncio.sleep(0.01) + + # Room disconnects before receiving SID + room._fail_first_sid(RuntimeError("room disconnected before receiving sid")) + + with pytest.raises(RuntimeError, match="room disconnected before receiving sid"): + await wait_task + + +def _create_mock_remote_participant(identity: str, sid: str) -> RemoteParticipant: + owned_info = proto_participant.OwnedParticipant() + owned_info.handle.id = 0 + owned_info.info.sid = sid + owned_info.info.identity = identity + p = RemoteParticipant(owned_info) + p._ffi_handle.mark_consumed() + return p + + +@pytest.mark.asyncio +async def test_active_speakers_deduplication() -> None: + room = rtc.Room() + p1 = _create_mock_remote_participant("user-1", "PA_1") + p2 = _create_mock_remote_participant("user-2", "PA_2") + room._remote_participants["user-1"] = p1 + room._remote_participants["user-2"] = p2 + + emitted_speaker_identities: list[list[str]] = [] + + def on_speakers_changed(speakers: list[Participant]) -> None: + emitted_speaker_identities.append([p.identity for p in speakers]) + + room.on("active_speakers_changed", on_speakers_changed) + + # 1. First event: user-1 is speaking + ev1 = proto_room.RoomEvent() + ev1.active_speakers_changed.participant_identities.append("user-1") + room._on_room_event(ev1) + + assert len(emitted_speaker_identities) == 1 + assert emitted_speaker_identities[-1] == ["user-1"] + assert [p.identity for p in room.active_speakers] == ["user-1"] + + # 2. Duplicate event: user-1 still speaking (no change) -> should NOT emit + ev2 = proto_room.RoomEvent() + ev2.active_speakers_changed.participant_identities.append("user-1") + room._on_room_event(ev2) + + assert len(emitted_speaker_identities) == 1 + + # 3. New event: user-1 and user-2 speaking -> should emit + ev3 = proto_room.RoomEvent() + ev3.active_speakers_changed.participant_identities.extend(["user-1", "user-2"]) + room._on_room_event(ev3) + + assert len(emitted_speaker_identities) == 2 + assert emitted_speaker_identities[-1] == ["user-1", "user-2"] + assert [p.identity for p in room.active_speakers] == ["user-1", "user-2"] + + # 4. Speakers stop: blank event -> should emit empty list + ev4 = proto_room.RoomEvent() + ev4.active_speakers_changed.SetInParent() + room._on_room_event(ev4) + + assert len(emitted_speaker_identities) == 3 + assert emitted_speaker_identities[-1] == [] + assert room.active_speakers == [] + + # 5. Subsequent blank event -> should NOT emit again + ev5 = proto_room.RoomEvent() + ev5.active_speakers_changed.SetInParent() + room._on_room_event(ev5) + + assert len(emitted_speaker_identities) == 3 + + +@pytest.mark.asyncio +async def test_participant_disconnected_updates_active_speakers() -> None: + room = rtc.Room() + p1 = _create_mock_remote_participant("user-1", "PA_1") + p2 = _create_mock_remote_participant("user-2", "PA_2") + room._remote_participants["user-1"] = p1 + room._remote_participants["user-2"] = p2 + + emitted_speaker_identities: list[list[str]] = [] + + def on_speakers_changed(speakers: list[Participant]) -> None: + emitted_speaker_identities.append([p.identity for p in speakers]) + + room.on("active_speakers_changed", on_speakers_changed) + + # Both users speaking + ev = proto_room.RoomEvent() + ev.active_speakers_changed.participant_identities.extend(["user-1", "user-2"]) + room._on_room_event(ev) + assert len(emitted_speaker_identities) == 1 + + # user-1 disconnects + disc_ev = proto_room.RoomEvent() + disc_ev.participant_disconnected.participant_identity = "user-1" + room._on_room_event(disc_ev) + + assert len(emitted_speaker_identities) == 2 + assert emitted_speaker_identities[-1] == ["user-2"] + assert [p.identity for p in room.active_speakers] == ["user-2"] + + +@pytest.mark.asyncio +async def test_room_disconnect_clears_active_speakers() -> None: + room = rtc.Room() + p1 = _create_mock_remote_participant("user-1", "PA_1") + room._remote_participants["user-1"] = p1 + + emitted_speaker_identities: list[list[str]] = [] + + def on_speakers_changed(speakers: list[Participant]) -> None: + emitted_speaker_identities.append([p.identity for p in speakers]) + + room.on("active_speakers_changed", on_speakers_changed) + + # user-1 speaking + ev = proto_room.RoomEvent() + ev.active_speakers_changed.participant_identities.append("user-1") + room._on_room_event(ev) + assert len(emitted_speaker_identities) == 1 + assert [p.identity for p in room.active_speakers] == ["user-1"] + + # Disconnect event arrives + disc_ev = proto_room.RoomEvent() + disc_ev.disconnected.reason = proto_participant.DisconnectReason.UNKNOWN_REASON + room._on_room_event(disc_ev) + + assert len(emitted_speaker_identities) == 2 + assert emitted_speaker_identities[-1] == [] + assert room.active_speakers == [] + + +@pytest.mark.asyncio +async def test_active_speakers_listener_mutation_does_not_corrupt_cache() -> None: + room = rtc.Room() + p1 = _create_mock_remote_participant("user-1", "PA_1") + room._remote_participants["user-1"] = p1 + + def mutating_listener(speakers: list[Participant]) -> None: + speakers.clear() + + room.on("active_speakers_changed", mutating_listener) + + ev = proto_room.RoomEvent() + ev.active_speakers_changed.participant_identities.append("user-1") + room._on_room_event(ev) + + assert len(room.active_speakers) == 1 + assert room.active_speakers[0].identity == "user-1" + + +@pytest.mark.asyncio +async def test_connect_cancellation_fails_sid_future_and_cleans_up( + monkeypatch: pytest.MonkeyPatch, +) -> None: + room = rtc.Room() + + # Mock FfiClient request to never answer connect response + def mock_request(req: proto_ffi.FfiRequest) -> proto_ffi.FfiResponse: + resp = proto_ffi.FfiResponse() + resp.connect.async_id = 9999 + return resp + + monkeypatch.setattr(rtc._ffi_client.FfiClient.instance, "request", mock_request) + + connect_task = asyncio.create_task(room.connect("ws://dummy", "dummy_token")) + await asyncio.sleep(0.01) + + assert room._is_connecting + + async def wait_sid() -> None: + await room.sid + + sid_task = asyncio.create_task(wait_sid()) + await asyncio.sleep(0.01) + + connect_task.cancel() + with pytest.raises(asyncio.CancelledError): + await connect_task + + assert not room._is_connecting + with pytest.raises(RuntimeError, match="room connection was cancelled"): + await sid_task