"""
The events manager contains the class that manages custom and abstract callbacks into the system callbacks.
Once a signals is registered here it could be used by string reference. This makes it easy to have dynamically signals
being created by other apps in a single place so it could be used over all apps.
For example you would create your own custom signal if you have a app for your own created game mode script that abstracts
all the raw XML-RPC events into nice structured and maybe even including fetched data from external sources.
"""
import importlib
import logging
import sys
from pyplanet.core.events import Signal
[docs]class _SignalManager:
"""
Signal Manager class.
.. note::
Access this in the app via ``self.context.signals``.
"""
def __init__(self):
self.app_managers = dict()
self.signals = dict()
self.callbacks = dict()
# Reserved signal receivers, this will be filled, and copied to real signals later on.
self.reserved = dict()
self.reserved_self = dict()
#
self.namespaces = list()
# This var is used to temporary override namespaces when processing apps.
self._current_namespace = None
[docs] def register_signal(self, signal, app=None, callback=False):
"""
Register a signal to be known in the signalling system.
:param signal: Signal(s)
:param app: App context/instance.
:param callback: Will a callback handle the response (mostly raw callbacks).
"""
if isinstance(signal, list):
for sig in signal:
self.register_signal(sig)
return
if not signal.code:
raise Exception('Signal code is not valid!')
if not signal.namespace and self._current_namespace:
namespace = self._current_namespace
else:
namespace = signal.namespace
code = signal.code
if not hasattr(signal, 'receivers'):
instance = signal()
else:
instance = signal
signal_code = '{}:{}'.format(namespace, code)
if callback:
self.callbacks[code] = instance
else:
self.signals[signal_code] = instance
[docs] def listen(self, signal, target, conditions=None, **kwargs):
"""
Register a listing client to the signal given (signal instance or string).
:param signal: Signal instance or string: "namespace:code"
:param target: Target method to call.
:param conditions: Reserved for future purposes.
"""
try:
if not isinstance(signal, Signal):
signal = self.get_signal(signal)
signal.register(target, **kwargs)
except:
if signal not in self.reserved:
self.reserved[signal] = list()
self.reserved[signal].append((target, kwargs))
[docs] def get_callback(self, call_name):
"""
Get signal by XML-RPC (script) callback.
:param call_name: Callback name.
:return: Signal class or nothing.
:rtype: pyplanet.core.events.Signal
"""
if call_name in self.callbacks:
return self.callbacks[call_name]
logging.debug('No callback registered for {}'.format(call_name))
return None
[docs] def get_signal(self, key):
"""
Get signal by key (namespace:code).
:param key: namespace:code key.
:return: signal or none
:rtype: pyplanet.core.events.Signal
"""
if key in self.signals:
return self.signals[key]
else:
raise KeyError('No such signal {}!'.format(key))
[docs] def finish_reservations(self): # pragma: no cover
"""
The method will copy all reservations to the actual signals. (PRIVATE)
"""
for sig_name, recs in self.reserved.items():
for func, kwargs in recs:
try:
signal = self.get_signal(sig_name)
signal.connect(func, **kwargs)
except Exception as e:
logging.warning('Signal not found: {}, {}'.format(
sig_name, e
), exc_info=sys.exc_info())
for sig_name, recs in self.reserved_self.items():
for func, slf in recs:
try:
signal = self.get_signal(sig_name)
signal.set_self(func, slf)
except Exception as e:
logging.warning(str(e), exc_info=sys.exc_info())
self.reserved = dict()
self.reserved_self = dict()
[docs] def init_app(self, app):
"""
Initiate app, load all signal/callbacks files. (just import, they should register with decorators).
:param app: App instance
:type app: pyplanet.apps.AppConfig
"""
self._current_namespace = app.label
# Import the signals module.
try:
importlib.import_module('{}.signals'.format(app.name))
except ImportError:
pass
self._current_namespace = None
# Import the callbacks module.
try:
importlib.import_module('{}.callbacks'.format(app.name))
except ImportError:
pass
[docs] async def finish_start(self, *args, **kwargs):
"""
Finish startup the core, this will copy reservations. (PRIVATE).
"""
self.finish_reservations()
[docs] def create_app_manager(self, app):
"""
This method will create the manager instance for the app context.
:param app: App instance.
:type app: pyplanet.apps.config.AppConfig
:return: SignalManager instance for the app.
:rtype: pyplanet.core.events.manager.AppSignalManager
"""
return AppSignalManager(self, app)
class AppSignalManager:
def __init__(self, manager, app):
"""
Create the app manager proxy.
:param manager: Signal manager (core global).
:param app: App instance.
:type manager: pyplanet.core.events.manager._SignalManager
:type app: pyplanet.apps.config.AppConfig
"""
self.manager = manager
self.app = app
# Hold these to delete when the context is destroyed.
self.signals = list()
self.listeners = list()
def register_signal(self, signal, callback=False):
"""
Register a signal to be known in the signalling system.
:param signal: Signal(s)
:param callback: Will a callback handle the response (mostly raw callbacks).
"""
self.manager.register_signal(signal, self.app, callback=callback)
self.signals = list
def listen(self, signal, target, conditions=None, **kwargs):
"""
Register a listing client to the signal given (signal instance or string).
:param signal: Signal instance or string: "namespace:code"
:param target: Target method to call.
:param conditions: Reserved for future purposes.
"""
self.manager.listen(signal, target, conditions, **kwargs)
self.listeners.append((signal, target))
def get_callback(self, call_name):
"""
Get signal by XML-RPC (script) callback.
:param call_name: Callback name.
:return: Signal class or nothing.
:rtype: pyplanet.core.events.Signal
"""
return self.manager.get_callback(call_name)
def get_signal(self, key):
"""
Get signal by key (namespace:code).
:param key: namespace:code key.
:return: signal or none
:rtype: pyplanet.core.events.Signal
"""
return self.manager.get_signal(key)
async def on_destroy(self):
for signal, target in self.listeners:
try:
signal.unregister(target)
except Exception as e:
logging.exception(e)
# TODO: Delete signals.
SignalManager = _SignalManager()
def public_signal(cls):
SignalManager.register_signal(cls)
return cls
def public_callback(cls):
SignalManager.register_signal(cls)
return cls