oss/Radicale
oss/Radicale
1
0
Fork 0
mirror of https://github.com/Kozea/Radicale.git synced 2025-08-01 18:18:31 +00:00
Code Wiki Activity

Added connection recovery and logging

Browse source
This commit is contained in:
Tuna Celik 2020-08-17 15:09:38 +02:00
parent e38b88f9f7
commit b8af0c7490
1 changed files with 21 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

28
radicale/hook/rabbitmq/__init__.py
View file

@ -1,30 +1,36 @@
import pika import pika
from pika.exceptions import ChannelWrongStateError
from radicale import hook from radicale import hook
from radicale.hook import HookNotificationItem from radicale.hook import HookNotificationItem
from radicale.log import logger
class Hook(hook.BaseHook): class Hook(hook.BaseHook):
def __init__(self, configuration): def __init__(self, configuration):
super().__init__(configuration) super().__init__(configuration)
endpoint = configuration.get("hook", "rabbitmq_endpoint") self._endpoint = configuration.get("hook", "rabbitmq_endpoint")
self._topic = configuration.get("hook", "rabbitmq_topic") self._topic = configuration.get("hook", "rabbitmq_topic")
self._encoding = configuration.get("encoding", "stock") self._encoding = configuration.get("encoding", "stock")
self._make_connection_synced(endpoint) self._make_connection_synced()
self._make_declare_queue_synced(self._topic) self._make_declare_queue_synced()
def _make_connection_synced(self, endpoint): def _make_connection_synced(self):
parameters = pika.URLParameters(endpoint) parameters = pika.URLParameters(self._endpoint)
connection = pika.BlockingConnection(parameters) connection = pika.BlockingConnection(parameters)
self._channel = connection.channel() self._channel = connection.channel()
def _make_declare_queue_synced(self, topic): def _make_declare_queue_synced(self):
self._channel.queue_declare(queue=topic, durable=True) self._channel.queue_declare(queue=self._topic, durable=True)
def notify(self, notification_item): def notify(self, notification_item):
if isinstance(notification_item, HookNotificationItem): if isinstance(notification_item, HookNotificationItem):
self._notify(notification_item, True)
def _notify(self, notification_item, recall):
try:
self._channel.basic_publish( self._channel.basic_publish(
exchange='', exchange='',
routing_key=self._topic, routing_key=self._topic,
@ -32,3 +38,11 @@ class Hook(hook.BaseHook):
encoding=self._encoding encoding=self._encoding
) )
) )
except ChannelWrongStateError as e:
if recall:
self._make_connection_synced()
self._notify(notification_item, False)
return
logger.error("An exception is occurred while "
"publishing hook notification item: %s",
e, exc_info=True)