platypush/platypush/bus/redis.py

69 lines
1.7 KiB
Python
Raw Normal View History

2018-09-20 09:41:19 +02:00
import logging
import threading
from typing import Final, Optional
2018-09-20 09:41:19 +02:00
from platypush.bus import Bus
from platypush.message import Message
2020-09-27 01:33:38 +02:00
logger = logging.getLogger('platypush:bus:redis')
2018-09-20 09:41:19 +02:00
class RedisBus(Bus):
2023-02-05 22:00:50 +01:00
"""
Overrides the in-process in-memory local bus with a Redis bus
"""
DEFAULT_REDIS_QUEUE: Final[str] = 'platypush/bus'
2018-09-20 09:41:19 +02:00
2022-01-23 14:25:00 +01:00
def __init__(self, *args, on_message=None, redis_queue=None, **kwargs):
2023-02-05 22:00:50 +01:00
from platypush.utils import get_redis
2023-02-05 22:00:50 +01:00
super().__init__(on_message=on_message)
self.redis = get_redis(*args, **kwargs)
self.redis_args = kwargs
self.redis_queue = redis_queue or self.DEFAULT_REDIS_QUEUE
2018-09-20 09:41:19 +02:00
self.on_message = on_message
self.thread_id = threading.get_ident()
2023-02-05 22:00:50 +01:00
def get(self) -> Optional[Message]:
"""
Reads one message from the Redis queue
"""
2018-09-20 09:41:19 +02:00
try:
if self.should_stop():
2023-02-05 22:00:50 +01:00
return None
msg = self.redis.blpop(self.redis_queue, timeout=1)
if not msg or msg[1] is None:
2023-02-05 22:00:50 +01:00
return None
2021-06-25 22:47:40 +02:00
msg = msg[1].decode('utf-8')
2022-01-23 14:25:00 +01:00
return Message.build(msg)
2018-09-20 09:41:19 +02:00
except Exception as e:
logger.exception(e)
2023-02-05 22:00:50 +01:00
return None
2018-09-20 09:41:19 +02:00
def post(self, msg):
2023-02-05 22:00:50 +01:00
"""
Sends a message to the Redis queue
"""
from redis import exceptions
2023-02-05 22:00:50 +01:00
try:
return self.redis.rpush(self.redis_queue, str(msg))
except exceptions.ConnectionError as e:
if not self.should_stop():
# Raise the exception only if the bus it not supposed to be
# stopped
raise e
return None
2018-09-20 09:41:19 +02:00
def stop(self):
super().stop()
self.redis.close()
2018-09-20 09:41:19 +02:00
# vim:sw=4:ts=4:et: