import logging import re import socket import time from threading import Thread, Event as ThreadEvent, get_ident from typing import Optional, Dict from platypush import __version__ from platypush.bus import Bus from platypush.common import ExtensionWithManifest from platypush.config import Config from platypush.context import get_backend from platypush.event import EventGenerator from platypush.message import Message from platypush.message.event import Event from platypush.message.event.zeroconf import ( ZeroconfServiceAddedEvent, ZeroconfServiceRemovedEvent, ) from platypush.message.request import Request from platypush.message.response import Response from platypush.utils import ( clear_timeout, get_backend_name_by_class, get_redis, get_redis_queue_name_by_message, get_remaining_timeout, set_timeout, ) class Backend(Thread, EventGenerator, ExtensionWithManifest): """ Parent class for backends. A backend is basically a thread that checks for new events on some channel (e.g. a network socket, a queue, some new entries on an API endpoint or an RSS feed, a voice command through an assistant, a new measure from a sensor etc.) and propagates event messages to the main application bus whenever a new event happens. You can then build whichever type of custom logic you want on such events. """ _default_response_timeout = 5 # Loop function, can be implemented by derived classes loop = None def __init__( self, bus: Optional[Bus] = None, poll_seconds: Optional[float] = None, **kwargs ): """ :param bus: Reference to the bus object to be used in the backend :param poll_seconds: If the backend implements a ``loop`` method, this parameter expresses how often the loop should run in seconds. :param kwargs: Key-value configuration for the backend """ self._thread_name = self.__class__.__name__ EventGenerator.__init__(self) ExtensionWithManifest.__init__(self) Thread.__init__(self, name=self._thread_name, daemon=True) # If no bus is specified, create an internal queue where # the received messages will be pushed self.bus = bus or Bus() self.poll_seconds = float(poll_seconds) if poll_seconds else None self.device_id = Config.get('device_id') self.thread_id = None self._stop_event = ThreadEvent() self._stop_thread: Optional[Thread] = None self._kwargs = kwargs self.logger = logging.getLogger( 'platypush:backend:' + get_backend_name_by_class(self.__class__) ) self.zeroconf = None self.zeroconf_info = None # Internal-only, we set the request context on a backend if that # backend is intended to react for a response to a specific request self._request_context = kwargs['_req_ctx'] if '_req_ctx' in kwargs else None if 'logging' in kwargs: self.logger.setLevel(getattr(logging, kwargs['logging'].upper())) def on_message(self, msg): """ Callback when a message is received on the backend. It parses and posts the message on the main bus. It should be called by the derived classes whenever a new message should be processed. :param msg: Received message. It can be either a key-value dictionary, a platypush.message.Message object, or a string/byte UTF-8 encoded string """ msg = Message.build(msg) if msg is None: return if not getattr(msg, 'target', None) or msg.target != self.device_id: return # Not for me self.logger.debug( 'Message received on the %s backend: %s', self.__class__.__name__, msg ) if self._is_expected_response(msg): # Expected response, trigger the response handler clear_timeout() if self._request_context: # pylint: disable=unsubscriptable-object self._request_context['on_response'](msg) self.stop() return msg.backend = self # Augment message to be able to process responses self.bus.post(msg) def _is_expected_response(self, msg): """ Internal only - returns true if we are expecting for a response and msg is that response. """ # pylint: disable=unsubscriptable-object return ( self._request_context and isinstance(msg, Response) and msg.id == self._request_context['request'].id ) def _get_backend_config(self): config_name = ( 'backend.' + self.__class__.__name__.split('Backend', maxsplit=1)[0].lower() ) return Config.get(config_name) or {} def _setup_response_handler(self, request, on_response, response_timeout): def _timeout_hndl(): raise RuntimeError( f'Timed out while waiting for a response from {request.target}' ) req_ctx = { 'request': request, 'on_response': on_response, 'response_timeout': response_timeout, } resp_backend = self.__class__( bus=self.bus, _req_ctx=req_ctx, **self._get_backend_config(), **self._kwargs ) # Set the response timeout if response_timeout: set_timeout(seconds=response_timeout, on_timeout=_timeout_hndl) resp_backend.start() def send_event(self, event, **kwargs): """ Send an event message on the backend. :param event: Event to send. It can be a dict, a string/bytes UTF-8 JSON, or a platypush.message.event.Event object. """ event = Event.build(event) assert isinstance(event, Event) event.origin = self.device_id if not hasattr(event, 'target'): event.target = self.device_id self.send_message(event, **kwargs) def send_request( self, request, on_response=None, response_timeout=_default_response_timeout, **kwargs, ): """ Send a request message on the backend. :param request: The request, either a dict, a string/bytes UTF-8 JSON, or a platypush.message.request.Request object. :param on_response: Optional callback that will be called when a response is received. If set, this method will synchronously wait for a response before exiting. :type on_response: function :param response_timeout: If on_response is set, the backend will raise an exception if the response isn't received within this number of seconds (default: None) :type response_timeout: float """ request = Request.build(request) assert isinstance(request, Request) request.origin = self.device_id if on_response and response_timeout != 0: self._setup_response_handler(request, on_response, response_timeout) self.send_message(request, **kwargs) def send_response(self, response, request, **kwargs): """ Send a response message on the backend. :param response: The response, either a dict, a string/bytes UTF-8 JSON, or a :class:`platypush.message.response.Response` object. :param request: Associated request, used to set the response parameters that will link them """ assert isinstance(response, Response) assert isinstance(request, Request) self.send_message(response, **kwargs) def send_message(self, msg, queue_name=None, **_): """ Sends a platypush.message.Message to a node. To be implemented in the derived classes. By default, if the Redis backend is configured then it will try to deliver the message to other consumers through the configured Redis main queue. :param msg: The message to send :param queue_name: Send the message on a specific queue (default: the queue_name configured on the Redis backend) """ try: redis = get_backend('redis') if not redis: raise KeyError() except KeyError: self.logger.warning( ( "Backend %s does not implement send_message " "and the fallback Redis backend isn't configured" ), self.__class__.__name__, ) return redis.send_message(msg, queue_name=queue_name) def run(self): """Starts the backend thread. To be implemented in the derived classes if the loop method isn't defined.""" self.thread_id = get_ident() if not callable(self.loop): return while not self.should_stop(): try: with self: has_error = False while not self.should_stop() and not has_error: try: # pylint: disable=not-callable self.loop() except Exception as e: has_error = True self.logger.error(str(e)) self.logger.exception(e) finally: if self.poll_seconds: time.sleep(self.poll_seconds) elif has_error: time.sleep(5) except Exception as e: self.logger.error( '%s initialization error: %s', self.__class__.__name__, e ) self.logger.exception(e) time.sleep(self.poll_seconds or 5) def __enter__(self): """Invoked when the backend is initialized, if the main logic is within a ``loop()`` function""" self.logger.info('Initialized backend %s', self.__class__.__name__) def __exit__(self, *_, **__): """Invoked when the backend is terminated, if the main logic is within a ``loop()`` function""" self.on_stop() self.logger.info('Terminated backend %s', self.__class__.__name__) def on_stop(self): """Callback invoked when the process stops""" def stop(self): """Stops the backend thread by sending a STOP event on its bus""" def _async_stop(): self._stop_event.set() self.unregister_service() self.on_stop() self._stop_thread = None if not (self._stop_thread and self._stop_thread.is_alive()): self._stop_thread = Thread(target=_async_stop) self._stop_thread.start() def should_stop(self): """ :return: True if the backend thread should be stopped, False otherwise. """ return self._stop_event.is_set() def wait_stop(self, timeout=None) -> bool: """ Waits for the backend thread to stop. :param timeout: The maximum time to wait for the backend thread to stop (default: None) :return: True if the backend thread has stopped, False otherwise. """ start = time.time() if self._stop_thread: try: self._stop_thread.join( get_remaining_timeout(timeout=timeout, start=start) ) except AttributeError: pass return self._stop_event.wait( get_remaining_timeout(timeout=timeout, start=start) ) def get_message_response(self, msg): queue = get_redis_queue_name_by_message(msg) if not queue: self.logger.warning('No response queue configured for the message') return None try: redis = get_redis() response = redis.blpop(queue, timeout=60) if response and len(response) > 1: response = Message.build(response[1]) else: response = None return response except Exception as e: self.logger.error('Error while processing response to %s: %s', msg, e) return None @staticmethod def _get_ip() -> str: """ Get the IP address of the machine. """ s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.connect(('8.8.8.8', 80)) addr = s.getsockname()[0] s.close() return addr def register_service( self, port: Optional[int] = None, name: Optional[str] = None, srv_type: Optional[str] = None, srv_name: Optional[str] = None, udp: bool = False, properties: Optional[Dict] = None, ): """ Initialize the Zeroconf service configuration for this backend. :param port: Service listen port (default: the backend ``port`` attribute if available, or ``None``). :param name: Service short name (default: backend name). :param srv_type: Service type (default: ``_platypush-{name}._{proto}.local.``). :param srv_name: Full service name (default: ``{hostname or device_id}.{type}``). :param udp: Set to True if this is a UDP service. :param properties: Extra properties to be passed on the service. Default: .. code-block:: json { "name": "Platypush", "vendor": "Platypush", "version": "{platypush_version}" } """ try: from zeroconf import ServiceInfo, Zeroconf from platypush.plugins.zeroconf import ZeroconfListener except ImportError: self.logger.warning( 'zeroconf package not available, service discovery will be disabled.' ) return if self.zeroconf: self.logger.info( 'Zeroconf service already registered for %s, removing the previous instance', self.__class__.__name__, ) self.unregister_service() self.zeroconf = Zeroconf() srv_desc = { 'name': 'Platypush', 'vendor': 'Platypush', 'version': __version__, **(properties or {}), } name = name or re.sub(r'Backend$', '', self.__class__.__name__).lower() srv_type = srv_type or f'_platypush-{name}._{"udp" if udp else "tcp"}.local.' srv_name = srv_name or f'{self.device_id}.{srv_type}' srv_port = port if port else getattr(self, 'port', None) self.zeroconf_info = ServiceInfo( srv_type, srv_name, addresses=[socket.inet_aton(self._get_ip())], port=srv_port, weight=0, priority=0, properties=srv_desc, ) if not self.zeroconf_info: self.logger.warning('Could not register Zeroconf service') return self.zeroconf.register_service(self.zeroconf_info) self.bus.post( ZeroconfServiceAddedEvent( service_type=srv_type, service_name=srv_name, service_info=ZeroconfListener.parse_service_info(self.zeroconf_info), ) ) def unregister_service(self): """ Unregister the Zeroconf service configuration if available. """ from redis import exceptions if self.zeroconf and self.zeroconf_info: try: self.zeroconf.unregister_service(self.zeroconf_info) except Exception as e: self.logger.warning( 'Could not register Zeroconf service %s: %s: %s', self.zeroconf_info.name, type(e).__name__, e, ) if self.zeroconf: try: self.zeroconf.close() except TimeoutError: pass try: if self.zeroconf_info: self.bus.post( ZeroconfServiceRemovedEvent( service_type=self.zeroconf_info.type, service_name=self.zeroconf_info.name, ) ) else: self.bus.post( ZeroconfServiceRemovedEvent( service_type=None, service_name=None ) ) except exceptions.ConnectionError: pass self.zeroconf_info = None self.zeroconf = None # vim:sw=4:ts=4:et: