o i@s`ddlZddlZddlmZmZmZmZmZmZddl m Z edddZ Gdd d ee Z dS) N)CallableDictSetOptionalGenericTypeVar)loggerT_contraT) contravariantc@sveZdZdddZdeddfddZddedeedefd d Zddedeedefd d Z dededdfd dZ dS) EventEmitterreturnNcCs t|_dS)z< Initialize a new instance of EventEmitter. N)dict_events)selfr[/var/www/html/livekit_bhavya/venv/lib/python3.10/site-packages/livekit/rtc/event_emitter.py__init__ s zEventEmitter.__init__eventc Gs||jvra|j|}|D]T}z8t|}|j}tdd|D}|r+||ndd|D}t|} tt|| } |d| } || Wqt yPt y`t d|YqwdSdS)a Trigger all callbacks associated with the given event. Args: event (T): The event to emit. *args: Positional arguments to pass to the callbacks. Example: Basic usage of emit: ```python emitter = EventEmitter[str]() def greet(name): print(f"Hello, {name}!") emitter.on('greet', greet) emitter.emit('greet', 'Alice') # Output: Hello, Alice! ``` css|] }|j|jkVqdSN)kindVAR_POSITIONAL.0prrr -sz$EventEmitter.emit..cSs"g|] }|j|j|jfvr|qSr)rPOSITIONAL_ONLYPOSITIONAL_OR_KEYWORDrrrr 1s z%EventEmitter.emit..Nzfailed to emit event ) rcopyinspect signature parametersvaluesanylenmin TypeError Exceptionr exception) rrargs callablescallbacksigparams has_varargspositional_params num_paramsnum_args callback_argsrrremits0       zEventEmitter.emitr,csBdurfddSdtdtffdd }|S)a Register a callback to be called only once when the event is emitted. If a callback is provided, it registers the callback directly. If no callback is provided, it returns a decorator for use with function definitions. Args: event (T): The event to listen for. callback (Callable, optional): The callback to register. Defaults to None. Returns: Callable: The registered callback or a decorator if callback is None. Example: Using once with a direct callback: ```python emitter = EventEmitter[str]() def greet_once(name): print(f"Hello once, {name}!") emitter.once('greet', greet_once) emitter.emit('greet', 'Bob') # Output: Hello once, Bob! emitter.emit('greet', 'Bob') # No output, callback was removed after first call ``` Using once as a decorator: ```python emitter = EventEmitter[str]() @emitter.once('greet') def greet_once(name): print(f"Hello once, {name}!") emitter.emit('greet', 'Bob') # Output: Hello once, Bob! emitter.emit('greet', 'Bob') # No output ``` Ncs|i|dSr)off)r*kwargsr,r once_callbackrrrr8ks z(EventEmitter.once..once_callbackr,r c||Sr)oncer,rrrr decoratorr z$EventEmitter.once..decorator)onrrrr,r=rr7rr:@s ) zEventEmitter.oncecs`|dur"t|r tdjvrtj<j||Sdtdtffdd }|S)a` Register a callback to be called whenever the event is emitted. If a callback is provided, it registers the callback directly. If no callback is provided, it returns a decorator for use with function definitions. Args: event (T): The event to listen for. callback (Callable, optional): The callback to register. Defaults to None. Returns: Callable: The registered callback or a decorator if callback is None. Example: Using on with a direct callback: ```python emitter = EventEmitter[str]() def greet(name): print(f"Hello, {name}!") emitter.on('greet', greet) emitter.emit('greet', 'Charlie') # Output: Hello, Charlie! ``` Using on as a decorator: ```python emitter = EventEmitter[str]() @emitter.on('greet') def greet(name): print(f"Hello, {name}!") emitter.emit('greet', 'Charlie') # Output: Hello, Charlie! ``` NzsCannot register an async callback with `.on()`. Use `asyncio.create_task` within your synchronous callback instead.r,r cr9r)r?r;r<rrr=r>z"EventEmitter.on..decorator)asyncioiscoroutinefunction ValueErrorrsetaddrr@rr<rr?xs'   zEventEmitter.oncCs"||jvr|j||dSdS)a Unregister a callback from an event. Args: event (T): The event to stop listening to. callback (Callable): The callback to remove. Example: Removing a callback: ```python emitter = EventEmitter[str]() def greet(name): print(f"Hello, {name}!") emitter.on('greet', greet) emitter.off('greet', greet) emitter.emit('greet', 'Dave') # No output, callback was removed ``` N)rdiscard)rrr,rrrr5s zEventEmitter.off)r Nr) __name__ __module__ __qualname__rr r4rrr:r?r5rrrrr s  /89r ) r rAtypingrrrrrrlogr r r rrrrs