342 lines
10 KiB
Python
342 lines
10 KiB
Python
import asyncio
|
|
import logging
|
|
import threading
|
|
import warnings
|
|
|
|
from abc import ABC, abstractmethod
|
|
from functools import wraps
|
|
from typing import Any, Callable, Optional
|
|
|
|
from platypush.bus import Bus
|
|
from platypush.common import ExtensionWithManifest
|
|
from platypush.event import EventGenerator
|
|
from platypush.message.response import Response
|
|
from platypush.utils import get_decorators, get_plugin_name_by_class
|
|
|
|
PLUGIN_STOP_TIMEOUT = 5 # Plugin stop timeout in seconds
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
|
|
def action(f: Callable[..., Any]) -> Callable[..., Response]:
|
|
"""
|
|
Decorator used to wrap the methods in the plugin classes that should be
|
|
exposed as actions.
|
|
|
|
It wraps the method's response into a generic
|
|
:meth:`platypush.message.response.Response` object.
|
|
"""
|
|
|
|
@wraps(f)
|
|
def _execute_action(*args, **kwargs) -> Response:
|
|
response = Response()
|
|
try:
|
|
result = f(*args, **kwargs)
|
|
except Exception as e:
|
|
if isinstance(e, KeyboardInterrupt):
|
|
return response
|
|
|
|
_logger.exception(e)
|
|
result = Response(errors=[str(e)])
|
|
|
|
if result and isinstance(result, Response):
|
|
result.errors = (
|
|
result.errors if isinstance(result.errors, list) else [result.errors]
|
|
)
|
|
response = result
|
|
elif isinstance(result, tuple) and len(result) == 2:
|
|
response.errors = result[1] if isinstance(result[1], list) else [result[1]]
|
|
|
|
if len(response.errors) == 1 and response.errors[0] is None:
|
|
response.errors = []
|
|
response.output = result[0]
|
|
else:
|
|
response = Response(output=result, errors=[])
|
|
|
|
return response
|
|
|
|
# Propagate the docstring
|
|
_execute_action.__doc__ = f.__doc__
|
|
# Expose the wrapped function
|
|
_execute_action.wrapped = f # type: ignore
|
|
return _execute_action
|
|
|
|
|
|
class Plugin(EventGenerator, ExtensionWithManifest): # lgtm [py/missing-call-to-init]
|
|
"""Base plugin class"""
|
|
|
|
def __init__(self, **kwargs):
|
|
super().__init__()
|
|
|
|
self.logger = logging.getLogger(
|
|
'platypush:plugin:' + get_plugin_name_by_class(self.__class__)
|
|
)
|
|
if 'logging' in kwargs:
|
|
self.logger.setLevel(getattr(logging, kwargs['logging'].upper()))
|
|
|
|
self.registered_actions = set(
|
|
get_decorators(self.__class__, climb_class_hierarchy=True).get('action', [])
|
|
)
|
|
|
|
@property
|
|
def _db(self):
|
|
"""
|
|
:return: The reference to the :class:`platypush.plugins.db.DbPlugin`.
|
|
"""
|
|
from platypush.context import get_plugin
|
|
from platypush.plugins.db import DbPlugin
|
|
|
|
db: DbPlugin = get_plugin(DbPlugin) # type: ignore
|
|
assert db, 'db plugin not initialized'
|
|
return db
|
|
|
|
@property
|
|
def _redis(self):
|
|
"""
|
|
:return: The reference to the :class:`platypush.plugins.redis.RedisPlugin`.
|
|
"""
|
|
from platypush.context import get_plugin
|
|
from platypush.plugins.redis import RedisPlugin
|
|
|
|
redis: RedisPlugin = get_plugin(RedisPlugin) # type: ignore
|
|
assert redis, 'db plugin not initialized'
|
|
return redis
|
|
|
|
@property
|
|
def _bus(self):
|
|
"""
|
|
:return: The reference to the :class:`platypush.bus.Bus`.
|
|
"""
|
|
from platypush.context import get_bus
|
|
|
|
return get_bus()
|
|
|
|
@property
|
|
def _entities(self):
|
|
"""
|
|
:return: The reference to the :class:`platypush.plugins.entities.EntitiesPlugin`.
|
|
"""
|
|
from platypush.context import get_plugin
|
|
from platypush.plugins.entities import EntitiesPlugin
|
|
|
|
entities: EntitiesPlugin = get_plugin('entities') # type: ignore
|
|
assert entities, 'entities plugin not initialized'
|
|
return entities
|
|
|
|
def __str__(self):
|
|
"""
|
|
:return: The qualified name of the plugin.
|
|
"""
|
|
return get_plugin_name_by_class(self.__class__)
|
|
|
|
def run(self, method, *args, **kwargs):
|
|
assert (
|
|
method in self.registered_actions
|
|
), f'{method} is not a registered action on {self.__class__.__name__}'
|
|
return getattr(self, method)(*args, **kwargs)
|
|
|
|
|
|
class RunnablePlugin(Plugin):
|
|
"""
|
|
Class for runnable plugins - i.e. plugins that have a start/stop method and can be started.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
poll_interval: Optional[float] = 15,
|
|
stop_timeout: Optional[float] = PLUGIN_STOP_TIMEOUT,
|
|
disable_monitor: bool = False,
|
|
**kwargs,
|
|
):
|
|
"""
|
|
:param poll_interval: How often the :meth:`.loop` function should be
|
|
executed (default: 15 seconds). *NOTE*: For back-compatibility
|
|
reasons, the `poll_seconds` argument is also supported, but it's
|
|
deprecated.
|
|
:param stop_timeout: How long we should wait for any running
|
|
threads/processes to stop before exiting (default: 5 seconds).
|
|
:param disable_monitor: If set to True then the plugin will not monitor
|
|
for new events. This is useful if you want to run a plugin in
|
|
stateless mode and only leverage its actions, without triggering any
|
|
events. Defaults to False.
|
|
"""
|
|
super().__init__(**kwargs)
|
|
self.poll_interval = poll_interval
|
|
self.bus: Optional[Bus] = None
|
|
self.disable_monitor = disable_monitor
|
|
self._should_stop = threading.Event()
|
|
self._stop_timeout = stop_timeout
|
|
self._thread: Optional[threading.Thread] = None
|
|
|
|
if kwargs.get('poll_seconds') is not None:
|
|
warnings.warn(
|
|
'poll_seconds is deprecated, use poll_interval instead',
|
|
DeprecationWarning,
|
|
stacklevel=2,
|
|
)
|
|
|
|
if self.poll_interval is None:
|
|
self.poll_interval = kwargs['poll_seconds']
|
|
|
|
def main(self):
|
|
"""
|
|
Implementation of the main loop of the plugin.
|
|
"""
|
|
raise NotImplementedError()
|
|
|
|
def should_stop(self) -> bool:
|
|
return self._should_stop.is_set()
|
|
|
|
def wait_stop(self, timeout=None):
|
|
"""
|
|
Wait until a stop event is received.
|
|
"""
|
|
if self.disable_monitor:
|
|
# Wait indefinitely if the monitor is disabled
|
|
return self._should_stop.wait(timeout=None)
|
|
|
|
return self._should_stop.wait(timeout=timeout)
|
|
|
|
def start(self):
|
|
"""
|
|
Start the plugin.
|
|
"""
|
|
self._thread = threading.Thread(
|
|
target=self._runner, name=self.__class__.__name__
|
|
)
|
|
self._thread.start()
|
|
|
|
def stop(self):
|
|
"""
|
|
Stop the plugin.
|
|
"""
|
|
self._should_stop.set()
|
|
if (
|
|
self._thread
|
|
and self._thread != threading.current_thread()
|
|
and self._thread.is_alive()
|
|
):
|
|
self.logger.info('Waiting for the plugin to stop')
|
|
try:
|
|
if self._thread:
|
|
self._thread.join(timeout=self._stop_timeout)
|
|
if self._thread and self._thread.is_alive():
|
|
self.logger.warning(
|
|
'Timeout (seconds=%s) on exit for the plugin',
|
|
self._stop_timeout,
|
|
)
|
|
except Exception as e:
|
|
self.logger.warning('Could not join thread on stop: %s', e)
|
|
|
|
self.logger.info(
|
|
'Stopped plugin: [%s]', get_plugin_name_by_class(self.__class__)
|
|
)
|
|
|
|
def _runner(self):
|
|
"""
|
|
Implementation of the runner thread.
|
|
"""
|
|
if self.disable_monitor:
|
|
return
|
|
|
|
self.logger.info(
|
|
'Starting plugin: [%s]', get_plugin_name_by_class(self.__class__)
|
|
)
|
|
|
|
while not self.should_stop():
|
|
try:
|
|
self.main()
|
|
except Exception as e:
|
|
self.logger.exception(e)
|
|
|
|
if self.poll_interval:
|
|
self.wait_stop(self.poll_interval)
|
|
|
|
self._thread = None
|
|
|
|
|
|
class AsyncRunnablePlugin(RunnablePlugin, ABC):
|
|
"""
|
|
Class for runnable plugins with an asynchronous event loop attached.
|
|
"""
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
|
|
self._loop: Optional[asyncio.AbstractEventLoop] = asyncio.new_event_loop()
|
|
self._task: Optional[asyncio.Task] = None
|
|
|
|
@property
|
|
def _should_start_runner(self):
|
|
"""
|
|
This property is used to determine if the runner and the event loop
|
|
should be started for this plugin.
|
|
"""
|
|
return True
|
|
|
|
@abstractmethod
|
|
async def listen(self):
|
|
"""
|
|
Main body of the async plugin. When it's called, the event loop should
|
|
already be running and available over `self._loop`.
|
|
"""
|
|
|
|
async def _listen(self):
|
|
"""
|
|
Wrapper for :meth:`.listen` that catches any exceptions and logs them.
|
|
"""
|
|
try:
|
|
await self.listen()
|
|
except KeyboardInterrupt:
|
|
pass
|
|
except RuntimeError as e:
|
|
if not (
|
|
str(e).startswith('Event loop stopped before ')
|
|
or str(e).startswith('no running event loop')
|
|
):
|
|
raise e
|
|
|
|
def _run_listener(self):
|
|
"""
|
|
Initialize an event loop and run the listener as a task.
|
|
"""
|
|
assert self._loop, 'The loop is not initialized'
|
|
asyncio.set_event_loop(self._loop)
|
|
|
|
self._task = self._loop.create_task(self._listen())
|
|
if hasattr(self._task, 'set_name'):
|
|
self._task.set_name(self.__class__.__name__ + '.listen')
|
|
try:
|
|
self._loop.run_until_complete(self._task)
|
|
except Exception as e:
|
|
if not self.should_stop():
|
|
self.logger.warning('The loop has terminated with an error')
|
|
self.logger.exception(e)
|
|
|
|
self._task.cancel()
|
|
|
|
def main(self):
|
|
if self.should_stop():
|
|
self.logger.info('The plugin is already scheduled to stop')
|
|
return
|
|
|
|
self._loop = asyncio.new_event_loop()
|
|
|
|
if self._should_start_runner:
|
|
while not self.should_stop():
|
|
try:
|
|
self._run_listener()
|
|
finally:
|
|
self.wait_stop(self.poll_interval)
|
|
else:
|
|
self.wait_stop()
|
|
|
|
def stop(self):
|
|
if self._loop and self._loop.is_running():
|
|
self._loop.call_soon_threadsafe(self._loop.stop)
|
|
self._loop = None
|
|
|
|
super().stop()
|
|
|
|
|
|
# vim:sw=4:ts=4:et:
|