Gecko [ renovalkoronred.com.tr ]

Name	Size	Permission
__pycache__	[ DIR ]	drwxr-xr-x
storage	[ DIR ]	drwxr-xr-x
__init__.py	8.49 KB	-rw-r--r--
_logs.py	597 B	-rw-r--r--
adjust.py	7.37 KB	-rw-r--r--
common.py	4.46 KB	-rw-r--r--
config.py	7.85 KB	-rw-r--r--
history.py	9.79 KB	-rw-r--r--
lve_sm.py	18.35 KB	-rw-r--r--
lves_tracker.py	8.91 KB	-rw-r--r--
notify.py	1.65 KB	-rw-r--r--
overload.py	3.24 KB	-rw-r--r--
profiler.py	10.29 KB	-rw-r--r--
utils.py	1.63 KB	-rw-r--r--

Code Editor : notify.py

# coding=utf-8
#
# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2023 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT
from typing import Callable, Protocol

class Signal(Protocol):
    def register(self, listener: Callable) -> Callable:
        ...

# def unregister(self, listener: Callable) -> Callable:
    #     ...

class Emitter(Signal):
    # NOTE(vlebedev): registering a listener will keep it alive (i.e. not being garbage collected)
    #                 until it's explicitly unregistered.

def __init__(self):
        self._listeners = set()

def register(self, listener: Callable):
        if not callable(listener):
            raise ValueError("Listener must be callable")
        self._listeners.add(listener)
        return listener

# TODO(vlebedev): Might be problems with references to e.g. class methods.
    #                 It's not that obvious how to handle them properly as they are transient/ephemeral.
    #                 For more info have a look at https://docs.python.org/3/library/weakref.html#weakref.WeakMethod
    # def unregister(self, listener: Callable):
    #     self._listeners.discard(listener)
    #     return listener

def __call__(self, *args, **kwargs) -> None:
        exceptions = []
        for listener in tuple(self._listeners):
            if listener is None:
                continue
            try:
                listener(*args, **kwargs)
            except Exception as e:
                exceptions.append(e)
        
        if exceptions:
            raise ExceptionGroup("Exceptions occurred in listeners", exceptions)

${this.title}

Code Editor : notify.py