FM / burster

/opt/cloudlinux/venv/lib/python3.11/site-packages/lvestats/plugins/generic/burster/

/opt/cloudlinux/venv/lib/python3.11/site-packages/lvestats/plugins/generic/burster

Name	Size	Mode	Actions
storage/	-	0755	rm
__pycache__/	-	0755	rm
adjust.py	7539	0644	edit dl rm
common.py	4668	0644	edit dl rm
config.py	8151	0644	edit dl rm
history.py	10030	0644	edit dl rm
lves_tracker.py	9139	0644	edit dl rm
lve_sm.py	18844	0644	edit dl rm
notify.py	1679	0644	edit dl rm
overload.py	3309	0644	edit dl rm
profiler.py	10546	0644	edit dl rm
utils.py	1665	0644	edit dl rm
_logs.py	597	0644	edit dl rm
__init__.py	8734	0644	edit dl rm

Edit: /opt/cloudlinux/venv/lib/python3.11/site-packages/lvestats/plugins/generic/burster/notify.py (1679B)

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)

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