import importlib import logging import sys import threading from threading import Thread from platypush.bus import Bus from platypush.config import Config from platypush.utils import get_message_class_by_type, set_timeout, clear_timeout from platypush.message import Message from platypush.message.event import Event, StopEvent from platypush.message.request import Request from platypush.message.response import Response class Backend(Thread): """ Parent class for backends """ _default_response_timeout = 5 def __init__(self, bus=None, **kwargs): """ Params: bus -- Reference to the Platypush bus where the requests and the responses will be posted [Bus] kwargs -- key-value configuration for this backend [Dict] """ # If no bus is specified, create an internal queue where # the received messages will be pushed self.bus = bus or Bus() self.device_id = Config.get('device_id') self.thread_id = None self._stop = False self._kwargs = kwargs # 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 Thread.__init__(self) logging.basicConfig(stream=sys.stdout, level=Config.get('logging') if 'logging' not in kwargs else getattr(logging, kwargs['logging'])) 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. Params: msg -- The 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 not getattr(msg, 'target') or msg.target != self.device_id: return # Not for me logging.debug('Message received on the {} backend: {}'.format( self.__class__.__name__, msg)) if self._is_expected_response(msg): # Expected response, trigger the response handler clear_timeout() self._request_context['on_response'](msg) self.stop() return if isinstance(msg, StopEvent) and msg.targets_me(): logging.info('Received STOP event on {}'.format(self.__class__.__name__)) self._stop = True else: 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 """ 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')[0].lower() return Config.get(config_name) def _setup_response_handler(self, request, on_response, response_timeout): def _timeout_hndl(): raise RuntimeError('Timed out while waiting for a response from {}'. format(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 Params: event -- The request, either 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 Params: request -- The request, either a dict, a string/bytes UTF-8 JSON, or a platypush.message.request.Request object. on_response -- Response handler, takes a platypush.message.response.Response as argument. If set, the method will wait for a response before exiting (default: None) 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: 5) """ 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 Params: response -- The response, either a dict, a string/bytes UTF-8 JSON, or a platypush.message.response.Response object request -- Associated request, used to set the response parameters that will link them """ response = Response.build(response) assert isinstance(response, Response) assert isinstance(request, Request) response.id = request.id response.target = request.origin response.origin = self.device_id self.send_message(response, **kwargs) def send_message(self, msg, **kwargs): """ Sends a platypush.message.Message to a node. To be implemented in the derived classes. Always call send_request or send_response instead of send_message directly Param: msg -- The message """ raise NotImplementedError("send_message should be implemented in a derived class") def run(self): """ Starts the backend thread. To be implemented in the derived classes """ self.thread_id = threading.get_ident() def on_stop(self): """ Callback invoked when the process stops """ pass def stop(self): """ Stops the backend thread by sending a STOP event on its bus """ def _async_stop(): evt = StopEvent(target=self.device_id, origin=self.device_id, thread_id=self.thread_id) self.send_message(evt) self.on_stop() Thread(target=_async_stop).start() def should_stop(self): return self._stop # vim:sw=4:ts=4:et: