Made an HTTP backend, #27

This commit is contained in:
Fabio Manganiello 2018-01-04 02:45:23 +01:00
parent 2743d46b1c
commit 20b07fb02f
7 changed files with 116 additions and 37 deletions

View file

@ -75,14 +75,6 @@ class Daemon(object):
msg -- platypush.message.Message instance """ msg -- platypush.message.Message instance """
if isinstance(msg, Request): if isinstance(msg, Request):
if msg.action.startswith('procedure.'):
logging.info('Executing procedure request: {}'.format(msg))
proc_name = msg.action.split('.')[-1]
proc_config = Config.get_procedures()[proc_name]
msg = Procedure.build(name=proc_name, requests=proc_config, backend=msg.backend, id=msg.id)
else:
logging.info('Processing request: {}'.format(msg))
msg.execute(n_tries=self.n_tries) msg.execute(n_tries=self.n_tries)
self.processed_requests += 1 self.processed_requests += 1
if self.requests_to_process \ if self.requests_to_process \

View file

@ -1,14 +1,10 @@
import sys import sys
from . import Daemon, __version__ from platypush import Daemon, __version__
def main(args=sys.argv[1:]): print('Starting platypush v.{}'.format(__version__))
print('Starting platypush v.{}'.format(__version__)) app = Daemon.build_from_cmdline(sys.argv[1:])
app = Daemon.build_from_cmdline(args) app.start()
app.start()
if __name__ == '__main__':
main()
# vim:sw=4:ts=4:et: # vim:sw=4:ts=4:et:

View file

@ -0,0 +1,66 @@
import logging
import json
from multiprocessing import Process
from flask import Flask, abort, jsonify, request
from platypush.message import Message
from platypush.message.request import Request
from .. import Backend
class HttpBackend(Backend):
""" Example interaction with the HTTP backend to make requests:
$ curl -XPOST \
-d 'msg={"type":"request","target":"volta","action":"tts.say","args": {"phrase":"This is a test"}}' \
http://localhost:8008 """
def __init__(self, port=8008, token=None, **kwargs):
super().__init__(**kwargs)
self.port = port
self.token = token
def send_message(self, msg):
raise NotImplementedError('Use cURL or any HTTP client to query the HTTP backend')
def _start_server(self):
def app_main():
app = Flask(__name__)
@app.route('/', methods=['POST'])
def index():
args = { k:v for (k,v) in request.form.items() }
if self.token:
if 'token' not in args or args['token'] != self.token:
abort(401)
if 'msg' not in args:
abort(400)
msg = Message.build(args['msg'])
logging.debug('Received message on HTTP backend: {}'.format(msg))
if isinstance(msg, Request):
response = msg.execute(async=False)
return str(response)
return jsonify({ 'status': 'ok' })
app.run(debug=True, host='0.0.0.0', port=self.port)
logging.info('Initialized HTTP backend on port {}'.format(self.port))
return app_main
def run(self):
super().run()
self.server_proc = Process(target=self._start_server())
self.server_proc.start()
self.server_proc.join()
# vim:sw=4:ts=4:et:

View file

