revil-O/platypush
1
0
Fork 0
forked from platypush/platypush
Code Pull requests Activity

More consistent flow for messages received by a backend

Browse source
This commit is contained in:
Fabio Manganiello 2018-09-24 23:13:44 +02:00
parent 59d84c4fcb
commit d215410a6a
3 changed files with 8 additions and 15 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
platypush/__init__.py
View file

@ -114,7 +114,6 @@ class Daemon:
def stop_app(self): def stop_app(self):
""" Stops the backends and the bus """ """ Stops the backends and the bus """
self.bus.post(ApplicationStoppedEvent())
for backend in self.backends.values(): for backend in self.backends.values():
backend.stop() backend.stop()
self.bus.stop() self.bus.stop()
@ -143,6 +142,7 @@ class Daemon:
except KeyboardInterrupt: except KeyboardInterrupt:
LOGGER.info('SIGINT received, terminating application') LOGGER.info('SIGINT received, terminating application')
finally: finally:
self.bus.post(ApplicationStoppedEvent())
self.stop_app() self.stop_app()

17
platypush/backend/http/__init__.py
View file

@ -169,19 +169,14 @@ class HttpBackend(Backend):
def redis_poll(self): def redis_poll(self):
""" Polls for new messages on the internal Redis queue """ """ Polls for new messages on the internal Redis queue """
while True: while not self.should_stop():
redis = self._get_redis() redis = self._get_redis()
if not redis: if not redis:
continue continue
msg = redis.blpop(self.redis_queue) msg = redis.blpop(self.redis_queue)
msg = Message.build(json.loads(msg[1].decode('utf-8'))) msg = Message.build(json.loads(msg[1].decode('utf-8')))
self.on_message(msg)
if isinstance(msg, StopEvent) and \
msg.args.get('thread_id') == get_ident():
break
self.bus.post(msg)
@classmethod @classmethod
@ -243,7 +238,7 @@ class HttpBackend(Backend):
msg.origin = 'http' msg.origin = 'http'
redis = self._get_redis() redis = self._get_redis()
self.bus.post(msg) self.on_message(msg)
if isinstance(msg, Request): if isinstance(msg, Request):
response = redis.blpop(get_redis_queue_name_by_message(msg), timeout=60) response = redis.blpop(get_redis_queue_name_by_message(msg), timeout=60)
@ -402,16 +397,14 @@ class HttpBackend(Backend):
def run(self): def run(self):
super().run() super().run()
os.putenv('FLASK_APP', 'platypush') os.putenv('FLASK_APP', 'platypush')
os.putenv('FLASK_ENV', 'development') os.putenv('FLASK_ENV', 'production')
self.logger.info('Initialized HTTP backend on port {}'.format(self.port)) self.logger.info('Initialized HTTP backend on port {}'.format(self.port))
webserver = self.webserver() webserver = self.webserver()
self.server_proc = Process(target=webserver.run, kwargs={ self.server_proc = Process(target=webserver.run, kwargs={
'debug':True, 'host':'0.0.0.0', 'port':self.port, 'use_reloader':False 'host':'0.0.0.0', 'port':self.port, 'use_reloader':False
}) })
time.sleep(1)
self.server_proc.start() self.server_proc.start()
if not self.disable_websocket: if not self.disable_websocket:

4
tests/test_http.py
View file

@ -28,12 +28,12 @@ class TestHttp(unittest.TestCase):
def test_request_exec_flow(self): def test_request_exec_flow(self):
self.start_daemon() self.start_daemon()
time.sleep(2) time.sleep(1)
self.send_request() self.send_request()
def start_daemon(self): def start_daemon(self):
def _f(): def _f():
self.receiver = Daemon(config_file=config_file, requests_to_process=1) self.receiver = Daemon(config_file=config_file)
self.receiver.start() self.receiver.start()
Thread(target=_f).start() Thread(target=_f).start()