Module `livekit.rtc.event_emitter`

Classes

class EventEmitter

Abstract base class for generic types.

On Python 3.12 and newer, generic classes implicitly inherit from Generic when they declare a parameter list after the class's name::

class Mapping[KT, VT]:
    def __getitem__(self, key: KT) -> VT:
        ...
    # Etc.

On older versions of Python, however, generic classes have to explicitly inherit from Generic.

After a class has been declared to be generic, it can then be used as follows::

def lookup_name[KT, VT](mapping: Mapping[KT, VT], key: KT, default: VT) -> VT:
    try:
        return mapping[key]
    except KeyError:
        return default

Initialize a new instance of EventEmitter.

Expand source code

class EventEmitter(Generic[T]):
    def __init__(self) -> None:
        """
        Initialize a new instance of EventEmitter.
        """
        self._events: Dict[T, Set[Callable]] = dict()

    def emit(self, event: T, *args) -> None:
        """
        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!
            ```
        """
        if event in self._events:
            callables = self._events[event].copy()
            for callback in callables:
                try:
                    sig = inspect.signature(callback)
                    params = sig.parameters.values()

                    has_varargs = any(p.kind == p.VAR_POSITIONAL for p in params)
                    if has_varargs:
                        callback(*args)
                    else:
                        positional_params = [
                            p
                            for p in params
                            if p.kind in (p.POSITIONAL_ONLY, p.POSITIONAL_OR_KEYWORD)
                        ]
                        num_params = len(positional_params)
                        num_args = min(len(args), num_params)
                        callback_args = args[:num_args]

                        callback(*callback_args)
                except Exception:
                    logger.exception(f"failed to emit event {event}")

    def once(self, event: T, callback: Optional[Callable] = None) -> Callable:
        """
        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
            ```
        """
        if callback is not None:

            def once_callback(*args, **kwargs):
                self.off(event, once_callback)
                callback(*args, **kwargs)

            return self.on(event, once_callback)
        else:

            def decorator(callback: Callable) -> Callable:
                self.once(event, callback)
                return callback

            return decorator

    def on(self, event: T, callback: Optional[Callable] = None) -> Callable:
        """
        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!
            ```
        """
        if callback is not None:
            if event not in self._events:
                self._events[event] = set()
            self._events[event].add(callback)
            return callback
        else:

            def decorator(callback: Callable) -> Callable:
                self.on(event, callback)
                return callback

            return decorator

    def off(self, event: T, callback: Callable) -> None:
        """
        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
            ```
        """
        if event in self._events:
            self._events[event].remove(callback)

Ancestors

typing.Generic

Subclasses

Methods

def emit(self, event: ~T, *args) ‑> None

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:

emitter = EventEmitter[str]()

def greet(name):
    print(f"Hello, {name}!")

emitter.on('greet', greet)
emitter.emit('greet', 'Alice')  # Output: Hello, Alice!

def off(self, event: ~T, callback: Callable) ‑> None

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:

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

def on(self, event: ~T, callback: Optional[Callable] = None) ‑> Callable

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:

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:

emitter = EventEmitter[str]()

@emitter.on('greet')
def greet(name):
    print(f"Hello, {name}!")

emitter.emit('greet', 'Charlie')  # Output: Hello, Charlie!

def once(self, event: ~T, callback: Optional[Callable] = None) ‑> Callable

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:

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:

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