platypush/platypush/__init__.py

212 lines
7.5 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
2018-12-19 21:15:06 +01:00
import os
import sys
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 .logger import Logger
from .message.event import Event
from .message.event.application import ApplicationStartedEvent
from .message.request import Request
from .message.response import Response
from .utils import set_thread_name
2017-10-29 05:17:42 +01:00
2020-09-27 01:33:38 +02:00
__author__ = 'Fabio Manganiello <info@fabiomanganiello.com>'
2021-06-22 23:40:42 +02:00
__version__ = '0.21.1'
2017-11-03 01:56:27 +01:00
2020-09-27 01:33:38 +02:00
logger = logging.getLogger('platypush')
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
# Default bus queue name
_default_redis_queue = 'platypush/bus'
2018-12-19 21:15:06 +01:00
pidfile = 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
def __init__(self, config_file=None, pidfile=None, requests_to_process=None,
no_capture_stdout=False, no_capture_stderr=False, redis_queue=None):
2018-07-30 22:08:06 +02:00
"""
Constructor
Params:
config_file -- Configuration file override (default: None)
2018-12-19 21:15:06 +01:00
pidfile -- File where platypush will store its PID upon launch,
useful if you're planning to integrate the application
within a service or a launcher script (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)
no_capture_stdout -- Set to true if you want to disable the stdout
capture by the logging system
no_capture_stderr -- Set to true if you want to disable the stderr
capture by the logging system
redis_queue -- Name of the (Redis) queue used for dispatching messages (default: platypush/bus).
"""
2018-12-19 21:15:06 +01:00
if pidfile:
self.pidfile = pidfile
with open(self.pidfile, 'w') as f:
2018-12-19 21:24:39 +01:00
f.write(str(os.getpid()))
2018-12-19 21:15:06 +01:00
self.redis_queue = redis_queue or self._default_redis_queue
self.config_file = config_file
Config.init(self.config_file)
logging.basicConfig(**Config.get('logging'))
redis_conf = Config.get('backend.redis') or {}
self.bus = RedisBus(redis_queue=self.redis_queue, on_message=self.on_message(),
**redis_conf.get('redis_args', {}))
self.no_capture_stdout = no_capture_stdout
self.no_capture_stderr = no_capture_stderr
self.event_processor = EventProcessor()
2017-12-22 02:11:56 +01:00
self.requests_to_process = requests_to_process
self.processed_requests = 0
self.cron_scheduler = None
2017-12-22 02:11:56 +01:00
@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__)
2018-12-19 21:15:06 +01:00
parser.add_argument('--pidfile', '-P', dest='pidfile', required=False,
default=None, help="File where platypush will " +
"store its PID, useful if you're planning to " +
"integrate it in a service")
parser.add_argument('--no-capture-stdout', dest='no_capture_stdout',
required=False, action='store_true',
help="Set this flag if you have max stack depth " +
"exceeded errors so stdout won't be captured by " +
"the logging system")
parser.add_argument('--no-capture-stderr', dest='no_capture_stderr',
required=False, action='store_true',
help="Set this flag if you have max stack depth " +
"exceeded errors so stderr won't be captured by " +
"the logging system")
parser.add_argument('--redis-queue', dest='redis_queue',
required=False, action='store_true',
default=cls._default_redis_queue,
help="Name of the Redis queue to be used to internally deliver messages "
"(default: platypush/bus)")
opts, args = parser.parse_known_args(args)
return cls(config_file=opts.config, pidfile=opts.pidfile,
no_capture_stdout=opts.no_capture_stdout,
no_capture_stderr=opts.no_capture_stderr,
redis_queue=opts.redis_queue)
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:
2020-09-27 01:33:38 +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):
2020-09-27 01:33:38 +02:00
logger.info('Received response: {}'.format(msg))
elif isinstance(msg, Event):
if not msg.disable_logging:
2020-09-27 01:33:38 +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()
2017-12-22 02:11:56 +01:00
self.bus.stop()
if self.cron_scheduler:
self.cron_scheduler.stop()
2017-12-22 02:11:56 +01:00
def run(self):
""" Start the daemon """
if not self.no_capture_stdout:
2020-09-27 01:33:38 +02:00
sys.stdout = Logger(logger.info)
if not self.no_capture_stderr:
2020-09-27 01:33:38 +02:00
sys.stderr = Logger(logger.warning)
set_thread_name('platypush')
logger.info('---- Starting platypush v.{}'.format(__version__))
# 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():
self.cron_scheduler = CronScheduler(jobs=Config.get_cronjobs())
self.cron_scheduler.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:
2020-09-27 01:33:38 +02:00
logger.info('SIGINT received, terminating application')
2017-12-22 02:11:56 +01:00
finally:
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
app = Daemon.build_from_cmdline(sys.argv[1:])
app.run()
2018-01-04 16:11:54 +01:00
# vim:sw=4:ts=4:et: