platypush/platypush/__init__.py

159 lines
4.6 KiB
Python
Raw Normal View History

2018-07-30 22:08:06 +02:00
"""
Platypush
.. moduleauthor:: Fabio Manganiello <blacklight86@gmail.com>
2018-07-30 23:18:01 +02:00
.. license: MIT
2018-07-30 22:08:06 +02:00
"""
import argparse
2017-10-29 05:17:42 +01:00
import logging
import sys
import traceback
2017-10-29 05:17:42 +01:00
from threading import Thread
2018-09-20 12:49:57 +02:00
from .bus.redis import RedisBus
2017-12-18 01:10:51 +01:00
from .config import Config
from .context import register_backends
2018-01-15 22:36:24 +01:00
from .cron.scheduler import CronScheduler
from .event.processor import EventProcessor
from .message.event import Event, StopEvent
from .message.event.application import ApplicationStartedEvent, ApplicationStoppedEvent
from .message.request import Request
from .message.response import Response
2017-10-29 05:17:42 +01:00
__author__ = 'Fabio Manganiello <blacklight86@gmail.com>'
2018-12-01 19:44:41 +01:00
__version__ = '0.9.2'
2017-11-03 01:56:27 +01:00
2018-07-30 22:08:06 +02:00
LOGGER = logging.getLogger(__name__)
2018-06-06 20:09:18 +02:00
2018-07-30 22:08:06 +02:00
class Daemon:
""" Main class for the Platypush daemon """
2018-07-30 22:08:06 +02:00
# Configuration file (default: either ~/.config/platypush/config.yaml or
# /etc/platypush/config.yaml
config_file = None
2018-07-30 22:08:06 +02:00
# Application bus. It's an internal queue where:
# - backends will post the messages they receive
# - plugins will post the responses they process
bus = None
2018-07-30 22:08:06 +02:00
# backend_name => backend_obj map
backends = None
2018-07-30 22:08:06 +02:00
# number of executions retries before a request fails
n_tries = 2
2017-12-22 02:11:56 +01:00
def __init__(self, config_file=None, requests_to_process=None):
2018-07-30 22:08:06 +02:00
"""
Constructor
Params:
config_file -- Configuration file override (default: None)
2017-12-22 02:11:56 +01:00
requests_to_process -- Exit after processing the specified number
of requests (default: None, loop forever)
"""
self.config_file = config_file
self.event_processor = EventProcessor()
2017-12-22 02:11:56 +01:00
self.requests_to_process = requests_to_process
self.processed_requests = 0
Config.init(self.config_file)
2018-04-14 00:59:56 +02:00
logging.basicConfig(**Config.get('logging'))
@classmethod
def build_from_cmdline(cls, args):
2018-07-30 22:08:06 +02:00
"""
Build the app from command line arguments.
Params:
args -- Your sys.argv[1:] [List of strings]
"""
parser = argparse.ArgumentParser()
parser.add_argument('--config', '-c', dest='config', required=False,
default=None, help=cls.config_file.__doc__)
opts, args = parser.parse_known_args(args)
return cls(config_file=opts.config)
def on_message(self):
2018-07-30 22:08:06 +02:00
"""
Default message handler
"""
def _f(msg):
2018-07-30 22:08:06 +02:00
"""
on_message closure
Params:
2018-07-30 22:08:06 +02:00
msg -- platypush.message.Message instance
"""
if isinstance(msg, Request):
try:
msg.execute(n_tries=self.n_tries)
except PermissionError:
2018-07-30 22:08:06 +02:00
LOGGER.info('Dropped unauthorized request: {}'.format(msg))
self.processed_requests += 1
if self.requests_to_process \
and self.processed_requests >= self.requests_to_process:
self.stop_app()
elif isinstance(msg, Response):
2018-07-30 22:08:06 +02:00
LOGGER.info('Received response: {}'.format(msg))
elif isinstance(msg, StopEvent) and msg.targets_me():
2018-07-30 22:08:06 +02:00
LOGGER.info('Received STOP event: {}'.format(msg))
self.stop_app()
elif isinstance(msg, Event):
2018-07-30 22:08:06 +02:00
LOGGER.info('Received event: {}'.format(msg))
self.event_processor.process_event(msg)
return _f
2017-12-22 02:11:56 +01:00
def stop_app(self):
""" Stops the backends and the bus """
2017-12-22 02:11:56 +01:00
for backend in self.backends.values():
backend.stop()
self.bus.stop()
def start(self):
""" Start the daemon """
2018-09-20 12:49:57 +02:00
self.bus = RedisBus(on_message=self.on_message())
# Initialize the backends and link them to the bus
2017-12-24 13:15:37 +01:00
self.backends = register_backends(bus=self.bus, global_scope=True)
# Start the backend threads
for backend in self.backends.values():
backend.start()
2018-01-15 22:36:24 +01:00
# Start the cron scheduler
if Config.get_cronjobs():
CronScheduler(jobs=Config.get_cronjobs()).start()
2018-01-15 22:36:24 +01:00
self.bus.post(ApplicationStartedEvent())
# Poll for messages on the bus
try:
self.bus.poll()
2018-07-30 22:08:06 +02:00
except KeyboardInterrupt:
LOGGER.info('SIGINT received, terminating application')
2017-12-22 02:11:56 +01:00
finally:
self.bus.post(ApplicationStoppedEvent())
2017-12-22 02:11:56 +01:00
self.stop_app()
2018-01-04 16:11:54 +01:00
def main():
2018-07-30 22:08:06 +02:00
"""
Platypush daemon main
"""
2018-01-04 16:11:54 +01:00
print('Starting platypush v.{}'.format(__version__))
app = Daemon.build_from_cmdline(sys.argv[1:])
app.start()
# vim:sw=4:ts=4:et: