Replace to_async with call_native, drop its timeout

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))

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(

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.

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

python-webrtc/python/webrtc/utils/native_calls.py

@@ -0,0 +1,50 @@
+#
+#  Copyright 2026 Ilya (Marshal) <https://github.com/MarshalX>. 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

tests/test_native_calls.py

@@ -0,0 +1,71 @@
+#
+#  Copyright 2026 Ilya (Marshal) <https://github.com/MarshalX>. 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 == []