Commit 5456fe87 authored by Jorrit Schaap's avatar Jorrit Schaap

SW-811: (re)connect in FromBus.receive and BusListener as well

parent 6d1909d4
......@@ -643,7 +643,7 @@ class FromBus(_AbstractBus):
:return: received message, or None if timeout occurred.
"""
if not self.is_connected:
raise MessageBusError("FromBus is not connected to queue %s at broker %s. Cannot receive messages." % (self.queue, self.broker))
self.reconnect()
kombu_msg = None
start = datetime.utcnow()
......@@ -1539,6 +1539,10 @@ class BusListener:
except MessagingError as me:
# just log any own MessagingError, and continue loop.
logger.error(me)
if not receiver.is_connected:
receiver.reconnect()
except Exception as e:
# Unknown problem in the library. Report this and continue.
logger.exception("[%s:] ERROR during processing of incoming message: %s", self.__class__.__name__, e)
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment