Skip to content

Commit 184d5a6

Browse files
davidzhaoclaude
andauthored
add support for RPC interceptors (#806)
Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
1 parent 11319e9 commit 184d5a6

5 files changed

Lines changed: 842 additions & 37 deletions

File tree

‎README.md‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -223,6 +223,29 @@ except Exception as e:
223223

224224
You may find it useful to adjust the `response_timeout` parameter, which indicates the amount of time you will wait for a response. We recommend keeping this value as low as possible while still satisfying the constraints of your application.
225225

226+
#### Intercepting RPC calls
227+
228+
An `RpcInterceptor` wraps every RPC the local participant performs or handles, which is useful for logging, tracing, or attaching metadata to payloads. Each method receives the call and a `next` continuation; return what `next` returns. Interceptors run in the order they were added, the first being outermost, and errors from the remote side or from your handler flow through them unchanged. On the incoming side the caller's `response_timeout` covers the whole chain, interceptors included.
229+
230+
```python
231+
class TimingInterceptor(rtc.RpcInterceptor):
232+
async def intercept_outgoing(self, call, next):
233+
start = time.perf_counter()
234+
try:
235+
return await next(call)
236+
finally:
237+
print(f"call {call.method} -> {call.destination_identity}: {time.perf_counter() - start:.3f}s")
238+
239+
async def intercept_incoming(self, invocation, next):
240+
start = time.perf_counter()
241+
try:
242+
return await next(invocation)
243+
finally:
244+
print(f"handled {invocation.method} from {invocation.caller_identity}: {time.perf_counter() - start:.3f}s")
245+
246+
room.local_participant.add_rpc_interceptor(TimingInterceptor())
247+
```
248+
226249
## Using local media devices
227250

228251
The `MediaDevices` class provides a high-level interface for working with local audio input (microphone) and output (speakers) devices. It's built on top of the `sounddevice` library and integrates seamlessly with LiveKit's audio processing features. In order to use `MediaDevices`, you must have the `sounddevice` library installed in your local Python environment, if it's not available, `MediaDevices` will not work.

‎livekit-rtc/livekit/rtc/__init__.py‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -115,7 +115,7 @@
115115
except Exception: # pragma: no cover - optional dependency (sounddevice)
116116
_HAS_MEDIA_DEVICES = False
117117
from .utils import combine_audio_frames
118-
from .rpc import RpcError, RpcInvocationData
118+
from .rpc import RpcCallInfo, RpcError, RpcInterceptor, RpcInvocationData
119119
from .synchronizer import AVSynchronizer
120120
from .data_stream import (
121121
TextStreamInfo,
@@ -211,6 +211,8 @@
211211
"AudioResamplerQuality",
212212
"RpcError",
213213
"RpcInvocationData",
214+
"RpcCallInfo",
215+
"RpcInterceptor",
214216
"EventEmitter",
215217
"combine_audio_frames",
216218
"AVSynchronizer",

‎livekit-rtc/livekit/rtc/participant.py‎

Lines changed: 184 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,8 @@
1616

1717
import ctypes
1818
import asyncio
19+
import functools
20+
import inspect
1921
import datetime
2022
import enum
2123
import os
@@ -48,11 +50,17 @@
4850
TrackPublication,
4951
)
5052
from .transcription import Transcription
51-
from .rpc import RpcError
53+
from .rpc import (
54+
RpcCallInfo,
55+
RpcError,
56+
RpcInterceptor,
57+
RpcInvocationData,
58+
_chain_incoming,
59+
_chain_outgoing,
60+
)
5261
from ._proto.rpc_pb2 import RpcMethodInvocationResponseRequest
5362
from .log import logger
5463

55-
from .rpc import RpcInvocationData
5664
from .data_stream import (
5765
TextStreamWriter,
5866
TextStreamInfo,
@@ -230,6 +238,26 @@ def disconnect_reason(
230238

231239

232240
RpcHandler = Callable[["RpcInvocationData"], Union[Awaitable[Optional[str]], Optional[str]]]
241+
242+
# how long a cancelled incoming RPC chain gets to unwind before the caller is answered
243+
# without it (the room's disconnect waits on these invocations)
244+
_RPC_CANCEL_UNWIND_TIMEOUT = 2.0
245+
246+
247+
def _observe_unwind(method: str, chain_task: "asyncio.Future[Optional[str]]") -> None:
248+
"""Consume what a cancelled chain raised while unwinding.
249+
250+
The caller has been (or is being) answered without it, so nobody awaits the chain
251+
anymore; unless its exception is retrieved here the loop reports it as never retrieved
252+
when the task is garbage collected.
253+
"""
254+
if chain_task.cancelled():
255+
return
256+
exc = chain_task.exception()
257+
if exc is not None:
258+
logger.warning("RPC handler for %s raised while being cancelled", method, exc_info=exc)
259+
260+
233261
F = TypeVar(
234262
"F", bound=Callable[[RpcInvocationData], Union[Awaitable[Optional[str]], Optional[str]]]
235263
)
@@ -247,6 +275,7 @@ def __init__(
247275
self._room_queue = room_queue
248276
self._track_publications: dict[str, LocalTrackPublication] = {}
249277
self._rpc_handlers: Dict[str, RpcHandler] = {}
278+
self._rpc_interceptors: List[RpcInterceptor] = []
250279
# Handles of data stream writers that have been opened but not yet
251280
# closed, so the room can drop them at disconnect. The FFI close
252281
# request consumes the handle (take_handle), so an entry is removed as
@@ -427,15 +456,27 @@ async def perform_rpc(
427456
Raises:
428457
RpcError: On failure. Details in `message`.
429458
"""
459+
call = RpcCallInfo(
460+
destination_identity=destination_identity,
461+
method=method,
462+
payload=payload,
463+
response_timeout=response_timeout,
464+
max_round_trip_latency=max_round_trip_latency,
465+
)
466+
# snapshot the interceptor list so add/remove during a call is well defined
467+
perform = _chain_outgoing(list(self._rpc_interceptors), self._perform_rpc_ffi)
468+
return await perform(call)
469+
470+
async def _perform_rpc_ffi(self, call: RpcCallInfo) -> str:
430471
req = proto_ffi.FfiRequest()
431472
req.perform_rpc.local_participant_handle = self._ffi_handle.handle
432-
req.perform_rpc.destination_identity = destination_identity
433-
req.perform_rpc.method = method
434-
req.perform_rpc.payload = payload
435-
if response_timeout is not None:
436-
req.perform_rpc.response_timeout_ms = int(response_timeout * 1000)
437-
if max_round_trip_latency is not None:
438-
req.perform_rpc.max_round_trip_latency_ms = int(max_round_trip_latency * 1000)
473+
req.perform_rpc.destination_identity = call.destination_identity
474+
req.perform_rpc.method = call.method
475+
req.perform_rpc.payload = call.payload
476+
if call.response_timeout is not None:
477+
req.perform_rpc.response_timeout_ms = int(call.response_timeout * 1000)
478+
if call.max_round_trip_latency is not None:
479+
req.perform_rpc.max_round_trip_latency_ms = int(call.max_round_trip_latency * 1000)
439480

440481
queue = FfiClient.instance.queue.subscribe()
441482
try:
@@ -449,6 +490,31 @@ async def perform_rpc(
449490

450491
return cast(str, cb.perform_rpc.payload)
451492

493+
def add_rpc_interceptor(self, interceptor: RpcInterceptor) -> None:
494+
"""
495+
Add an :class:`RpcInterceptor` that wraps every RPC this participant performs or
496+
handles. Interceptors run in the order they were added, the first being outermost.
497+
Adding the same instance twice is a no-op.
498+
499+
Args:
500+
interceptor (RpcInterceptor): The interceptor to add.
501+
"""
502+
# identity, not equality: two distinct interceptors that compare equal must coexist
503+
if not any(existing is interceptor for existing in self._rpc_interceptors):
504+
self._rpc_interceptors.append(interceptor)
505+
506+
def remove_rpc_interceptor(self, interceptor: RpcInterceptor) -> None:
507+
"""
508+
Remove a previously added :class:`RpcInterceptor`. Calls already in flight keep the
509+
chain they started with.
510+
511+
Args:
512+
interceptor (RpcInterceptor): The interceptor to remove.
513+
"""
514+
self._rpc_interceptors = [
515+
existing for existing in self._rpc_interceptors if existing is not interceptor
516+
]
517+
452518
def register_rpc_method(
453519
self,
454520
method_name: str,
@@ -552,33 +618,20 @@ async def _handle_rpc_method_invocation(
552618
response_error: Optional[RpcError] = None
553619
response_payload: Optional[str] = None
554620

555-
params = RpcInvocationData(request_id, caller_identity, payload, response_timeout)
556-
557-
handler = self._rpc_handlers.get(method)
621+
params = RpcInvocationData(
622+
request_id, caller_identity, payload, response_timeout, method=method
623+
)
558624

559-
if not handler:
560-
response_error = RpcError._built_in(RpcError.ErrorCode.UNSUPPORTED_METHOD)
561-
else:
562-
try:
563-
if asyncio.iscoroutinefunction(handler):
564-
try:
565-
response_payload = await asyncio.wait_for(
566-
handler(params), timeout=response_timeout
567-
)
568-
except asyncio.TimeoutError:
569-
raise RpcError._built_in(RpcError.ErrorCode.RESPONSE_TIMEOUT)
570-
except asyncio.CancelledError:
571-
raise RpcError._built_in(RpcError.ErrorCode.RECIPIENT_DISCONNECTED)
572-
else:
573-
response_payload = cast(Optional[str], handler(params))
574-
except RpcError as error:
575-
response_error = error
576-
except Exception:
577-
logger.exception(
578-
f"Uncaught error returned by RPC handler for {method}. "
579-
"Returning APPLICATION_ERROR instead. "
580-
)
581-
response_error = RpcError._built_in(RpcError.ErrorCode.APPLICATION_ERROR)
625+
try:
626+
response_payload = await self._run_incoming_chain(params)
627+
except RpcError as error:
628+
response_error = error
629+
except Exception:
630+
logger.exception(
631+
f"Uncaught error returned by RPC handler for {method}. "
632+
"Returning APPLICATION_ERROR instead. "
633+
)
634+
response_error = RpcError._built_in(RpcError.ErrorCode.APPLICATION_ERROR)
582635

583636
req = proto_ffi.FfiRequest(
584637
rpc_method_invocation_response=RpcMethodInvocationResponseRequest(
@@ -595,6 +648,102 @@ async def _handle_rpc_method_invocation(
595648
err = res.rpc_method_invocation_response.error
596649
logger.error(f"error sending rpc method invocation response: {err}")
597650

651+
async def _run_incoming_chain(self, invocation: RpcInvocationData) -> Optional[str]:
652+
"""Run the interceptor chain and the handler under the caller's response deadline.
653+
654+
The deadline covers the whole chain, so time an interceptor spends before or after
655+
``next`` counts against it; when it passes, the chain is cancelled and the caller
656+
gets ``RESPONSE_TIMEOUT``, whatever the chain raises while unwinding. Cancellation
657+
from outside (the room disconnecting) maps to ``RECIPIENT_DISCONNECTED``, as before.
658+
659+
A ``TimeoutError`` raised by the handler or an interceptor itself before the
660+
deadline (an HTTP client timing out, say) is not the response deadline: it
661+
propagates as an application error rather than being reported as
662+
``RESPONSE_TIMEOUT``.
663+
"""
664+
handle = _chain_incoming(list(self._rpc_interceptors), self._invoke_rpc_handler)
665+
loop = asyncio.get_running_loop()
666+
# ensure_future: `next` continuations are typed as Awaitable, not Coroutine
667+
chain_task: asyncio.Future[Optional[str]] = asyncio.ensure_future(handle(invocation))
668+
669+
# the deadline is recorded independently of whatever the chain raises while it
670+
# unwinds, so a TimeoutError (or anything else) from cancellation cleanup cannot be
671+
# mistaken for an application failure, nor a genuine in-chain timeout for the deadline
672+
deadline_fired = False
673+
674+
def _on_deadline() -> None:
675+
nonlocal deadline_fired
676+
# only a cancel the chain accepted counts: cancel() is False when the chain has
677+
# already finished, which can happen in the same loop iteration the timer fires
678+
# while this task has not resumed yet; that result is the caller's, not a timeout
679+
deadline_fired = chain_task.cancel()
680+
681+
deadline = loop.call_later(invocation.response_timeout, _on_deadline)
682+
try:
683+
# asyncio.wait never cancels what it waits on, so a cancel from outside (the room
684+
# disconnecting) is raised here at once with the chain untouched. Awaiting the
685+
# chain directly would forward the cancel to it and keep this task parked until
686+
# the chain finished, so a handler that ignored cancellation held up
687+
# room.disconnect() for as long as the caller's deadline. (Not asyncio.shield:
688+
# since Python 3.14 it reports the inner future's exception to the loop's
689+
# exception handler once its outer future was cancelled.)
690+
await asyncio.wait([chain_task])
691+
except asyncio.CancelledError:
692+
# cancelled from outside: stop the chain and let it unwind before answering the
693+
# caller, but not for long; this is the path room.disconnect() waits on
694+
chain_task.cancel()
695+
_, pending = await asyncio.wait([chain_task], timeout=_RPC_CANCEL_UNWIND_TIMEOUT)
696+
if pending:
697+
logger.warning(
698+
"RPC handler for %s did not stop within %.1fs of being cancelled; "
699+
"answering the caller without it",
700+
invocation.method,
701+
_RPC_CANCEL_UNWIND_TIMEOUT,
702+
)
703+
chain_task.add_done_callback(functools.partial(_observe_unwind, invocation.method))
704+
else:
705+
_observe_unwind(invocation.method, chain_task)
706+
raise RpcError._built_in(RpcError.ErrorCode.RECIPIENT_DISCONNECTED) from None
707+
finally:
708+
deadline.cancel()
709+
710+
if deadline_fired:
711+
# whatever the chain raised while unwinding is not the caller's business
712+
_observe_unwind(invocation.method, chain_task)
713+
raise RpcError._built_in(RpcError.ErrorCode.RESPONSE_TIMEOUT)
714+
if chain_task.cancelled():
715+
# nothing here cancelled it: an interceptor or the handler let a CancelledError
716+
# of its own escape. That is a failure of the handler, and it must still be
717+
# answered; result() would re-raise the CancelledError past the response code
718+
# and leave the caller waiting for its timeout.
719+
logger.warning(
720+
"RPC handler for %s raised CancelledError; returning APPLICATION_ERROR",
721+
invocation.method,
722+
)
723+
raise RpcError._built_in(RpcError.ErrorCode.APPLICATION_ERROR)
724+
# the chain's own outcome: its exception, if any, propagates unchanged
725+
return chain_task.result()
726+
727+
async def _invoke_rpc_handler(self, invocation: RpcInvocationData) -> Optional[str]:
728+
"""Run the registered handler for ``invocation`` (the innermost step of the chain).
729+
730+
Raises ``RpcError(UNSUPPORTED_METHOD)`` when nothing is registered; any exception
731+
from the handler propagates unchanged so interceptors can observe it before the
732+
caller's response is built. The response deadline is enforced by the caller around
733+
the whole chain.
734+
"""
735+
handler = self._rpc_handlers.get(invocation.method)
736+
if not handler:
737+
raise RpcError._built_in(RpcError.ErrorCode.UNSUPPORTED_METHOD)
738+
739+
# RpcHandler admits any callable returning a payload or an awaitable of one: a
740+
# coroutine function, but also a callable object with an async __call__ or a sync
741+
# wrapper handing back a coroutine, which iscoroutinefunction would not recognize
742+
result = handler(invocation)
743+
if inspect.isawaitable(result):
744+
result = await result
745+
return result
746+
598747
async def set_metadata(self, metadata: str) -> None:
599748
"""
600749
Set the metadata for the local participant.

0 commit comments

Comments
 (0)