@ -51,15 +51,33 @@ class Request(Message):
id += '%.2x' % random.randint(0, 255) id += '%.2x' % random.randint(0, 255)
return id return id
def execute(self, n_tries=1):
def _execute_procedure(self, *args, **kwargs):
from config import Config
logging.info('Executing procedure request: {}'.format(procedure))
proc_name = self.action.split('.')[-1]
proc_config = Config.get_procedures()[proc_name]
proc = Procedure.build(name=proc_name, requests=proc_config, backend=self.backend, id=self.id)
proc.execute(*args, **kwargs)
def execute(self, n_tries=1, async=True):
""" """
Execute this request and returns a Response object Execute this request and returns a Response object
Params: Params:
n_tries -- Number of tries in case of failure before raising a RuntimeError n_tries -- Number of tries in case of failure before raising a RuntimeError
async -- If True, the request will be run asynchronously and the
response posted on the bus when available (default),
otherwise the current thread will wait for the response
to be returned synchronously.
""" """
def _thread_func(n_tries):
(module_name, method_name) = get_module_and_method_from_action(self.action)
def _thread_func(n_tries):
if self.action.startswith('procedure.'):
return self._execute_procedure(n_tries=n_tries)
(module_name, method_name) = get_module_and_method_from_action(self.action)
plugin = get_plugin(module_name) plugin = get_plugin(module_name)
try: try:
@ -80,14 +98,20 @@ class Request(Message):
_thread_func(n_tries-1) _thread_func(n_tries-1)
return return
finally: finally:
# Send the response on the backend if async:
if self.backend and self.origin: # Send the response on the backend
self.backend.send_response(response=response, request=self) if self.backend and self.origin:
self.backend.send_response(response=response, request=self)
else:
logging.info('Response whose request has no ' +
'origin attached: {}'.format(response))
else: else:
logging.info('Response whose request has no ' + return response
'origin attached: {}'.format(response))
Thread(target=_thread_func, args=(n_tries,)).start() if async:
Thread(target=_thread_func, args=(n_tries,)).start()
else:
return _thread_func(n_tries)
def __str__(self): def __str__(self):

View file

@ -2,17 +2,14 @@ import sys
from . import Pusher from . import Pusher
def main(args=sys.argv[1:]): opts = Pusher.parse_build_args(sys.argv[1:])
opts = Pusher.parse_build_args(args) pusher = Pusher(config_file=opts.config, backend=opts.backend)
pusher = Pusher(config_file=opts.config, backend=opts.backend)
if opts.type == 'event': if opts.type == 'event':
pusher.send_event(target=opts.target, type=opts.event, **opts.args) pusher.send_event(target=opts.target, type=opts.event, **opts.args)
else: else:
pusher.send_request(target=opts.target, action=opts.action, timeout=opts.timeout, **opts.args) pusher.send_request(target=opts.target, action=opts.action, timeout=opts.timeout, **opts.args)
main()
# vim:sw=4:ts=4:et: # vim:sw=4:ts=4:et:

View file

@ -8,6 +8,9 @@ kafka-python
# PushBullet backend support # PushBullet backend support
websocket-client websocket-client
# HTTP backend support
flask
# Philips Hue plugin support # Philips Hue plugin support
phue phue

View file

@ -43,8 +43,8 @@ setup(
packages = find_packages(), packages = find_packages(),
entry_points = { entry_points = {
'console_scripts': [ 'console_scripts': [
'platypush=platypush.__main__:main', 'platypush=platypush.__main__',
'pusher=platypush.pusher.__main__:main', 'pusher=platypush.pusher.__main__',
], ],
}, },
data_files = [ data_files = [
@ -63,6 +63,7 @@ setup(
extras_require = { extras_require = {
'Support for Apache Kafka backend': ['kafka-python'], 'Support for Apache Kafka backend': ['kafka-python'],
'Support for Pushbullet backend': ['requests', 'websocket-client'], 'Support for Pushbullet backend': ['requests', 'websocket-client'],
'Support for HTTP backend': ['flask'],
'Support for Philips Hue plugin': ['phue'], 'Support for Philips Hue plugin': ['phue'],
'Support for MPD/Mopidy music server plugin': ['python-mpd2'], 'Support for MPD/Mopidy music server plugin': ['python-mpd2'],
'Support for Belkin WeMo Switch plugin': ['ouimeaux'], 'Support for Belkin WeMo Switch plugin': ['ouimeaux'],
@ -70,7 +71,7 @@ setup(
'Support for OMXPlayer plugin': ['omxplayer'], 'Support for OMXPlayer plugin': ['omxplayer'],
'Support for YouTube in the OMXPlayer plugin': ['youtube-dl'], 'Support for YouTube in the OMXPlayer plugin': ['youtube-dl'],
'Support for Google Assistant': ['google-assistant-sdk[samples]'], 'Support for Google Assistant': ['google-assistant-sdk[samples]'],
'Support for Flic buttons': ['-e git+https://github.com/50ButtonsEach/fliclib-linux-hci'] # 'Support for Flic buttons': ['-e git+https://github.com/50ButtonsEach/fliclib-linux-hci']
}, },
) )