diff --git a/examples/telegram_group_calls.py b/examples/telegram_group_calls.py index 50c3ed4..7e5856a 100644 --- a/examples/telegram_group_calls.py +++ b/examples/telegram_group_calls.py @@ -117,7 +117,11 @@ async def send_audio_data(generator, input_filename): await asyncio.sleep(max(0.0, start + chunks / 100 - loop.time())) -async def main(client, input_peer, input_filename): +async def main(input_peer, input_filename): + client = pyrogram.Client( + os.environ.get('SESSION_NAME'), api_hash=os.environ.get('API_HASH'), api_id=os.environ.get('API_ID') + ) + await client.start() pc = webrtc.RTCPeerConnection() generator = webrtc.MediaStreamTrackGenerator('audio') @@ -151,15 +155,11 @@ def pre_update_processing(): await pyrogram.idle() sending.cancel() + await client.stop() if __name__ == '__main__': - pyro_client = pyrogram.Client( - os.environ.get('SESSION_NAME'), api_hash=os.environ.get('API_HASH'), api_id=os.environ.get('API_ID') - ) - pyro_client.start() - peer = input('Input peer:') filename = input('Input filename to play:') - asyncio.get_event_loop().run_until_complete(main(pyro_client, peer, filename)) + asyncio.run(main(peer, filename)) diff --git a/python-webrtc/python/webrtc/interfaces/rtc_peer_connection.py b/python-webrtc/python/webrtc/interfaces/rtc_peer_connection.py index 56dc67e..5cf695c 100644 --- a/python-webrtc/python/webrtc/interfaces/rtc_peer_connection.py +++ b/python-webrtc/python/webrtc/interfaces/rtc_peer_connection.py @@ -35,9 +35,9 @@ WebRTCObject, wrtc, ) -from webrtc.utils.callbacks_to_async import to_async from webrtc.utils.events import EventTarget from webrtc.utils.names import snake_case +from webrtc.utils.native_calls import call_native from webrtc.utils.operations import OperationsChain, later from webrtc.utils.task_queue import TaskQueue @@ -311,7 +311,7 @@ async def create_offer( self._apply_legacy_offer_option(MediaType.audio, offer_to_receive_audio) self._apply_legacy_offer_option(MediaType.video, offer_to_receive_video) await later() - return _init_of(await to_async(self._native_obj.createOffer)(ice_restart, voice_activity_detection)) + return _init_of(await call_native(self._native_obj.createOffer, ice_restart, voice_activity_detection)) async def create_answer(self, *, voice_activity_detection: bool = True) -> 'webrtc.RTCSessionDescriptionInit': """Initiates the creation an SDP answer to an offer received from a remote peer during the offer/answer @@ -332,7 +332,7 @@ async def create_answer(self, *, voice_activity_detection: bool = True) -> 'webr 'create an answer', RTCSignalingState.have_remote_offer, RTCSignalingState.have_local_pranswer ) await later() - return _init_of(await to_async(self._native_obj.createAnswer)(voice_activity_detection)) + return _init_of(await call_native(self._native_obj.createAnswer, voice_activity_detection)) async def set_local_description(self, description: Optional[_Description] = None) -> None: """Changes the local description associated with the connection. This description specifies the properties @@ -356,7 +356,7 @@ async def set_local_description(self, description: Optional[_Description] = None allowed = _LOCAL_DESCRIPTION_STATES[init.type] if init is not None else () self._check_state('set the local description', *allowed) await later() - await to_async(self._native_obj.setLocalDescription)(init) + await call_native(self._native_obj.setLocalDescription, init) self._completed_description() async def set_remote_description(self, description: _Description) -> None: @@ -379,7 +379,7 @@ async def set_remote_description(self, description: _Description) -> None: async with self._operation(): self._check_state('set the remote description', *_REMOTE_DESCRIPTION_STATES.get(init.type, ())) await later() - await to_async(self._native_obj.setRemoteDescription)(init) + await call_native(self._native_obj.setRemoteDescription, init) self._completed_description() def add_track( @@ -547,7 +547,7 @@ async def add_ice_candidate( if self.remote_description is None: raise InvalidStateError('A candidate can only be added once there is a remote description') await later() - await to_async(self._native_obj.addIceCandidate)(candidate_str, sdp_mid, sdp_m_line_index, ufrag) + await call_native(self._native_obj.addIceCandidate, candidate_str, sdp_mid, sdp_m_line_index, ufrag) def create_data_channel( self, @@ -626,7 +626,7 @@ async def get_stats(self, selector: Optional['webrtc.MediaStreamTrack'] = None) if len(matches) != 1: raise InvalidAccessError(f'{len(matches)} senders and receivers have the track, not exactly one') return await matches[0].get_stats() - return RTCStatsReport._from_native(await to_async(self._native_obj.getStats)(), self.get_receivers()) + return RTCStatsReport._from_native(await call_native(self._native_obj.getStats), self.get_receivers()) @staticmethod async def generate_certificate( diff --git a/python-webrtc/python/webrtc/interfaces/rtc_rtp_receiver.py b/python-webrtc/python/webrtc/interfaces/rtc_rtp_receiver.py index 474b7b1..a9c6c41 100644 --- a/python-webrtc/python/webrtc/interfaces/rtc_rtp_receiver.py +++ b/python-webrtc/python/webrtc/interfaces/rtc_rtp_receiver.py @@ -17,7 +17,7 @@ WebRTCObject, wrtc, ) -from webrtc.utils.callbacks_to_async import to_async +from webrtc.utils.native_calls import call_native if TYPE_CHECKING: import webrtc @@ -92,7 +92,7 @@ async def get_stats(self) -> 'webrtc.RTCStatsReport': Raises: :obj:`webrtc.InvalidStateError`: If the connection is closed. """ - return RTCStatsReport._from_native(await to_async(self._native_obj.getStats)(), [self]) + return RTCStatsReport._from_native(await call_native(self._native_obj.getStats), [self]) def get_synchronization_sources(self) -> List['webrtc.RTCRtpSynchronizationSource']: """Returns the synchronization sources (SSRCs) of the media received in the last 10 seconds. diff --git a/python-webrtc/python/webrtc/interfaces/rtc_rtp_sender.py b/python-webrtc/python/webrtc/interfaces/rtc_rtp_sender.py index a84282e..3912368 100644 --- a/python-webrtc/python/webrtc/interfaces/rtc_rtp_sender.py +++ b/python-webrtc/python/webrtc/interfaces/rtc_rtp_sender.py @@ -18,7 +18,7 @@ WebRTCObject, wrtc, ) -from webrtc.utils.callbacks_to_async import to_async +from webrtc.utils.native_calls import call_native from webrtc.utils.operations import later from webrtc.utils.task_queue import TaskQueue @@ -140,7 +140,7 @@ async def set_parameters( native.requestKeyFrame = bool(key_frame) last.encodings = encodings last.degradationPreference = parameters.degradation_preference - await to_async(self._native_obj.setParameters)(last) + await call_native(self._native_obj.setParameters, last) async def replace_track(self, track: Optional['webrtc.MediaStreamTrack']) -> None: """Replaces the track the sender sends, without negotiation. @@ -209,7 +209,7 @@ async def get_stats(self) -> 'webrtc.RTCStatsReport': Raises: :obj:`webrtc.InvalidStateError`: If the connection is closed. """ - return RTCStatsReport._from_native(await to_async(self._native_obj.getStats)()) + return RTCStatsReport._from_native(await call_native(self._native_obj.getStats)) #: Alias for :attr:`get_stats` getStats = get_stats diff --git a/python-webrtc/python/webrtc/utils/callbacks_to_async.py b/python-webrtc/python/webrtc/utils/callbacks_to_async.py deleted file mode 100644 index c6e09ca..0000000 --- a/python-webrtc/python/webrtc/utils/callbacks_to_async.py +++ /dev/null @@ -1,66 +0,0 @@ -# -# Copyright 2022 Il`ya (Marshal) . All rights reserved. -# -# Use of this source code is governed by a BSD-style license -# that can be found in the LICENSE.md file in the root of the project. -# - -import asyncio - -from webrtc.utils.task_queue import TaskQueue - - -class _QueuedEvent(asyncio.Event): - """An :obj:`asyncio.Event` set from any thread through the task queue of its loop, so the code awaiting the - result of an operation runs after the handlers of the events libwebrtc emitted before completing it.""" - - def __init__(self): - self.loop = asyncio.get_running_loop() - super().__init__() - - def set(self): - TaskQueue.of(self.loop).post(super().set, resumes=True, after_ready=True) - - -class _AsyncWrapper: - def __init__(self, func: callable): - self.__event = _QueuedEvent() - self.__func = func - - self.__args_for_run = [] - self.__kwargs_for_run = {} - - self.__result = self.__error = None - - def set(self): - self.__event.set() - - def _on_success(self, result=None): - self.__result = result - self.set() - - def _on_failure(self, error): - self.__error = error - self.set() - - async def run(self, timeout=10): - self.__func(self._on_success, self._on_failure, *self.__args_for_run, **self.__kwargs_for_run) - await asyncio.wait_for(self.__event.wait(), timeout) - - if self.__error: - # an RTCCallbackException - raise self.__error.toPython() - - return self.__result - - def __call__(self, *args, **kwargs): - self.__args_for_run = args - self.__kwargs_for_run = kwargs - - return self - - def __await__(self): - return self.run().__await__() - - -to_async = _AsyncWrapper diff --git a/python-webrtc/python/webrtc/utils/native_calls.py b/python-webrtc/python/webrtc/utils/native_calls.py new file mode 100644 index 0000000..ea66cfd --- /dev/null +++ b/python-webrtc/python/webrtc/utils/native_calls.py @@ -0,0 +1,50 @@ +# +# Copyright 2026 Ilya (Marshal) . All rights reserved. +# +# Use of this source code is governed by a BSD-style license +# that can be found in the LICENSE.md file in the root of the project. +# + +import asyncio +from typing import Any, Callable + +from webrtc.utils.task_queue import TaskQueue + + +async def call_native(method: Callable, *args) -> Any: + """Calls a native method taking success and failure callbacks, called from a libwebrtc thread, and awaits them. + + The result goes through the task queue of the loop, so the code awaiting it runs after the handlers of the events + libwebrtc emitted before completing the call. + + Args: + method (:obj:`callable`): The native method, called as ``method(on_success, on_failure, *args)``. + *args: Its arguments. + + Returns: + The result passed to ``on_success``, if any. + + Raises: + The error passed to ``on_failure``, as a Python exception. + """ + loop = asyncio.get_running_loop() + future = loop.create_future() + + def settle(result: Any, error: Any) -> None: + # the caller may have been canceled meanwhile + if future.done(): + return + if error is not None: + future.set_exception(error.toPython()) + else: + future.set_result(result) + + # libwebrtc threads, with the GIL held: only schedule + def on_success(result: Any = None) -> None: + TaskQueue.of(loop).post(settle, result, None, resumes=True, after_ready=True) + + def on_failure(error: Any) -> None: + TaskQueue.of(loop).post(settle, None, error, resumes=True, after_ready=True) + + method(on_success, on_failure, *args) + return await future diff --git a/tests/test_native_calls.py b/tests/test_native_calls.py new file mode 100644 index 0000000..9e0895c --- /dev/null +++ b/tests/test_native_calls.py @@ -0,0 +1,71 @@ +# +# Copyright 2026 Ilya (Marshal) . All rights reserved. +# +# Use of this source code is governed by a BSD-style license +# that can be found in the LICENSE.md file in the root of the project. +# + +"""call_native, which awaits native methods reporting to callbacks from libwebrtc threads.""" + +import asyncio +import threading + +import pytest + +from webrtc.utils.native_calls import call_native + + +class _Error: + def __init__(self, error): + self._error = error + + def toPython(self): + return self._error + + +def _later(callback, *args, delay=0.0): + threading.Timer(delay, callback, args).start() + + +@pytest.mark.asyncio +async def test_result(): + def method(on_success, on_failure, a, b): + _later(on_success, a + b) + + assert await call_native(method, 1, 2) == 3 + + +@pytest.mark.asyncio +async def test_no_result(): + assert await call_native(lambda on_success, on_failure: _later(on_success)) is None + + +@pytest.mark.asyncio +async def test_failure_is_raised_as_python_error(): + def method(on_success, on_failure): + _later(on_failure, _Error(ValueError('native'))) + + with pytest.raises(ValueError, match='native'): + await call_native(method) + + +@pytest.mark.asyncio +async def test_late_result_after_cancel_is_dropped(): + """A result arriving after the caller was canceled doesn't reach the loop's exception handler""" + loop = asyncio.get_running_loop() + errors = [] + loop.set_exception_handler(lambda loop, context: errors.append(context)) + settled = threading.Event() + + def method(on_success, on_failure): + def succeed(): + on_success('late') + settled.set() + + _later(succeed, delay=0.05) + + with pytest.raises(asyncio.TimeoutError): + await asyncio.wait_for(call_native(method), 0.01) + await asyncio.get_running_loop().run_in_executor(None, settled.wait) + await asyncio.sleep(0.01) + assert errors == []