pbf.controller.ListenerManager

  1command_listeners = {}
  2message_listeners = {}
  3notice_listeners = {}
  4request_listeners = {}
  5meta_listeners = {}
  6
  7
  8class ListenerManager:
  9    @staticmethod
 10    def set_listener(event_type: str, listener_dict: dict):
 11        """
 12        Set the listener dict.
 13        :param event_type: str event type
 14        :param listener_dict: dict listener dict
 15        :return:
 16        """
 17        global command_listeners, message_listeners, notice_listeners, request_listeners, meta_listeners
 18        if event_type == "command":
 19            command_listeners = listener_dict
 20        elif event_type == "message":
 21            message_listeners = listener_dict
 22        elif event_type == "notice":
 23            notice_listeners = listener_dict
 24        elif event_type == "request":
 25            request_listeners = listener_dict
 26        elif event_type == "meta":
 27            meta_listeners = listener_dict
 28        else:
 29            raise ValueError("Invalid event type")
 30
 31    @staticmethod
 32    def add_listener(event_type: str, plugin_name: str, listener):
 33        """
 34        Add a listener.
 35        :param event_type: str event type
 36        :param plugin_name: str plugin name
 37        :param listener: object listener
 38        :return:
 39        """
 40        listeners = ListenerManager.get_listeners_by_type(event_type)
 41        if plugin_name not in listeners:
 42            listeners[plugin_name] = []
 43        listeners[plugin_name].append(listener)
 44        ListenerManager.set_listener(event_type, listeners)
 45
 46    @staticmethod
 47    def clear():
 48        """
 49        Clear all listeners.
 50        :return: None
 51        """
 52        global command_listeners, message_listeners, notice_listeners, request_listeners, meta_listeners
 53        del command_listeners
 54        del message_listeners
 55        del notice_listeners
 56        del request_listeners
 57        del meta_listeners
 58        command_listeners = {}
 59        message_listeners = {}
 60        notice_listeners = {}
 61        request_listeners = {}
 62        meta_listeners = {}
 63
 64    @staticmethod
 65    def get_listeners_by_plugin_name(plugin_name: str):
 66        """
 67        Get listeners by plugin name.
 68        :param plugin_name: str plugin name
 69        :return: list
 70        """
 71        return command_listeners.get(plugin_name, []) + \
 72            message_listeners.get(plugin_name, []) + \
 73            notice_listeners.get(plugin_name, []) + \
 74            request_listeners.get(plugin_name, []) + \
 75            meta_listeners.get(plugin_name, [])
 76
 77    @staticmethod
 78    def get_listeners_by_type(event_type: str):
 79        """
 80        Get listeners by type.
 81        :param event_type: str event type
 82        :return: dict
 83        """
 84        if event_type == "command":
 85            return command_listeners
 86        elif event_type == "message":
 87            return message_listeners
 88        elif event_type == "notice":
 89            return notice_listeners
 90        elif event_type == "request":
 91            return request_listeners
 92        elif event_type == "meta":
 93            return meta_listeners
 94        else:
 95            return {}
 96
 97    @staticmethod
 98    def remove_listeners_by_plugin_name(plugin_name: str):
 99        """
100        Remove listeners by plugin name.
101        :param plugin_name: str plugin name
102        :return: None
103        """
104        # global command_listeners, message_listeners, notice_listeners, request_listeners, meta_listeners
105        for listeners in [command_listeners, message_listeners, notice_listeners, request_listeners, meta_listeners]:
106            if plugin_name in listeners:
107                listeners.pop(plugin_name)
108        ListenerManager.set_listener("command", command_listeners)
109        ListenerManager.set_listener("message", message_listeners)
110        ListenerManager.set_listener("notice", notice_listeners)
111        ListenerManager.set_listener("request", request_listeners)
112        ListenerManager.set_listener("meta", meta_listeners)
113
114    @staticmethod
115    def remove_listener_by_plugin_name(plugin_name: str, listener):
116        """
117        Remove listener by plugin name.
118        :param plugin_name: str plugin name
119        :param listener: object listener
120        :return: None
121        """
122        listeners = ListenerManager.get_listeners_by_plugin_name(plugin_name)
123        for i in range(len(listeners)):
124            if listeners[i] == listener:
125                listeners.pop(i)
126                break